path: root/src/debputy/commands/debputy_cmd/context.py

import argparse
import dataclasses
import errno
import os
from typing import (
    Optional,
    Tuple,
    Mapping,
    FrozenSet,
    Set,
    Union,
    Sequence,
    Iterable,
    Callable,
    Dict,
    TYPE_CHECKING,
)

from debian.debian_support import DpkgArchTable

from debputy._deb_options_profiles import DebBuildOptionsAndProfiles
from debputy.architecture_support import (
    DpkgArchitectureBuildProcessValuesTable,
    dpkg_architecture_table,
)
from debputy.exceptions import DebputyRuntimeError
from debputy.filesystem_scan import FSROOverlay
from debputy.highlevel_manifest import HighLevelManifest
from debputy.highlevel_manifest_parser import YAMLManifestParser
from debputy.packages import SourcePackage, BinaryPackage, parse_source_debian_control
from debputy.plugin.api import VirtualPath
from debputy.plugin.api.impl import load_plugin_features
from debputy.plugin.api.feature_set import PluginProvidedFeatureSet
from debputy.substitution import (
    Substitution,
    VariableContext,
    SubstitutionImpl,
    NULL_SUBSTITUTION,
)
from debputy.util import _error, PKGNAME_REGEX, resolve_source_date_epoch, setup_logging

if TYPE_CHECKING:
    from argparse import _SubParsersAction


CommandHandler = Callable[["CommandContext"], None]
ArgparserConfigurator = Callable[[argparse.ArgumentParser], None]


def add_arg(
    *name_or_flags: str,
    **kwargs,
) -> Callable[[argparse.ArgumentParser], None]:
    def _configurator(argparser: argparse.ArgumentParser) -> None:
        argparser.add_argument(
            *name_or_flags,
            **kwargs,
        )

    return _configurator


@dataclasses.dataclass(slots=True, frozen=True)
class CommandArg:
    parsed_args: argparse.Namespace
    plugin_search_dirs: Sequence[str]


@dataclasses.dataclass
class Command:
    handler: Callable[["CommandContext"], None]
    require_substitution: bool = True
    requested_plugins_only: bool = False


class CommandContext:
    def __init__(
        self,
        parsed_args: argparse.Namespace,
        plugin_search_dirs: Sequence[str],
        require_substitution: bool = True,
        requested_plugins_only: bool = False,
    ) -> None:
        self.parsed_args = parsed_args
        self.plugin_search_dirs = plugin_search_dirs
        self._require_substitution = require_substitution
        self._requested_plugins_only = requested_plugins_only
        self._debputy_plugin_feature_set: PluginProvidedFeatureSet = (
            PluginProvidedFeatureSet()
        )
        self._debian_dir = FSROOverlay.create_root_dir("debian", "debian")
        self._mtime: Optional[int] = None
        self._source_variables: Optional[Mapping[str, str]] = None
        self._substitution: Optional[Substitution] = None
        self._requested_plugins: Optional[Sequence[str]] = None
        self._plugins_loaded = False
        self._dctrl_data: Optional[
            Tuple[
                DpkgArchitectureBuildProcessValuesTable,
                DpkgArchTable,
                DebBuildOptionsAndProfiles,
                "SourcePackage",
                Mapping[str, "BinaryPackage"],
            ]
        ] = None

    @property
    def debian_dir(self) -> VirtualPath:
        return self._debian_dir

    @property
    def mtime(self) -> int:
        if self._mtime is None:
            self._mtime = resolve_source_date_epoch(
                None,
                substitution=self.substitution,
            )
        return self._mtime

    def source_package(self) -> SourcePackage:
        _a, _b, _c, source, _d = self._parse_dctrl()
        return source

    def binary_packages(self) -> Mapping[str, "BinaryPackage"]:
        _a, _b, _c, _source, binary_package_table = self._parse_dctrl()
        return binary_package_table

    def requested_plugins(self) -> Sequence[str]:
        if self._requested_plugins is None:
            self._requested_plugins = self._resolve_requested_plugins()
        return self._requested_plugins

    def required_plugins(self) -> Set[str]:
        return set(getattr(self.parsed_args, "required_plugins") or [])

    @property
    def deb_build_options_and_profiles(self) -> "DebBuildOptionsAndProfiles":
        _a, _b, deb_build_options_and_profiles, _c, _d = self._parse_dctrl()
        return deb_build_options_and_profiles

    @property
    def deb_build_options(self) -> Mapping[str, Optional[str]]:
        return self.deb_build_options_and_profiles.deb_build_options

    def _create_substitution(
        self,
        parsed_args: argparse.Namespace,
        plugin_feature_set: PluginProvidedFeatureSet,
        debian_dir: VirtualPath,
    ) -> Substitution:
        requested_subst = self._require_substitution
        if hasattr(parsed_args, "substitution"):
            requested_subst = parsed_args.substitution
        if requested_subst is False and self._require_substitution:
            _error(f"--no-substitution cannot be used with {parsed_args.command}")
        if self._require_substitution or requested_subst is not False:
            variable_context = VariableContext(debian_dir)
            return SubstitutionImpl(
                plugin_feature_set=plugin_feature_set,
                unresolvable_substitutions=frozenset(["PACKAGE"]),
                variable_context=variable_context,
            )
        return NULL_SUBSTITUTION

    def load_plugins(self) -> PluginProvidedFeatureSet:
        if not self._plugins_loaded:
            requested_plugins = None
            required_plugins = self.required_plugins()
            if self._requested_plugins_only:
                requested_plugins = self.requested_plugins()
            debug_mode = getattr(self.parsed_args, "debug_mode", False)
            load_plugin_features(
                self.plugin_search_dirs,
                self.substitution,
                requested_plugins_only=requested_plugins,
                required_plugins=required_plugins,
                plugin_feature_set=self._debputy_plugin_feature_set,
                debug_mode=debug_mode,
            )
            self._plugins_loaded = True
        return self._debputy_plugin_feature_set

    @staticmethod
    def _plugin_from_dependency_field(dep_field: str) -> Iterable[str]:
        package_prefix = "debputy-plugin-"
        for dep_clause in (d.strip() for d in dep_field.split(",")):
            dep = dep_clause.split("|")[0].strip()
            if not dep.startswith(package_prefix):
                continue
            m = PKGNAME_REGEX.search(dep)
            assert m
            package_name = m.group(0)
            plugin_name = package_name[len(package_prefix) :]
            yield plugin_name

    def _resolve_requested_plugins(self) -> Sequence[str]:
        _a, _b, _c, source_package, _d = self._parse_dctrl()
        bd = source_package.fields.get("Build-Depends", "")
        plugins = list(self._plugin_from_dependency_field(bd))
        for field_name in ("Build-Depends-Arch", "Build-Depends-Indep"):
            f = source_package.fields.get(field_name)
            if not f:
                continue
            for plugin in self._plugin_from_dependency_field(f):
                raise DebputyRuntimeError(
                    f"Cannot load plugins via {field_name}:"
                    f" Please move debputy-plugin-{plugin} dependency to Build-Depends."
                )

        return plugins

    @property
    def substitution(self) -> Substitution:
        if self._substitution is None:
            self._substitution = self._create_substitution(
                self.parsed_args,
                self._debputy_plugin_feature_set,
                self.debian_dir,
            )
        return self._substitution

    def _parse_dctrl(
        self,
    ) -> Tuple[
        DpkgArchitectureBuildProcessValuesTable,
        DpkgArchTable,
        DebBuildOptionsAndProfiles,
        "SourcePackage",
        Mapping[str, "BinaryPackage"],
    ]:
        if self._dctrl_data is None:
            build_env = DebBuildOptionsAndProfiles.instance()
            dpkg_architecture_variables = dpkg_architecture_table()
            dpkg_arch_query_table = DpkgArchTable.load_arch_table()

            packages: Union[Set[str], FrozenSet[str]] = frozenset()
            if hasattr(self.parsed_args, "packages"):
                packages = self.parsed_args.packages

            try:
                debian_control = self.debian_dir.get("control")
                if debian_control is None:
                    raise FileNotFoundError(
                        errno.ENOENT,
                        os.strerror(errno.ENOENT),
                        os.path.join(self.debian_dir.fs_path, "control"),
                    )
                source_package, binary_packages = parse_source_debian_control(
                    debian_control,
                    packages,  # -p/--package
                    set(),  # -N/--no-package
                    False,  # -i
                    False,  # -a
                    dpkg_architecture_variables=dpkg_architecture_variables,
                    dpkg_arch_query_table=dpkg_arch_query_table,
                    build_env=build_env,
                )
                assert packages <= binary_packages.keys()
            except FileNotFoundError:
                _error(
                    "This subcommand must be run from a source package root; expecting debian/control to exist."
                )

            self._dctrl_data = (
                dpkg_architecture_variables,
                dpkg_arch_query_table,
                build_env,
                source_package,
                binary_packages,
            )

        return self._dctrl_data

    @property
    def has_dctrl_file(self) -> bool:
        debian_control = self.debian_dir.get("control")
        return debian_control is not None

    def manifest_parser(
        self,
        *,
        manifest_path: Optional[str] = None,
    ) -> YAMLManifestParser:
        substitution = self.substitution

        (
            dpkg_architecture_variables,
            dpkg_arch_query_table,
            build_env,
            source_package,
            binary_packages,
        ) = self._parse_dctrl()

        if self.parsed_args.debputy_manifest is not None:
            manifest_path = self.parsed_args.debputy_manifest
        if manifest_path is None:
            manifest_path = os.path.join(self.debian_dir.fs_path, "debputy.manifest")
        return YAMLManifestParser(
            manifest_path,
            source_package,
            binary_packages,
            substitution,
            dpkg_architecture_variables,
            dpkg_arch_query_table,
            build_env,
            self.load_plugins(),
            debian_dir=self.debian_dir,
        )

    def parse_manifest(
        self,
        *,
        manifest_path: Optional[str] = None,
    ) -> HighLevelManifest:
        substitution = self.substitution
        manifest_required = False

        (
            dpkg_architecture_variables,
            dpkg_arch_query_table,
            build_env,
            _,
            binary_packages,
        ) = self._parse_dctrl()

        if self.parsed_args.debputy_manifest is not None:
            manifest_path = self.parsed_args.debputy_manifest
            manifest_required = True
        if manifest_path is None:
            manifest_path = os.path.join(self.debian_dir.fs_path, "debputy.manifest")
        parser = self.manifest_parser(manifest_path=manifest_path)

        os.environ["SOURCE_DATE_EPOCH"] = substitution.substitute(
            "{{SOURCE_DATE_EPOCH}}",
            "Internal resolution",
        )
        if os.path.isfile(manifest_path):
            return parser.parse_manifest()
        if manifest_required:
            _error(f'The path "{manifest_path}" is not a file!')
        return parser.build_manifest()


class CommandBase:
    __slots__ = ()

    def configure(self, argparser: argparse.ArgumentParser) -> None:
        # Does nothing by default
        pass

    def __call__(self, command_arg: CommandArg) -> None:
        raise NotImplementedError


class SubcommandBase(CommandBase):
    __slots__ = ("name", "aliases", "help_description")

    def __init__(
        self,
        name: str,
        *,
        aliases: Sequence[str] = tuple(),
        help_description: Optional[str] = None,
    ) -> None:
        self.name = name
        self.aliases = aliases
        self.help_description = help_description

    def add_subcommand_to_subparser(
        self,
        subparser: "_SubParsersAction",
    ) -> argparse.ArgumentParser:
        parser = subparser.add_parser(
            self.name,
            aliases=self.aliases,
            help=self.help_description,
            allow_abbrev=False,
        )
        self.configure(parser)
        return parser


class GenericSubCommand(SubcommandBase):
    __slots__ = (
        "_handler",
        "_configure_handler",
        "_require_substitution",
        "_requested_plugins_only",
        "_log_only_to_stderr",
    )

    def __init__(
        self,
        name: str,
        handler: Callable[[CommandContext], None],
        *,
        aliases: Sequence[str] = tuple(),
        help_description: Optional[str] = None,
        configure_handler: Optional[Callable[[argparse.ArgumentParser], None]] = None,
        require_substitution: bool = True,
        requested_plugins_only: bool = False,
        log_only_to_stderr: bool = False,
    ) -> None:
        super().__init__(name, aliases=aliases, help_description=help_description)
        self._handler = handler
        self._configure_handler = configure_handler
        self._require_substitution = require_substitution
        self._requested_plugins_only = requested_plugins_only
        self._log_only_to_stderr = log_only_to_stderr

    def configure_handler(
        self,
        handler: Callable[[argparse.ArgumentParser], None],
    ) -> None:
        if self._configure_handler is not None:
            raise TypeError("Only one argument handler can be provided")
        self._configure_handler = handler

    def configure(self, argparser: argparse.ArgumentParser) -> None:
        handler = self._configure_handler
        if handler is not None:
            handler(argparser)

    def __call__(self, command_arg: CommandArg) -> None:
        context = CommandContext(
            command_arg.parsed_args,
            command_arg.plugin_search_dirs,
            self._require_substitution,
            self._requested_plugins_only,
        )
        if self._log_only_to_stderr:
            setup_logging(reconfigure_logging=True, log_only_to_stderr=True)
        return self._handler(context)


class DispatchingCommandMixin(CommandBase):
    __slots__ = ()

    def add_subcommand(self, subcommand: SubcommandBase) -> None:
        raise NotImplementedError

    def add_dispatching_subcommand(
        self,
        name: str,
        dest: str,
        *,
        aliases: Sequence[str] = tuple(),
        help_description: Optional[str] = None,
        metavar: str = "command",
        default_subcommand: Optional[str] = None,
    ) -> "DispatcherCommand":
        ds = DispatcherCommand(
            name,
            dest,
            aliases=aliases,
            help_description=help_description,
            metavar=metavar,
            default_subcommand=default_subcommand,
        )
        self.add_subcommand(ds)
        return ds

    def register_subcommand(
        self,
        name: Union[str, Sequence[str]],
        *,
        help_description: Optional[str] = None,
        argparser: Optional[
            Union[ArgparserConfigurator, Sequence[ArgparserConfigurator]]
        ] = None,
        require_substitution: bool = True,
        requested_plugins_only: bool = False,
        log_only_to_stderr: bool = False,
    ) -> Callable[[CommandHandler], GenericSubCommand]:
        if isinstance(name, str):
            cmd_name = name
            aliases = []
        else:
            cmd_name = name[0]
            aliases = name[1:]

        if argparser is not None and not callable(argparser):
            args = argparser

            def _wrapper(parser: argparse.ArgumentParser) -> None:
                for configurator in args:
                    configurator(parser)

            argparser = _wrapper

        def _annotation_impl(func: CommandHandler) -> GenericSubCommand:
            subcommand = GenericSubCommand(
                cmd_name,
                func,
                aliases=aliases,
                help_description=help_description,
                require_substitution=require_substitution,
                requested_plugins_only=requested_plugins_only,
                log_only_to_stderr=log_only_to_stderr,
            )
            self.add_subcommand(subcommand)
            if argparser is not None:
                subcommand.configure_handler(argparser)

            return subcommand

        return _annotation_impl


class DispatcherCommand(SubcommandBase, DispatchingCommandMixin):
    __slots__ = (
        "_subcommands",
        "_aliases",
        "_dest",
        "_metavar",
        "_required",
        "_default_subcommand",
        "_argparser",
    )

    def __init__(
        self,
        name: str,
        dest: str,
        *,
        aliases: Sequence[str] = tuple(),
        help_description: Optional[str] = None,
        metavar: str = "command",
        default_subcommand: Optional[str] = None,
    ) -> None:
        super().__init__(name, aliases=aliases, help_description=help_description)
        self._aliases: Dict[str, SubcommandBase] = {}
        self._subcommands: Dict[str, SubcommandBase] = {}
        self._dest = dest
        self._metavar = metavar
        self._default_subcommand = default_subcommand
        self._argparser: Optional[argparse.ArgumentParser] = None

    def add_subcommand(self, subcommand: SubcommandBase) -> None:
        all_names = [subcommand.name]
        if subcommand.aliases:
            all_names.extend(subcommand.aliases)
        aliases = self._aliases
        for n in all_names:
            if n in aliases:
                raise ValueError(
                    f"Internal error: Multiple handlers for {n} on topic {self.name}"
                )

            aliases[n] = subcommand
        self._subcommands[subcommand.name] = subcommand

    def configure(self, argparser: argparse.ArgumentParser) -> None:
        if self._argparser is not None:
            raise TypeError("Cannot configure twice!")
        self._argparser = argparser
        subcommands = self._subcommands
        if not subcommands:
            raise ValueError(
                f"Internal error: No subcommands for subcommand {self.name} (then why do we have it?)"
            )
        default_subcommand = self._default_subcommand
        required = default_subcommand is None
        if (
            default_subcommand is not None
            and default_subcommand not in ("--help", "-h")
            and default_subcommand not in subcommands
        ):
            raise ValueError(
                f"Internal error: Subcommand {self.name} should have {default_subcommand} as default,"
                " but it was not registered?"
            )
        subparser = argparser.add_subparsers(
            dest=self._dest,
            required=required,
            metavar=self._metavar,
        )
        for subcommand in subcommands.values():
            subcommand.add_subcommand_to_subparser(subparser)

    def has_command(self, command: str) -> bool:
        return command in self._aliases

    def __call__(self, command_arg: CommandArg) -> None:
        argparser = self._argparser
        assert argparser is not None
        v = getattr(command_arg.parsed_args, self._dest, None)
        if v is None:
            v = self._default_subcommand
            if v in ("--help", "-h"):
                argparser.parse_args([v])
                _error("Missing command", prog=argparser.prog)

        assert (
            v is not None
        ), f"Internal error: No default subcommand and argparse did not provide the required subcommand {self._dest}?"
        assert (
            v in self._aliases
        ), f"Internal error: {v} was accepted as a topic, but it was not registered?"
        self._aliases[v](command_arg)


ROOT_COMMAND = DispatcherCommand(
    "root",
    dest="command",
    metavar="COMMAND",
)