cleveractors-core/robot/RegistryClientLib.py

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

Provides keywords for testing the httpx-based RegistryClient against
a fake registry server (started by the test suite).
"""

from __future__ import annotations

import asyncio
from typing import Any, Optional

from cleveractors.registry.client import RegistryClient
from cleveractors.registry.exceptions import (
    PackageNotFoundError,
    RegistryNetworkError,
)

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


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

    def __init__(self) -> None:
        self._client: Optional[RegistryClient] = None
        self._result: Any = None
        self._error: Optional[Exception] = None

    def create_registry_client(self, base_url: str) -> None:
        self._client = RegistryClient(base_url=base_url)

    def create_registry_client_with_key(self, base_url: str, api_key: str) -> None:
        self._client = RegistryClient(base_url=base_url, api_key=api_key)

    def get_package_by_id(self, package_id: str) -> None:
        async def _call() -> None:
            self._result = await self._client.get_package(package_id)

        self._run(_call)

    def resolve_package_ref(
        self,
        package_type: str,
        namespace: str,
        name: str,
        version: str = "",
    ) -> None:
        async def _call() -> None:
            ver: Optional[str] = version if version else None
            self._result = await self._client.resolve_package(
                package_type, namespace, name, ver
            )

        self._run(_call)

    def browse_packages(self) -> None:
        async def _call() -> None:
            self._result = await self._client.browse()

        self._run(_call)

    def browse_packages_by_type(self, package_type: str) -> None:
        async def _call() -> None:
            self._result = await self._client.browse(package_type=package_type)

        self._run(_call)

    def discover_registry(self) -> None:
        async def _call() -> None:
            self._result = await self._client.discover()

        self._run(_call)

    def result_has_key(self, key: str) -> None:
        if key not in self._result:
            raise AssertionError(
                f"Result missing key {key!r}: {sorted(self._result.keys())}"
            )

    def result_key_equals(self, key: str, expected: str) -> None:
        actual = str(self._result.get(key))
        if actual != expected:
            raise AssertionError(f"Key {key}: expected {expected!r}, got {actual!r}")

    def result_count_equals(self, expected: str) -> None:
        actual = len(self._result.get("packages", []))
        if str(actual) != expected:
            raise AssertionError(f"Count: expected {expected!r}, got {actual!r}")

    def get_package_should_raise_not_found(self, package_id: str) -> None:
        async def _call() -> None:
            try:
                await self._client.get_package(package_id)
            except PackageNotFoundError as exc:
                self._error = exc

        self._run(_call)
        if self._error is None:
            raise AssertionError(
                f"Expected PackageNotFoundError for {package_id!r}, none raised"
            )

    def get_package_should_raise_network_error(self, package_id: str) -> None:
        async def _call() -> None:
            try:
                await self._client.get_package(package_id)
            except RegistryNetworkError as exc:
                self._error = exc

        self._run(_call)
        if self._error is None:
            raise AssertionError(
                f"Expected RegistryNetworkError for {package_id!r}, none raised"
            )

    def close_client(self) -> None:
        async def _call() -> None:
            await self._client.close()

        self._run(_call)

    @staticmethod
    def _run(coro_fn: Any, timeout: float = 30.0) -> None:
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                future = asyncio.run_coroutine_threadsafe(coro_fn(), loop)
                future.result(timeout=timeout)
            else:
                loop.run_until_complete(asyncio.wait_for(coro_fn(), timeout=timeout))
        except RuntimeError:
            asyncio.run(asyncio.wait_for(coro_fn(), timeout=timeout))