cleveractors-core/robot/ResolverLib.py

"""Robot Framework keyword library for ReferenceResolver integration tests.

Provides keywords for testing the ReferenceResolver against a fake
registry server (started by the test suite).
"""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable

from cleveractors.registry.client import RegistryClient
from cleveractors.registry.exceptions import InvalidPackageReferenceError
from cleveractors.registry.resolver import (
    ReferenceResolver,
    is_concrete_version,
    is_version_alias,
    resolve_version,
)
from cleveractors.registry.types import PackageId, PackageReference, ReferenceType

ROBOT_LIBRARY_SCOPE = "TEST SUITE"  # pragma: no cover - integration test library


class ResolverLib:  # pragma: no cover - integration test library
    """Keyword library for ReferenceResolver Robot Framework integration tests."""

    def __init__(self) -> None:
        self._resolver: ReferenceResolver | None = None
        self._client: RegistryClient | None = None
        self._parsed_ref: PackageReference | None = None
        self._resolved_id: PackageId | None = None
        self._result: str | bool | None = None
        self._error: Exception | None = None

    # ── setup / teardown ────────────────────────────────────────────────

    def create_resolver_without_client(self) -> None:
        self._resolver = ReferenceResolver()

    def create_resolver_with_client(self, base_url: str) -> None:
        self._client = RegistryClient(base_url=base_url)
        self._resolver = ReferenceResolver(client=self._client)

    def close_resolver_client(self) -> None:
        async def _close() -> None:
            if self._client is not None:
                await self._client.close()

        self._run(_close)

    # ── parse ───────────────────────────────────────────────────────────

    def parse_reference(self, ref_str: str) -> None:
        self._error = None
        try:
            self._parsed_ref = ReferenceResolver.parse(ref_str)
        except InvalidPackageReferenceError as exc:
            self._error = exc

    def parsed_reference_type_should_be(self, expected: str) -> None:
        if self._parsed_ref is None:
            raise AssertionError("No parsed reference available")
        actual = self._parsed_ref.reference_type.value
        if actual != expected:
            raise AssertionError(
                f"Expected reference_type {expected!r}, got {actual!r}"
            )

    def parsed_server_should_be(self, expected: str) -> None:
        actual = self._parsed_ref.server
        if actual != expected:
            raise AssertionError(f"Expected server {expected!r}, got {actual!r}")

    def parsed_namespace_should_be(self, expected: str) -> None:
        actual = self._parsed_ref.namespace
        if actual != expected:
            raise AssertionError(f"Expected namespace {expected!r}, got {actual!r}")

    def parsed_name_should_be(self, expected: str) -> None:
        actual = self._parsed_ref.name
        if actual != expected:
            raise AssertionError(f"Expected name {expected!r}, got {actual!r}")

    def parsed_version_should_be(self, expected: str) -> None:
        actual = self._parsed_ref.version
        if actual != expected:
            raise AssertionError(f"Expected version {expected!r}, got {actual!r}")

    def parsed_id_string_should_be(self, expected: str) -> None:
        actual = self._parsed_ref.id_string
        if actual != expected:
            raise AssertionError(f"Expected id_string {expected!r}, got {actual!r}")

    def parsed_original_reference_should_be(self, expected: str) -> None:
        actual = self._parsed_ref.original_reference
        if actual != expected:
            raise AssertionError(
                f"Expected original_reference {expected!r}, got {actual!r}"
            )

    def parsed_server_should_be_none(self) -> None:
        if self._parsed_ref.server is not None:
            raise AssertionError(
                f"Expected server None, got {self._parsed_ref.server!r}"
            )

    def parsed_namespace_should_be_none(self) -> None:
        if self._parsed_ref.namespace is not None:
            raise AssertionError(
                f"Expected namespace None, got {self._parsed_ref.namespace!r}"
            )

    def parse_should_have_raised(self, phrase: str) -> None:
        if self._error is None:
            raise AssertionError("Expected error but none was raised")
        if phrase not in str(self._error):
            raise AssertionError(
                f"Expected error containing {phrase!r}, got {self._error!r}"
            )

    # ── resolve ─────────────────────────────────────────────────────────

    def resolve_reference(self, ref_str: str) -> None:
        self._error = None

        async def _resolve() -> None:
            try:
                self._resolved_id = await self._resolver.resolve(ref_str)
            except InvalidPackageReferenceError as exc:
                self._error = exc

        self._run(_resolve)

    def resolved_package_id_should_be(self, expected: str) -> None:
        if self._resolved_id is None:
            raise AssertionError("No resolved package ID available")
        actual = self._resolved_id.id_string
        if actual != expected:
            raise AssertionError(f"Expected resolved id {expected!r}, got {actual!r}")

    def resolve_should_have_raised(self, phrase: str) -> None:
        if self._error is None:
            raise AssertionError("Expected error but none was raised")
        if phrase not in str(self._error):
            raise AssertionError(
                f"Expected error containing {phrase!r}, got {self._error!r}"
            )

    # ── version resolution ──────────────────────────────────────────────

    def resolve_version_against(self, version: str, versions_csv: str) -> None:
        self._error = None
        available = (
            [v.strip() for v in versions_csv.split(",") if v.strip()]
            if versions_csv.strip()
            else []
        )
        try:
            self._result = resolve_version(version, available)
        except InvalidPackageReferenceError as exc:
            self._error = exc

    def resolved_version_should_be(self, expected: str) -> None:
        if self._result != expected:
            raise AssertionError(
                f"Expected resolved version {expected!r}, got {self._result!r}"
            )

    def resolve_version_should_have_raised(self, phrase: str) -> None:
        if self._error is None:
            raise AssertionError("Expected error but none was raised")
        if phrase not in str(self._error):
            raise AssertionError(
                f"Expected error containing {phrase!r}, got {self._error!r}"
            )

    # ── version classification ──────────────────────────────────────────

    def check_is_concrete_version(self, version: str) -> None:
        self._result = is_concrete_version(version)

    def check_is_version_alias(self, version: str) -> None:
        self._result = is_version_alias(version)

    def classification_result_should_be_true(self) -> None:
        if self._result is not True:
            raise AssertionError(f"Expected True, got {self._result!r}")

    def classification_result_should_be_false(self) -> None:
        if self._result is not False:
            raise AssertionError(f"Expected False, got {self._result!r}")

    # ── async runner ────────────────────────────────────────────────────

    @staticmethod
    def _run(coro_fn: Callable[[], Awaitable[None]], timeout: float = 30.0) -> None:
        asyncio.run(asyncio.wait_for(coro_fn(), timeout=timeout))