ansible/test/lib/ansible_test/_internal/metadata.py

"""Test metadata for passing data to delegated tests."""

from __future__ import annotations

import dataclasses
import typing as t

from .util import (
    display,
    generate_name,
    ANSIBLE_TEST_ROOT,
    ANSIBLE_LIB_ROOT,
)

from .io import (
    write_json_file,
    read_json_file,
)

from .diff import (
    parse_diff,
    FileDiff,
)

if t.TYPE_CHECKING:
    from .debugging import DebuggerSettings


class Metadata:
    """Metadata object for passing data to delegated tests."""

    def __init__(self, debugger_flags: DebuggerFlags) -> None:
        """Initialize metadata."""
        self.changes: dict[str, tuple[tuple[int, int], ...]] = {}
        self.cloud_config: t.Optional[dict[str, dict[str, t.Union[int, str, bool]]]] = None
        self.change_description: t.Optional[ChangeDescription] = None
        self.ci_provider: t.Optional[str] = None
        self.session_id = generate_name()
        self.ansible_lib_root = ANSIBLE_LIB_ROOT
        self.ansible_test_root = ANSIBLE_TEST_ROOT
        self.collection_root: str | None = None
        self.debugger_flags = debugger_flags
        self.debugger_settings: DebuggerSettings | None = None
        self.loaded = False

    def populate_changes(self, diff: t.Optional[list[str]]) -> None:
        """Populate the changeset using the given diff."""
        patches = parse_diff(diff)
        patches: list[FileDiff] = sorted(patches, key=lambda k: k.new.path)

        self.changes = dict((patch.new.path, tuple(patch.new.ranges)) for patch in patches)

        renames = [patch.old.path for patch in patches if patch.old.path != patch.new.path and patch.old.exists and patch.new.exists]
        deletes = [patch.old.path for patch in patches if not patch.new.exists]

        # make sure old paths which were renamed or deleted are registered in changes
        for path in renames + deletes:
            if path in self.changes:
                # old path was replaced with another file
                continue

            # failed tests involving deleted files should be using line 0 since there is no content remaining
            self.changes[path] = ((0, 0),)

    def to_dict(self) -> dict[str, t.Any]:
        """Return a dictionary representation of the metadata."""
        return dict(
            changes=self.changes,
            cloud_config=self.cloud_config,
            ci_provider=self.ci_provider,
            change_description=self.change_description.to_dict() if self.change_description else None,
            session_id=self.session_id,
            ansible_lib_root=self.ansible_lib_root,
            ansible_test_root=self.ansible_test_root,
            collection_root=self.collection_root,
            debugger_flags=dataclasses.asdict(self.debugger_flags),
            debugger_settings=self.debugger_settings.as_dict() if self.debugger_settings else None,
        )

    def to_file(self, path: str) -> None:
        """Write the metadata to the specified file."""
        data = self.to_dict()

        display.info('>>> Metadata: %s\n%s' % (path, data), verbosity=3)

        write_json_file(path, data)

    @staticmethod
    def from_file(path: str) -> Metadata:
        """Return metadata loaded from the specified file."""
        data = read_json_file(path)
        return Metadata.from_dict(data)

    @staticmethod
    def from_dict(data: dict[str, t.Any]) -> Metadata:
        """Return metadata loaded from the specified dictionary."""
        from .debugging import DebuggerSettings

        metadata = Metadata(
            debugger_flags=DebuggerFlags(**data['debugger_flags']),
        )

        metadata.changes = data['changes']
        metadata.cloud_config = data['cloud_config']
        metadata.ci_provider = data['ci_provider']
        metadata.change_description = ChangeDescription.from_dict(data['change_description']) if data['change_description'] else None
        metadata.session_id = data['session_id']
        metadata.ansible_lib_root = data['ansible_lib_root']
        metadata.ansible_test_root = data['ansible_test_root']
        metadata.collection_root = data['collection_root']
        metadata.debugger_settings = DebuggerSettings.from_dict(data['debugger_settings']) if data['debugger_settings'] else None
        metadata.loaded = True

        return metadata


class ChangeDescription:
    """Description of changes."""

    def __init__(self) -> None:
        self.command: str = ''
        self.changed_paths: list[str] = []
        self.deleted_paths: list[str] = []
        self.regular_command_targets: dict[str, list[str]] = {}
        self.focused_command_targets: dict[str, list[str]] = {}
        self.no_integration_paths: list[str] = []

    @property
    def targets(self) -> t.Optional[list[str]]:
        """Optional list of target names."""
        return self.regular_command_targets.get(self.command)

    @property
    def focused_targets(self) -> t.Optional[list[str]]:
        """Optional list of focused target names."""
        return self.focused_command_targets.get(self.command)

    def to_dict(self) -> dict[str, t.Any]:
        """Return a dictionary representation of the change description."""
        return dict(
            command=self.command,
            changed_paths=self.changed_paths,
            deleted_paths=self.deleted_paths,
            regular_command_targets=self.regular_command_targets,
            focused_command_targets=self.focused_command_targets,
            no_integration_paths=self.no_integration_paths,
        )

    @staticmethod
    def from_dict(data: dict[str, t.Any]) -> ChangeDescription:
        """Return a change description loaded from the given dictionary."""
        changes = ChangeDescription()
        changes.command = data['command']
        changes.changed_paths = data['changed_paths']
        changes.deleted_paths = data['deleted_paths']
        changes.regular_command_targets = data['regular_command_targets']
        changes.focused_command_targets = data['focused_command_targets']
        changes.no_integration_paths = data['no_integration_paths']

        return changes


@dataclasses.dataclass(frozen=True, kw_only=True)
class DebuggerFlags:
    """Flags for enabling specific debugging features."""

    self: bool = False
    """Debug ansible-test itself."""

    ansiballz: bool = False
    """Debug AnsiballZ modules."""

    cli: bool = False
    """Debug Ansible CLI programs other than ansible-test."""

    on_demand: bool = False
    """Enable debugging features only when ansible-test is running under a debugger."""

    @property
    def enable(self) -> bool:
        """Return `True` if any debugger feature other than on-demand is enabled."""
        return any(getattr(self, field.name) for field in dataclasses.fields(self) if field.name != 'on_demand')

    @classmethod
    def all(cls, enabled: bool) -> t.Self:
        """Return a `DebuggerFlags` instance with all flags enabled or disabled."""
        return cls(**{field.name: enabled for field in dataclasses.fields(cls)})