Fix ``mull_targets()`` with mamba 2.x

    abstractmethod,

)

from typing import (

    Any,

    Callable,

    Container as TypingContainer,

    Dict,

        self.namespace = namespace

        self.hash_func = hash_func

        self.auto_install = string_as_bool(auto_install)

        self._mulled_kwds = {

        self._mulled_kwds: Dict[str, Any] = {

            "namespace": namespace,

            "hash_func": self.hash_func,

            "command": "build-and-test",

    .using(conda_image)

        .withHostConfig({binds = bind_args})

        .run('/bin/sh', '-c', preinstall

            .. conda_bin .. ' install '

            .. conda_bin .. ' create '

            .. channel_args .. ' '

            .. target_args

            .. ' --strict-channel-priority -p /usr/local --copy --yes '

from sys import platform as _platform

from typing import (

    Any,

    Callable,

    Dict,

    Iterable,

    List,

    NoReturn,

    Optional,

    TYPE_CHECKING,

)

import yaml

from typing_extensions import Literal

from galaxy.tool_util.deps import installable

from galaxy.tool_util.deps.conda_util import (

)

from ..conda_compat import MetaData

if TYPE_CHECKING:

    from galaxy.util.path import StrPath

log = logging.getLogger(__name__)

INVFILE = os.environ.get("INVFILE", os.path.join(os.path.dirname(__file__), "invfile.lua"))

def mull_targets(

    targets: List[CondaTarget],

    involucro_context=None,

    command="build",

    channels=DEFAULT_CHANNELS,

    namespace="biocontainers",

    test="true",

    test_files=None,

    image_build=None,

    name_override=None,

    repository_template=DEFAULT_REPOSITORY_TEMPLATE,

    dry_run=False,

    conda_version=None,

    mamba_version=None,

    use_mamba=False,

    verbose=False,

    binds=DEFAULT_BINDS,

    rebuild=True,

    oauth_token=None,

    hash_func="v2",

    singularity=False,

    singularity_image_dir="singularity_import",

    base_image=None,

    determine_base_image=True,

    invfile=INVFILE,

):

    involucro_context: Optional["InvolucroContext"] = None,

    command: str = "build",

    channels: List[str] = DEFAULT_CHANNELS,

    namespace: str = "biocontainers",

    test: str = "true",

    test_files: Optional[List[str]] = None,

    image_build: Optional[str] = None,

    name_override: Optional[str] = None,

    repository_template: str = DEFAULT_REPOSITORY_TEMPLATE,

    dry_run: bool = False,

    conda_version: Optional[str] = None,

    mamba_version: Optional[str] = None,

    use_mamba: bool = False,

    verbose: bool = False,

    binds: List[str] = DEFAULT_BINDS,

    rebuild: bool = True,

    oauth_token: Optional[str] = None,

    hash_func: Literal["v1", "v2"] = "v2",

    singularity: bool = False,

    singularity_image_dir: "StrPath" = "singularity_import",

    base_image: Optional[str] = None,

    determine_base_image: bool = True,

    invfile: str = INVFILE,

) -> int:

    if involucro_context is None:

        involucro_context = InvolucroContext()

    if test:

        involucro_args.extend(["-set", f"TEST={test}"])

    verbose = "--verbose" if verbose else "--quiet"

    verbose_opt = "--verbose" if verbose else "--quiet"

    specs: List[str] = []

    if conda_version is not None:

        specs.append(f"conda={conda_version}")

    conda_bin = "conda"

    if use_mamba:

        conda_bin = "mamba"

        if mamba_version is None:

            mamba_version = ""

    involucro_args.extend(["-set", f"CONDA_BIN={conda_bin}"])

    if conda_version is not None or mamba_version is not None:

        mamba_test = "true"

        specs = []

        if conda_version is not None:

            specs.append(f"conda={conda_version}")

        if mamba_version is not None:

            if mamba_version == "" and not specs:

                # If nothing but mamba without a specific version is requested,

                # then only run conda install if mamba is not already installed.

                mamba_test = "[ '[]' = \"$( conda list --json --full-name mamba )\" ]"

            specs.append(f"mamba={mamba_version}")

        conda_install = f"""conda install {verbose} --yes {" ".join(f"'{spec}'" for spec in specs)}"""

        involucro_args.extend(["-set", f"PREINSTALL=if {mamba_test} ; then {conda_install} ; fi"])

        else:

            # For https://github.com/mamba-org/mamba/pull/3919

            specs.append("mamba>=2.2.0")

    involucro_args.extend(["-set", f"CONDA_BIN={conda_bin}"])

    if specs:

        conda_install = f"""conda install {verbose_opt} --yes {" ".join(f"'{spec}'" for spec in specs)}"""

        involucro_args.extend(["-set", f"PREINSTALL={conda_install}"])

    involucro_args.append(command)

    if test_files:

class InvolucroContext(installable.InstallableContext):

    installable_description = "Involucro"

    def __init__(self, involucro_bin=None, shell_exec=None, verbose="3"):

    def __init__(

        self,

        involucro_bin: Optional[str] = None,

        shell_exec: Optional[Callable[[List[str]], int]] = None,

        verbose: str = "3",

    ) -> None:

        if involucro_bin is None:

            if os.path.exists("./involucro"):

                self.involucro_bin = "./involucro"

        self.shell_exec = shell_exec or commands.shell

        self.verbose = verbose

    def build_command(self, involucro_args):

    def build_command(self, involucro_args: List[str]) -> List[str]:

        return [self.involucro_bin, f"-v={self.verbose}"] + involucro_args

    def exec_command(self, involucro_args):

    def exec_command(self, involucro_args: List[str]) -> int:

        cmd = self.build_command(involucro_args)

        # Create ./build dir manually, otherwise Docker will do it as root

        created_build_dir = False

    return kwds

def main(argv=None):

def main(argv=None) -> NoReturn:

    """Main entry-point for the CLI tool."""

    parser = arg_parser(argv, globals())

    add_build_arguments(parser)

    return sconda - squay  # sconda.symmetric_difference(squay)

def run_channel(args, build_last_n_versions=1):

def run_channel(args, build_last_n_versions: int = 1) -> None:

    """Build list of involucro commands (as shell snippet) to run."""

    session = requests.session()

    for pkg_name, pkg_tests in get_affected_packages(args):

"""

import collections

import glob

import os

import sys

from dataclasses import dataclass

from typing import (

    Any,

    Iterator,

    List,

    Optional,

    Sequence,

)

from galaxy.tool_util.deps.conda_util import CondaTarget

from ._cli import arg_parser

from .mulled_build import (

    add_build_arguments,

)

KNOWN_FIELDS = ["targets", "image_build", "name_override", "base_image"]

FALLBACK_LINE_TUPLE = collections.namedtuple("FALLBACK_LINE_TUPLE", "targets image_build name_override base_image")

FALLBACK_FIELD_ORDER = ("targets", "image_build", "name_override", "base_image")

@dataclass

class Target:

    targets: List[CondaTarget]

    image_build: Optional[str]

    name_override: Optional[str]

    base_image: Optional[str]

def main(argv=None):

            sys.exit(ret)

def generate_targets(target_source):

def generate_targets(target_source) -> Iterator[Target]:

    """Generate all targets from TSV files in specified file or directory."""

    target_source = os.path.abspath(target_source)

    if os.path.isdir(target_source):

    for target_source_file in target_source_files:

        # If no headers are defined we use the 4 default fields in the order

        # that has been used in galaxy-tool-util / galaxy-lib < 20.01

        line_tuple = FALLBACK_LINE_TUPLE

        field_order: Sequence[str] = FALLBACK_FIELD_ORDER

        with open(target_source_file) as f:

            for line in f.readlines():

                if line:

                    line = line.strip()

                    if line.startswith("#"):

                        # headers can define a different column order

                        line_tuple = tuple_from_header(line)

                        field_order = field_order_from_header(line)

                    else:

                        yield line_to_targets(line, line_tuple)

                        yield line_to_targets(line, field_order)

def tuple_from_header(header):

def field_order_from_header(header: str) -> List[str]:

    fields = header[1:].split("\t")

    for field in fields:

        assert field in KNOWN_FIELDS, f"'{field}' is not one of {KNOWN_FIELDS}"

    for field in KNOWN_FIELDS:

        if field not in fields:

            fields.append(field)

    return collections.namedtuple("_Line", f"{' '.join(fields)}")

    return fields

def line_to_targets(line_str, line_tuple):

def line_to_targets(line_str: str, field_order: Sequence[str]) -> Target:

    """Parse a line so that some columns can remain unspecified."""

    line_parts = line_str.split("\t")

    n_fields = len(line_tuple._fields)

    targets_column = line_tuple._fields.index("targets")

    line_parts: List[Any] = line_str.split("\t")

    n_fields = len(field_order)

    targets_column = field_order.index("targets")

    assert (

        len(line_parts) <= n_fields

    ), f"Too many fields in line [{line_str}], expect at most {n_fields} - targets, image build number, and name override."

    line_parts += [None] * (n_fields - len(line_parts))

    line_parts[targets_column] = target_str_to_targets(line_parts[targets_column])

    return line_tuple(*line_parts)

    return Target(**dict(zip(field_order, line_parts)))

__all__ = ("main",)