path: root/src/debputy/plugin/api/impl_types.py

import dataclasses
import os.path
import textwrap
from typing import (
    Optional,
    Callable,
    FrozenSet,
    Dict,
    List,
    Tuple,
    Generic,
    TYPE_CHECKING,
    TypeVar,
    cast,
    Any,
    Sequence,
    Union,
    Type,
    TypedDict,
    Iterable,
    Mapping,
    NotRequired,
    Literal,
    Set,
    Iterator,
)
from weakref import ref

from debputy import DEBPUTY_DOC_ROOT_DIR
from debputy.exceptions import (
    DebputyFSIsROError,
    PluginAPIViolationError,
    PluginConflictError,
    UnhandledOrUnexpectedErrorFromPluginError,
)
from debputy.filesystem_scan import as_path_def
from debputy.installations import InstallRule
from debputy.maintscript_snippet import DpkgMaintscriptHelperCommand
from debputy.manifest_conditions import ManifestCondition
from debputy.manifest_parser.base_types import DebputyParsedContent, TypeMapping
from debputy.manifest_parser.exceptions import ManifestParseException
from debputy.manifest_parser.util import AttributePath
from debputy.packages import BinaryPackage
from debputy.plugin.api import (
    VirtualPath,
    BinaryCtrlAccessor,
    PackageProcessingContext,
)
from debputy.plugin.api.spec import (
    DebputyPluginInitializer,
    MetadataAutoDetector,
    DpkgTriggerType,
    ParserDocumentation,
    PackageProcessor,
    PathDef,
    ParserAttributeDocumentation,
    undocumented_attr,
    documented_attr,
    reference_documentation,
    PackagerProvidedFileReferenceDocumentation,
    TypeMappingDocumentation,
)
from debputy.substitution import VariableContext
from debputy.transformation_rules import TransformationRule
from debputy.util import _normalize_path, package_cross_check_precheck

if TYPE_CHECKING:
    from debputy.plugin.api.spec import (
        ServiceDetector,
        ServiceIntegrator,
        PackageTypeSelector,
    )
    from debputy.manifest_parser.parser_data import ParserContextData
    from debputy.highlevel_manifest import (
        HighLevelManifest,
        PackageTransformationDefinition,
        BinaryPackageData,
    )


_PACKAGE_TYPE_DEB_ONLY = frozenset(["deb"])
_ALL_PACKAGE_TYPES = frozenset(["deb", "udeb"])


TD = TypeVar("TD", bound="DebputyParsedContent")
PF = TypeVar("PF")
SF = TypeVar("SF")
TP = TypeVar("TP")
TTP = Type[TP]

DIPKWHandler = Callable[[str, AttributePath, "ParserContextData"], TP]
DIPHandler = Callable[[str, PF, AttributePath, "ParserContextData"], TP]


def resolve_package_type_selectors(
    package_type: "PackageTypeSelector",
) -> FrozenSet[str]:
    if package_type is _ALL_PACKAGE_TYPES or package_type is _PACKAGE_TYPE_DEB_ONLY:
        return cast("FrozenSet[str]", package_type)
    if isinstance(package_type, str):
        return (
            _PACKAGE_TYPE_DEB_ONLY
            if package_type == "deb"
            else frozenset([package_type])
        )
    else:
        return frozenset(package_type)


@dataclasses.dataclass(slots=True)
class DebputyPluginMetadata:
    plugin_name: str
    api_compat_version: int
    plugin_loader: Optional[Callable[[], Callable[["DebputyPluginInitializer"], None]]]
    plugin_initializer: Optional[Callable[["DebputyPluginInitializer"], None]]
    plugin_path: str
    _is_initialized: bool = False

    @property
    def is_loaded(self) -> bool:
        return self.plugin_initializer is not None

    @property
    def is_initialized(self) -> bool:
        return self._is_initialized

    def initialize_plugin(self, api: "DebputyPluginInitializer") -> None:
        if self.is_initialized:
            raise RuntimeError("Cannot load plugins twice")
        if not self.is_loaded:
            self.load_plugin()
        plugin_initializer = self.plugin_initializer
        assert plugin_initializer is not None
        plugin_initializer(api)
        self._is_initialized = True

    def load_plugin(self) -> None:
        plugin_loader = self.plugin_loader
        assert plugin_loader is not None
        self.plugin_initializer = plugin_loader()
        assert self.plugin_initializer is not None


@dataclasses.dataclass(slots=True, frozen=True)
class PluginProvidedParser(Generic[PF, TP]):
    parser: "DeclarativeInputParser[PF]"
    handler: Callable[[str, PF, "AttributePath", "ParserContextData"], TP]
    plugin_metadata: DebputyPluginMetadata

    def parse(
        self,
        name: str,
        value: object,
        attribute_path: "AttributePath",
        *,
        parser_context: "ParserContextData",
    ) -> TP:
        parsed_value = self.parser.parse_input(
            value, attribute_path, parser_context=parser_context
        )
        return self.handler(name, parsed_value, attribute_path, parser_context)


class PPFFormatParam(TypedDict):
    priority: Optional[int]
    name: str
    owning_package: str


@dataclasses.dataclass(slots=True, frozen=True)
class PackagerProvidedFileClassSpec:
    debputy_plugin_metadata: DebputyPluginMetadata
    stem: str
    installed_as_format: str
    default_mode: int
    default_priority: Optional[int]
    allow_name_segment: bool
    allow_architecture_segment: bool
    post_formatting_rewrite: Optional[Callable[[str], str]]
    packageless_is_fallback_for_all_packages: bool
    reservation_only: bool
    formatting_callback: Optional[Callable[[str, PPFFormatParam, VirtualPath], str]] = (
        None
    )
    reference_documentation: Optional[PackagerProvidedFileReferenceDocumentation] = None
    bug_950723: bool = False

    @property
    def supports_priority(self) -> bool:
        return self.default_priority is not None

    def compute_dest(
        self,
        assigned_name: str,
        # Note this method is currently used 1:1 inside plugin tests.
        *,
        owning_package: Optional[str] = None,
        assigned_priority: Optional[int] = None,
        path: Optional[VirtualPath] = None,
    ) -> Tuple[str, str]:
        if assigned_priority is not None and not self.supports_priority:
            raise ValueError(
                f"Cannot assign priority to packager provided files with stem"
                f' "{self.stem}" (e.g., "debian/foo.{self.stem}"). They'
                " do not use priority at all."
            )

        path_format = self.installed_as_format
        if self.supports_priority and assigned_priority is None:
            assigned_priority = self.default_priority

        if owning_package is None:
            owning_package = assigned_name

        params: PPFFormatParam = {
            "priority": assigned_priority,
            "name": assigned_name,
            "owning_package": owning_package,
        }

        if self.formatting_callback is not None:
            if path is None:
                raise ValueError(
                    "The path parameter is required for PPFs with formatting_callback"
                )
            dest_path = self.formatting_callback(path_format, params, path)
        else:
            dest_path = path_format.format(**params)

        dirname, basename = os.path.split(dest_path)
        dirname = _normalize_path(dirname)

        if self.post_formatting_rewrite:
            basename = self.post_formatting_rewrite(basename)
        return dirname, basename


@dataclasses.dataclass(slots=True)
class MetadataOrMaintscriptDetector:
    plugin_metadata: DebputyPluginMetadata
    detector_id: str
    detector: MetadataAutoDetector
    applies_to_package_types: FrozenSet[str]
    enabled: bool = True

    def applies_to(self, binary_package: BinaryPackage) -> bool:
        return binary_package.package_type in self.applies_to_package_types

    def run_detector(
        self,
        fs_root: "VirtualPath",
        ctrl: "BinaryCtrlAccessor",
        context: "PackageProcessingContext",
    ) -> None:
        try:
            self.detector(fs_root, ctrl, context)
        except DebputyFSIsROError as e:
            nv = self.plugin_metadata.plugin_name
            raise PluginAPIViolationError(
                f'The plugin {nv} violated the API contract for "metadata detectors"'
                " by attempting to mutate the provided file system in its metadata detector"
                f" with id {self.detector_id}.  File system mutation is *not* supported at"
                " this stage (file system layout is committed and the attempted changes"
                " would be lost)."
            ) from e
        except (ChildProcessError, RuntimeError, AttributeError) as e:
            nv = f"{self.plugin_metadata.plugin_name}"
            raise UnhandledOrUnexpectedErrorFromPluginError(
                f"The plugin {nv} threw an unhandled or unexpected exception from its metadata"
                f" detector with id {self.detector_id}."
            ) from e


class DeclarativeInputParser(Generic[TD]):
    @property
    def inline_reference_documentation(self) -> Optional[ParserDocumentation]:
        return None

    @property
    def reference_documentation_url(self) -> Optional[str]:
        doc = self.inline_reference_documentation
        return doc.documentation_reference_url if doc is not None else None

    def parse_input(
        self,
        value: object,
        path: "AttributePath",
        *,
        parser_context: Optional["ParserContextData"] = None,
    ) -> TD:
        raise NotImplementedError


class DispatchingParserBase(Generic[TP]):
    def __init__(self, manifest_attribute_path_template: str) -> None:
        self.manifest_attribute_path_template = manifest_attribute_path_template
        self._parsers: Dict[str, PluginProvidedParser[Any, TP]] = {}

    def is_known_keyword(self, keyword: str) -> bool:
        return keyword in self._parsers

    def registered_keywords(self) -> Iterable[str]:
        yield from self._parsers

    def parser_for(self, keyword: str) -> PluginProvidedParser[Any, TP]:
        return self._parsers[keyword]

    def register_keyword(
        self,
        keyword: Union[str, Sequence[str]],
        handler: DIPKWHandler,
        plugin_metadata: DebputyPluginMetadata,
        *,
        inline_reference_documentation: Optional[ParserDocumentation] = None,
    ) -> None:
        reference_documentation_url = None
        if inline_reference_documentation:
            if inline_reference_documentation.attribute_doc:
                raise ValueError(
                    "Cannot provide per-attribute documentation for a value-less keyword!"
                )
            if inline_reference_documentation.alt_parser_description:
                raise ValueError(
                    "Cannot provide non-mapping-format documentation for a value-less keyword!"
                )
            reference_documentation_url = (
                inline_reference_documentation.documentation_reference_url
            )
        parser = DeclarativeValuelessKeywordInputParser(
            inline_reference_documentation,
            documentation_reference=reference_documentation_url,
        )

        def _combined_handler(
            name: str,
            _ignored: Any,
            attr_path: AttributePath,
            context: "ParserContextData",
        ) -> TP:
            return handler(name, attr_path, context)

        p = PluginProvidedParser(
            parser,
            _combined_handler,
            plugin_metadata,
        )

        self._add_parser(keyword, p)

    def register_parser(
        self,
        keyword: Union[str, List[str]],
        parser: "DeclarativeInputParser[PF]",
        handler: Callable[[str, PF, "AttributePath", "ParserContextData"], TP],
        plugin_metadata: DebputyPluginMetadata,
    ) -> None:
        p = PluginProvidedParser(
            parser,
            handler,
            plugin_metadata,
        )
        self._add_parser(keyword, p)

    def _add_parser(
        self,
        keyword: Union[str, List[str]],
        ppp: "PluginProvidedParser[PF, TP]",
    ) -> None:
        ks = [keyword] if isinstance(keyword, str) else keyword
        for k in ks:
            existing_parser = self._parsers.get(k)
            if existing_parser is not None:
                message = (
                    f'The rule name "{k}" is already taken by the plugin'
                    f" {existing_parser.plugin_metadata.plugin_name}. This conflict was triggered"
                    f" when plugin {ppp.plugin_metadata.plugin_name} attempted to register its parser."
                )
                raise PluginConflictError(
                    message,
                    existing_parser.plugin_metadata,
                    ppp.plugin_metadata,
                )
            self._new_parser(k, ppp)

    def _new_parser(self, keyword: str, ppp: "PluginProvidedParser[PF, TP]") -> None:
        self._parsers[keyword] = ppp

    def parse(
        self,
        orig_value: object,
        attribute_path: "AttributePath",
        *,
        parser_context: "ParserContextData",
    ) -> TP:
        raise NotImplementedError


class DispatchingObjectParser(
    DispatchingParserBase[Mapping[str, Any]],
    DeclarativeInputParser[Mapping[str, Any]],
):
    def __init__(
        self,
        manifest_attribute_path_template: str,
        *,
        parser_documentation: Optional[ParserDocumentation] = None,
    ) -> None:
        super().__init__(manifest_attribute_path_template)
        self._attribute_documentation: List[ParserAttributeDocumentation] = []
        if parser_documentation is None:
            parser_documentation = reference_documentation()
        self._parser_documentation = parser_documentation

    @property
    def reference_documentation_url(self) -> Optional[str]:
        return self._parser_documentation.documentation_reference_url

    @property
    def inline_reference_documentation(self) -> Optional[ParserDocumentation]:
        ref_doc = self._parser_documentation
        return reference_documentation(
            title=ref_doc.title,
            description=ref_doc.description,
            attributes=self._attribute_documentation,
            reference_documentation_url=self.reference_documentation_url,
        )

    def _new_parser(self, keyword: str, ppp: "PluginProvidedParser[PF, TP]") -> None:
        super()._new_parser(keyword, ppp)
        doc = ppp.parser.inline_reference_documentation
        if doc is None or doc.description is None:
            self._attribute_documentation.append(undocumented_attr(keyword))
        else:
            self._attribute_documentation.append(
                documented_attr(keyword, doc.description)
            )

    def register_child_parser(
        self,
        keyword: str,
        parser: "DispatchingObjectParser",
        plugin_metadata: DebputyPluginMetadata,
        *,
        on_end_parse_step: Optional[
            Callable[
                [str, Optional[Mapping[str, Any]], AttributePath, "ParserContextData"],
                None,
            ]
        ] = None,
    ) -> None:
        def _handler(
            name: str,
            value: Mapping[str, Any],
            path: AttributePath,
            parser_context: "ParserContextData",
        ) -> Mapping[str, Any]:
            on_end_parse_step(name, value, path, parser_context)
            return value

        p = PluginProvidedParser(
            parser,
            _handler,
            plugin_metadata,
        )
        self._add_parser(keyword, p)

    # FIXME: Agree on naming (parse vs. parse_input)
    def parse_input(
        self,
        value: object,
        path: "AttributePath",
        *,
        parser_context: Optional["ParserContextData"] = None,
    ) -> TD:
        return self.parse(value, path, parser_context=parser_context)

    def parse(
        self,
        orig_value: object,
        attribute_path: "AttributePath",
        *,
        parser_context: "ParserContextData",
    ) -> TP:
        doc_ref = ""
        if self.reference_documentation_url is not None:
            doc_ref = (
                f" Please see {self.reference_documentation_url} for the documentation."
            )
        if not isinstance(orig_value, dict):
            raise ManifestParseException(
                f"The attribute {attribute_path.path} must be a non-empty mapping.{doc_ref}"
            )
        if not orig_value:
            raise ManifestParseException(
                f"The attribute {attribute_path.path} must be a non-empty mapping.{doc_ref}"
            )
        result = {}
        unknown_keys = orig_value.keys() - self._parsers.keys()
        if unknown_keys:
            first_key = next(iter(unknown_keys))
            remaining_valid_attributes = self._parsers.keys() - orig_value.keys()
            if not remaining_valid_attributes:
                raise ManifestParseException(
                    f'The attribute "{first_key}" is not applicable at {attribute_path.path} (with the'
                    f" current set of plugins).{doc_ref}"
                )
            remaining_valid_attribute_names = ", ".join(remaining_valid_attributes)
            raise ManifestParseException(
                f'The attribute "{first_key}" is not applicable at {attribute_path.path}(with the current set'
                " of plugins). Possible attributes available (and not already used) are:"
                f" {remaining_valid_attribute_names}.{doc_ref}"
            )
        # Parse order is important for the root level (currently we use rule registration order)
        for key, provided_parser in self._parsers.items():
            value = orig_value.get(key)
            if value is None:
                if isinstance(provided_parser.parser, DispatchingObjectParser):
                    provided_parser.handler(
                        key, {}, attribute_path[key], parser_context
                    )
                continue
            value_path = attribute_path[key]
            if provided_parser is None:
                valid_keys = ", ".join(sorted(self._parsers.keys()))
                raise ManifestParseException(
                    f'Unknown or unsupported option "{key}" at {value_path.path}.'
                    " Valid options at this location are:"
                    f" {valid_keys}\n{doc_ref}"
                )
            parsed_value = provided_parser.parse(
                key, value, value_path, parser_context=parser_context
            )
            result[key] = parsed_value
        return result


class DispatchingTableParser(DispatchingParserBase[TP]):
    def __init__(self, base_type: TTP, manifest_attribute_path_template: str) -> None:
        super().__init__(manifest_attribute_path_template)
        self.base_type = base_type

    def parse(
        self,
        orig_value: object,
        attribute_path: "AttributePath",
        *,
        parser_context: "ParserContextData",
    ) -> TP:
        if isinstance(orig_value, str):
            key = orig_value
            value = None
            value_path = attribute_path
        elif isinstance(orig_value, dict):
            if len(orig_value) != 1:
                valid_keys = ", ".join(sorted(self._parsers.keys()))
                raise ManifestParseException(
                    f'The mapping "{attribute_path.path}" had two keys, but it should only have one top level key.'
                    " Maybe you are missing a list marker behind the second key or some indentation.  The"
                    f" possible keys are: {valid_keys}"
                )
            key, value = next(iter(orig_value.items()))
            value_path = attribute_path[key]
        else:
            raise ManifestParseException(
                f"The attribute {attribute_path.path} must be a string or a mapping."
            )
        provided_parser = self._parsers.get(key)
        if provided_parser is None:
            valid_keys = ", ".join(sorted(self._parsers.keys()))
            raise ManifestParseException(
                f'Unknown or unsupported action "{key}" at {value_path.path}.'
                " Valid actions at this location are:"
                f" {valid_keys}"
            )
        return provided_parser.parse(
            key, value, value_path, parser_context=parser_context
        )


@dataclasses.dataclass(slots=True)
class DeclarativeValuelessKeywordInputParser(DeclarativeInputParser[None]):
    inline_reference_documentation: Optional[ParserDocumentation] = None
    documentation_reference: Optional[str] = None

    def parse_input(
        self,
        value: object,
        path: "AttributePath",
        *,
        parser_context: Optional["ParserContextData"] = None,
    ) -> TD:
        if value is None:
            return cast("TD", value)
        if self.documentation_reference is not None:
            doc_ref = f" (Documentation: {self.documentation_reference})"
        else:
            doc_ref = ""
        raise ManifestParseException(
            f"Expected attribute {path.path} to be a string.{doc_ref}"
        )


SUPPORTED_DISPATCHABLE_TABLE_PARSERS = {
    InstallRule: "installations",
    TransformationRule: "packages.{{PACKAGE}}.transformations",
    DpkgMaintscriptHelperCommand: "packages.{{PACKAGE}}.conffile-management",
    ManifestCondition: "*.when",
}

OPARSER_MANIFEST_ROOT = "<ROOT>"
OPARSER_PACKAGES = "packages.{{PACKAGE}}"
OPARSER_MANIFEST_DEFINITIONS = "definitions"

SUPPORTED_DISPATCHABLE_OBJECT_PARSERS = {
    OPARSER_MANIFEST_ROOT: reference_documentation(
        reference_documentation_url=f"{DEBPUTY_DOC_ROOT_DIR}/MANIFEST-FORMAT.md",
    ),
    OPARSER_MANIFEST_DEFINITIONS: reference_documentation(
        title="Packager provided definitions",
        description="Reusable packager provided definitions such as manifest variables.",
        reference_documentation_url=f"{DEBPUTY_DOC_ROOT_DIR}/MANIFEST-FORMAT.md#packager-provided-definitions",
    ),
    OPARSER_PACKAGES: reference_documentation(
        title="Binary package rules",
        description=textwrap.dedent(
            """\
            Inside the manifest, the `packages` mapping can be used to define requests for the binary packages
            you want `debputy` to produce.  Each key inside `packages` must be the name of a binary package
            defined in `debian/control`.  The value is a dictionary defining which features that `debputy`
            should apply to that binary package.  An example could be:

                packages:
                    foo:
                        transformations:
                            - create-symlink:
                                  path: usr/share/foo/my-first-symlink
                                  target: /usr/share/bar/symlink-target
                            - create-symlink:
                                  path: usr/lib/{{DEB_HOST_MULTIARCH}}/my-second-symlink
                                  target: /usr/lib/{{DEB_HOST_MULTIARCH}}/baz/symlink-target
                    bar:
                        transformations:
                        - create-directories:
                           - some/empty/directory.d
                           - another/empty/integration-point.d
                        - create-directories:
                             path: a/third-empty/directory.d
                             owner: www-data
                             group: www-data

            In this case, `debputy` will create some symlinks inside the `foo` package and some directories for
            the `bar` package.  The following subsections define the keys you can use under each binary package.
        """
        ),
        reference_documentation_url=f"{DEBPUTY_DOC_ROOT_DIR}/MANIFEST-FORMAT.md#binary-package-rules",
    ),
}


@dataclasses.dataclass(slots=True)
class PluginProvidedManifestVariable:
    plugin_metadata: DebputyPluginMetadata
    variable_name: str
    variable_value: Optional[Union[str, Callable[[VariableContext], str]]]
    is_context_specific_variable: bool
    variable_reference_documentation: Optional[str] = None
    is_documentation_placeholder: bool = False
    is_for_special_case: bool = False

    @property
    def is_internal(self) -> bool:
        return self.variable_name.startswith("_") or ":_" in self.variable_name

    @property
    def is_token(self) -> bool:
        return self.variable_name.startswith("token:")

    def resolve(self, variable_context: VariableContext) -> str:
        value_resolver = self.variable_value
        if isinstance(value_resolver, str):
            res = value_resolver
        else:
            res = value_resolver(variable_context)
        return res


@dataclasses.dataclass(slots=True, frozen=True)
class AutomaticDiscardRuleExample:
    content: Sequence[Tuple[PathDef, bool]]
    description: Optional[str] = None


def automatic_discard_rule_example(
    *content: Union[str, PathDef, Tuple[Union[str, PathDef], bool]],
    example_description: Optional[str] = None,
) -> AutomaticDiscardRuleExample:
    """Provide an example for an automatic discard rule

    The return value of this method should be passed to the `examples` parameter of
    `automatic_discard_rule` method - either directly for a single example or as a
    part of a sequence of examples.

    >>> # Possible example for an exclude rule for ".la" files
    >>> # Example shows two files; The ".la" file that will be removed and another file that
    >>> # will be kept.
    >>> automatic_discard_rule_example(   # doctest: +ELLIPSIS
    ...     "usr/lib/libfoo.la",
    ...     ("usr/lib/libfoo.so.1.0.0", False),
    ... )
    AutomaticDiscardRuleExample(...)

    Keep in mind that you have to explicitly include directories that are relevant for the test
    if you want them shown. Also, if a directory is excluded, all path beneath it will be
    automatically excluded in the example as well. Your example data must account for that.

    >>> # Possible example for python cache file discard rule
    >>> # In this example, we explicitly list the __pycache__ directory itself because we
    >>> # want it shown in the output (otherwise, we could have omitted it)
    >>> automatic_discard_rule_example(   # doctest: +ELLIPSIS
    ...     (".../foo.py", False),
    ...     ".../__pycache__/",
    ...     ".../__pycache__/...",
    ...     ".../foo.pyc",
    ...     ".../foo.pyo",
    ... )
    AutomaticDiscardRuleExample(...)

    Note: Even if `__pycache__` had been implicit, the result would have been the same. However,
    the rendered example would not have shown the directory on its own.  The use of `...` as
    path names is useful for denoting "anywhere" or "anything". Though, there is nothing "magic"
    about this name - it happens to be allowed as a path name (unlike `.` or `..`).

    These examples can be seen via `debputy plugin show automatic-discard-rules <name-here>`.

    :param content: The content of the example.  Each element can be either a path definition or
      a tuple of a path definition followed by a verdict (boolean). Each provided path definition
      describes the paths to be presented in the example. Implicit paths such as parent
      directories will be created but not shown in the example.  Therefore, if a directory is
      relevant to the example, be sure to explicitly list it.

      The verdict associated with a path determines whether the path should be discarded (when
      True) or kept (when False). When a path is not explicitly associated with a verdict, the
      verdict is assumed to be discarded (True).
    :param example_description: An optional description displayed together with the example.
    :return: An opaque data structure containing the example.
    """
    example = []
    for d in content:
        if not isinstance(d, tuple):
            pd = d
            verdict = True
        else:
            pd, verdict = d

        path_def = as_path_def(pd)
        example.append((path_def, verdict))

    if not example:
        raise ValueError("At least one path must be given for an example")

    return AutomaticDiscardRuleExample(
        tuple(example),
        description=example_description,
    )


@dataclasses.dataclass(slots=True, frozen=True)
class PluginProvidedPackageProcessor:
    processor_id: str
    applies_to_package_types: FrozenSet[str]
    package_processor: PackageProcessor
    dependencies: FrozenSet[Tuple[str, str]]
    plugin_metadata: DebputyPluginMetadata

    def applies_to(self, binary_package: BinaryPackage) -> bool:
        return binary_package.package_type in self.applies_to_package_types

    @property
    def dependency_id(self) -> Tuple[str, str]:
        return self.plugin_metadata.plugin_name, self.processor_id

    def run_package_processor(
        self,
        fs_root: "VirtualPath",
        unused: None,
        context: "PackageProcessingContext",
    ) -> None:
        self.package_processor(fs_root, unused, context)


@dataclasses.dataclass(slots=True, frozen=True)
class PluginProvidedDiscardRule:
    name: str
    plugin_metadata: DebputyPluginMetadata
    discard_check: Callable[[VirtualPath], bool]
    reference_documentation: Optional[str]
    examples: Sequence[AutomaticDiscardRuleExample] = tuple()

    def should_discard(self, path: VirtualPath) -> bool:
        return self.discard_check(path)


@dataclasses.dataclass(slots=True, frozen=True)
class ServiceManagerDetails:
    service_manager: str
    service_detector: "ServiceDetector"
    service_integrator: "ServiceIntegrator"
    plugin_metadata: DebputyPluginMetadata


ReferenceValue = TypedDict(
    "ReferenceValue",
    {
        "description": str,
    },
)


def _reference_data_value(
    *,
    description: str,
) -> ReferenceValue:
    return {
        "description": description,
    }


KnownPackagingFileCategories = Literal[
    "generated",
    "generic-template",
    "ppf-file",
    "ppf-control-file",
    "maint-config",
    "pkg-metadata",
    "pkg-helper-config",
    "testing",
    "lint-config",
]
KNOWN_PACKAGING_FILE_CATEGORY_DESCRIPTIONS: Mapping[
    KnownPackagingFileCategories, ReferenceValue
] = {
    "generated": _reference_data_value(
        description="The file is (likely) generated from another file"
    ),
    "generic-template": _reference_data_value(
        description="The file is (likely) a generic template that generates a known packaging file. While the"
        " file is annotated as if it was the target file, the file might uses a custom template"
        " language inside it."
    ),
    "ppf-file": _reference_data_value(
        description="Packager provided file to be installed on the file system - usually as-is."
        " When `install-pattern` or `install-path` are provided, this is where the file is installed."
    ),
    "ppf-control-file": _reference_data_value(
        description="Packager provided file that becomes a control file - possible after processing. "
        " If `install-pattern` or `install-path` are provided, they denote where the is placed"
        " (generally, this will be of the form `DEBIAN/<name>`)"
    ),
    "maint-config": _reference_data_value(
        description="Maintenance configuration for a specific tool that the maintainer uses (tool / style preferences)"
    ),
    "pkg-metadata": _reference_data_value(
        description="The file is related to standard package metadata (usually documented in Debian Policy)"
    ),
    "pkg-helper-config": _reference_data_value(
        description="The file is packaging helper configuration or instruction file"
    ),
    "testing": _reference_data_value(
        description="The file is related to automated testing (autopkgtests, salsa/gitlab CI)."
    ),
    "lint-config": _reference_data_value(
        description="The file is related to a linter (such as overrides for false-positives or style preferences)"
    ),
}

KnownPackagingConfigFeature = Literal[
    "dh-filearray",
    "dh-filedoublearray",
    "dh-hash-subst",
    "dh-dollar-subst",
    "dh-glob",
    "dh-partial-glob",
    "dh-late-glob",
    "dh-glob-after-execute",
    "dh-executable-config",
    "dh-custom-format",
    "dh-file-list",
    "dh-install-list",
    "dh-install-list-dest-dir-like-dh_install",
    "dh-install-list-fixed-dest-dir",
    "dh-fixed-dest-dir",
    "dh-exec-rename",
    "dh-docs-only",
]

KNOWN_PACKAGING_FILE_CONFIG_FEATURE_DESCRIPTION: Mapping[
    KnownPackagingConfigFeature, ReferenceValue
] = {
    "dh-filearray": _reference_data_value(
        description="The file will be read as a list of space/newline separated tokens",
    ),
    "dh-filedoublearray": _reference_data_value(
        description="Each line in the file will be read as a list of space-separated tokens",
    ),
    "dh-hash-subst": _reference_data_value(
        description="Supports debhelper #PACKAGE# style substitutions (udebs often excluded)",
    ),
    "dh-dollar-subst": _reference_data_value(
        description="Supports debhelper ${PACKAGE} style substitutions (usually requires compat 13+)",
    ),
    "dh-glob": _reference_data_value(
        description="Supports standard debhelper globing",
    ),
    "dh-partial-glob": _reference_data_value(
        description="Supports standard debhelper globing but only to a subset of the values (implies dh-late-glob)",
    ),
    "dh-late-glob": _reference_data_value(
        description="Globbing is done separately instead of using the built-in function",
    ),
    "dh-glob-after-execute": _reference_data_value(
        description="When the dh config file is executable, the generated output will be subject to globbing",
    ),
    "dh-executable-config": _reference_data_value(
        description="If marked executable, debhelper will execute the file and read its output",
    ),
    "dh-custom-format": _reference_data_value(
        description="The dh tool will or may have a custom parser for this file",
    ),
    "dh-file-list": _reference_data_value(
        description="The dh file contains a list of paths to be processed",
    ),
    "dh-install-list": _reference_data_value(
        description="The dh file contains a list of paths/globs to be installed but the tool specific knowledge"
        " required to understand the file cannot be conveyed via this interface.",
    ),
    "dh-install-list-dest-dir-like-dh_install": _reference_data_value(
        description="The dh file is processed similar to dh_install (notably dest-dir handling derived"
        " from the path or the last token on the line)",
    ),
    "dh-install-list-fixed-dest-dir": _reference_data_value(
        description="The dh file is an install list and the dest-dir is always the same for all patterns"
        " (when `install-pattern` or `install-path` are provided, they identify the directory - not the file location)",
    ),
    "dh-exec-rename": _reference_data_value(
        description="When `dh-exec` is the interpreter of this dh config file, its renaming (=>) feature can be"
        " requested/used",
    ),
    "dh-docs-only": _reference_data_value(
        description="The dh config file is used for documentation only. Implicit <!nodocs> Build-Profiles support",
    ),
}

CONFIG_FEATURE_ALIASES: Dict[
    KnownPackagingConfigFeature, List[Tuple[KnownPackagingConfigFeature, int]]
] = {
    "dh-filearray": [
        ("dh-filearray", 0),
        ("dh-executable-config", 9),
        ("dh-dollar-subst", 13),
    ],
    "dh-filedoublearray": [
        ("dh-filedoublearray", 0),
        ("dh-executable-config", 9),
        ("dh-dollar-subst", 13),
    ],
}


def _implies(
    features: List[KnownPackagingConfigFeature],
    seen: Set[KnownPackagingConfigFeature],
    implying: Sequence[KnownPackagingConfigFeature],
    implied: KnownPackagingConfigFeature,
) -> None:
    if implied in seen:
        return
    if all(f in seen for f in implying):
        seen.add(implied)
        features.append(implied)


def expand_known_packaging_config_features(
    compat_level: int,
    features: List[KnownPackagingConfigFeature],
) -> List[KnownPackagingConfigFeature]:
    final_features: List[KnownPackagingConfigFeature] = []
    seen = set()
    for feature in features:
        expanded = CONFIG_FEATURE_ALIASES.get(feature)
        if not expanded:
            expanded = [(feature, 0)]
        for v, c in expanded:
            if compat_level < c or v in seen:
                continue
            seen.add(v)
            final_features.append(v)
    if "dh-glob" in seen and "dh-late-glob" in seen:
        final_features.remove("dh-glob")

    _implies(final_features, seen, ["dh-partial-glob"], "dh-late-glob")
    _implies(
        final_features,
        seen,
        ["dh-late-glob", "dh-executable-config"],
        "dh-glob-after-execute",
    )
    return sorted(final_features)


class InstallPatternDHCompatRule(DebputyParsedContent):
    install_pattern: NotRequired[str]
    add_config_features: NotRequired[List[KnownPackagingConfigFeature]]
    starting_with_compat_level: NotRequired[int]


class KnownPackagingFileInfo(DebputyParsedContent):
    # Exposed directly in the JSON plugin parsing; be careful with changes
    path: NotRequired[str]
    pkgfile: NotRequired[str]
    detection_method: NotRequired[Literal["path", "dh.pkgfile"]]
    file_categories: NotRequired[List[KnownPackagingFileCategories]]
    documentation_uris: NotRequired[List[str]]
    debputy_cmd_templates: NotRequired[List[List[str]]]
    debhelper_commands: NotRequired[List[str]]
    config_features: NotRequired[List[KnownPackagingConfigFeature]]
    install_pattern: NotRequired[str]
    dh_compat_rules: NotRequired[List[InstallPatternDHCompatRule]]
    default_priority: NotRequired[int]
    post_formatting_rewrite: NotRequired[Literal["period-to-underscore"]]
    packageless_is_fallback_for_all_packages: NotRequired[bool]


@dataclasses.dataclass(slots=True)
class PluginProvidedKnownPackagingFile:
    info: KnownPackagingFileInfo
    detection_method: Literal["path", "dh.pkgfile"]
    detection_value: str
    plugin_metadata: DebputyPluginMetadata


@dataclasses.dataclass(slots=True, frozen=True)
class PluginProvidedTypeMapping:
    mapped_type: TypeMapping[Any, Any]
    reference_documentation: Optional[TypeMappingDocumentation]
    plugin_metadata: DebputyPluginMetadata


class PackageDataTable:
    def __init__(self, package_data_table: Mapping[str, "BinaryPackageData"]) -> None:
        self._package_data_table = package_data_table
        # This is enabled for metadata-detectors. But it is deliberate not enabled for package processors,
        # because it is not clear how it should interact with dependencies. For metadata-detectors, things
        # read-only and there are no dependencies, so we cannot "get them wrong".
        self.enable_cross_package_checks = False

    def __iter__(self) -> Iterator["BinaryPackageData"]:
        return iter(self._package_data_table.values())

    def __getitem__(self, item: str) -> "BinaryPackageData":
        return self._package_data_table[item]

    def __contains__(self, item: str) -> bool:
        return item in self._package_data_table


class PackageProcessingContextProvider(PackageProcessingContext):
    __slots__ = (
        "_manifest",
        "_binary_package",
        "_related_udeb_package",
        "_package_data_table",
        "_cross_check_cache",
    )

    def __init__(
        self,
        manifest: "HighLevelManifest",
        binary_package: BinaryPackage,
        related_udeb_package: Optional[BinaryPackage],
        package_data_table: PackageDataTable,
    ) -> None:
        self._manifest = manifest
        self._binary_package = binary_package
        self._related_udeb_package = related_udeb_package
        self._package_data_table = ref(package_data_table)
        self._cross_check_cache: Optional[
            Sequence[Tuple[BinaryPackage, "VirtualPath"]]
        ] = None

    def _package_state_for(
        self,
        package: BinaryPackage,
    ) -> "PackageTransformationDefinition":
        return self._manifest.package_state_for(package.name)

    def _package_version_for(
        self,
        package: BinaryPackage,
    ) -> str:
        package_state = self._package_state_for(package)
        version = package_state.binary_version
        if version is not None:
            return version
        return self._manifest.source_version(
            include_binnmu_version=not package.is_arch_all
        )

    @property
    def binary_package(self) -> BinaryPackage:
        return self._binary_package

    @property
    def related_udeb_package(self) -> Optional[BinaryPackage]:
        return self._related_udeb_package

    @property
    def binary_package_version(self) -> str:
        return self._package_version_for(self._binary_package)

    @property
    def related_udeb_package_version(self) -> Optional[str]:
        udeb = self._related_udeb_package
        if udeb is None:
            return None
        return self._package_version_for(udeb)

    def accessible_package_roots(self) -> Iterable[Tuple[BinaryPackage, "VirtualPath"]]:
        package_table = self._package_data_table()
        if package_table is None:
            raise ReferenceError(
                "Internal error: package_table was garbage collected too early"
            )
        if not package_table.enable_cross_package_checks:
            raise PluginAPIViolationError(
                "Cross package content checks are not available at this time."
            )
        cache = self._cross_check_cache
        if cache is None:
            matches = []
            pkg = self.binary_package
            for pkg_data in package_table:
                if pkg_data.binary_package.name == pkg.name:
                    continue
                res = package_cross_check_precheck(pkg, pkg_data.binary_package)
                if not res[0]:
                    continue
                matches.append((pkg_data.binary_package, pkg_data.fs_root))
            cache = tuple(matches) if matches else tuple()
            self._cross_check_cache = cache
        return cache


@dataclasses.dataclass(slots=True, frozen=True)
class PluginProvidedTrigger:
    dpkg_trigger_type: DpkgTriggerType
    dpkg_trigger_target: str
    provider: DebputyPluginMetadata
    provider_source_id: str

    def serialized_format(self) -> str:
        return f"{self.dpkg_trigger_type} {self.dpkg_trigger_target}"