CoreRasurae
054b432197
CI / lint (pull_request) Successful in 56s
CI / typecheck (pull_request) Successful in 56s
CI / security (pull_request) Successful in 1m2s
CI / quality (pull_request) Successful in 41s
CI / integration_tests (pull_request) Successful in 1m11s
CI / build (pull_request) Successful in 45s
CI / unit_tests (pull_request) Successful in 3m29s
CI / coverage (pull_request) Successful in 3m14s
CI / status-check (pull_request) Successful in 4s
CI / lint (push) Successful in 48s
CI / typecheck (push) Successful in 58s
CI / security (push) Successful in 1m1s
CI / quality (push) Successful in 39s
CI / integration_tests (push) Successful in 1m15s
CI / build (push) Successful in 41s
CI / unit_tests (push) Successful in 3m31s
CI / coverage (push) Successful in 3m27s
CI / status-check (push) Successful in 3s
- Add RegistryCache with LRU eviction, TTL expiry, SHA-1 content validation - Add CacheFactory for centralised cache configuration (Factory Method pattern) - Add CacheStats dataclass for hit/miss/eviction observability - Implement singleflight coalescing for concurrent-miss protection - Use tuple keys for resolve store to prevent separator collision - Fix cancellation-safe singleflight via plain Future + asyncio.shield - Independent max_size budgets per store (content + resolve) - Concurrent test assertions strengthened - All test imports moved to module level per CONTRIBUTING.md - Unused dead code removed from tests ISSUES CLOSED: #28
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""Robot Framework keyword library for RegistryCache integration tests.
|
|
|
|
Provides keywords for testing the RegistryCache against a fake registry
|
|
server started by the test suite.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any, Optional
|
|
|
|
from cleveractors.registry.cache import RegistryCache
|
|
from cleveractors.registry.client import RegistryClient
|
|
|
|
ROBOT_LIBRARY_SCOPE = "TEST SUITE" # pragma: no cover - integration test library
|
|
|
|
|
|
class RegistryCacheLib: # pragma: no cover - integration test library
|
|
"""Keyword library for RegistryCache Robot Framework integration tests."""
|
|
|
|
def __init__(self) -> None:
|
|
self._client: Optional[RegistryClient] = None
|
|
self._cache: Optional[RegistryCache] = None
|
|
self._result: Any = None
|
|
|
|
def create_registry_cache(self, base_url: str) -> None:
|
|
self._client = RegistryClient(base_url=base_url)
|
|
self._cache = RegistryCache(self._client, validate_content=False)
|
|
|
|
def create_registry_cache_with_config(
|
|
self, base_url: str, max_size: str, ttl: str
|
|
) -> None:
|
|
self._client = RegistryClient(base_url=base_url)
|
|
self._cache = RegistryCache(
|
|
self._client, max_size=int(max_size), ttl=float(ttl), validate_content=False
|
|
)
|
|
|
|
def cache_get_package(self, package_id: str) -> None:
|
|
async def _call() -> None:
|
|
self._result = await self._cache.get_package(package_id)
|
|
|
|
self._run(_call)
|
|
|
|
def cache_stats_hits_equals(self, expected: str) -> None:
|
|
actual = str(self._cache.stats.hits)
|
|
if actual != expected:
|
|
raise AssertionError(f"Hits: expected {expected!r}, got {actual!r}")
|
|
|
|
def cache_stats_misses_equals(self, expected: str) -> None:
|
|
actual = str(self._cache.stats.misses)
|
|
if actual != expected:
|
|
raise AssertionError(f"Misses: expected {expected!r}, got {actual!r}")
|
|
|
|
def cache_stats_evictions_equals(self, expected: str) -> None:
|
|
actual = str(self._cache.stats.evictions)
|
|
if actual != expected:
|
|
raise AssertionError(f"Evictions: expected {expected!r}, got {actual!r}")
|
|
|
|
def cache_should_not_contain(self, package_id: str) -> None:
|
|
if package_id in self._cache:
|
|
raise AssertionError(f"Cache still contains {package_id!r}")
|
|
|
|
def cache_should_contain(self, package_id: str) -> None:
|
|
if package_id not in self._cache:
|
|
raise AssertionError(f"Cache does not contain {package_id!r}")
|
|
|
|
def cache_clear(self) -> None:
|
|
async def _call() -> None:
|
|
await self._cache.clear()
|
|
|
|
self._run(_call)
|
|
|
|
def cache_invalidate(self, package_id: str) -> None:
|
|
async def _call() -> None:
|
|
self._result = await self._cache.invalidate(package_id)
|
|
|
|
self._run(_call)
|
|
|
|
def invalidate_result_should_be(self, expected: str) -> None:
|
|
result_str = str(self._result)
|
|
if result_str != expected:
|
|
raise AssertionError(
|
|
f"Invalidate result: expected {expected!r}, got {result_str!r}"
|
|
)
|
|
|
|
def close_cache(self) -> None:
|
|
async def _call() -> None:
|
|
await self._cache.close()
|
|
|
|
self._run(_call)
|
|
|
|
@staticmethod
|
|
def _run(coro_fn: Any, timeout: float = 30.0) -> None:
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
future = asyncio.run_coroutine_threadsafe(coro_fn(), loop)
|
|
future.result(timeout=timeout)
|
|
except RuntimeError:
|
|
asyncio.run(asyncio.wait_for(coro_fn(), timeout=timeout))
|