Brian Bjarke JensenandCursor
f6ee9a3724
PR Title Check / check-title (pull_request) Successful in 7s
Test Python Package / unit-tests (pull_request) Successful in 9s
Code Quality Pipeline / code-quality (pull_request) Successful in 22s
Test Python Package / integration-tests (pull_request) Successful in 24s
Test Python Package / coverage-report (pull_request) Successful in 8s
Enables structural subtyping for consumers while preserving nominal adapter inheritance, instantiation guards, and scan_keys defaults. Co-authored-by: Cursor <cursoragent@cursor.com>
135 lines
3.7 KiB
Python
135 lines
3.7 KiB
Python
"""Unit tests for structural typing of public interfaces."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
from io import BytesIO
|
|
from typing import Any
|
|
|
|
from python_repositories.interfaces import (
|
|
ConnectionAwareInterface,
|
|
ContextAwareInterface,
|
|
JsonRepositoryInterface,
|
|
ObjectRepositoryInterface,
|
|
)
|
|
|
|
|
|
class InMemoryJsonRepo:
|
|
"""Plain class that satisfies JsonRepositoryInterface without inheritance."""
|
|
|
|
def get(self, key: str) -> dict[str, Any] | None:
|
|
return None
|
|
|
|
def set(self, key: str, data: dict[str, Any]) -> None:
|
|
pass
|
|
|
|
def delete(self, key: str) -> None:
|
|
pass
|
|
|
|
def list_keys(self, pattern: str) -> list[str]:
|
|
return []
|
|
|
|
def scan_keys(
|
|
self,
|
|
pattern: str,
|
|
*,
|
|
count: int | None = None,
|
|
) -> Iterator[str]:
|
|
del count
|
|
yield from self.list_keys(pattern)
|
|
|
|
|
|
class InMemoryObjectRepo:
|
|
"""Plain class that satisfies ObjectRepositoryInterface without inheritance."""
|
|
|
|
def get(self, object_name: str) -> BytesIO | None:
|
|
return None
|
|
|
|
def put(
|
|
self,
|
|
object_name: str,
|
|
data: BytesIO,
|
|
content_type: str = "application/octet-stream",
|
|
) -> None:
|
|
pass
|
|
|
|
def delete(self, object_name: str) -> None:
|
|
pass
|
|
|
|
def list_objects(self, prefix: str = "") -> list[str]:
|
|
return []
|
|
|
|
|
|
class FakeConnection:
|
|
"""Plain class that satisfies ConnectionAwareInterface without inheritance."""
|
|
|
|
def connect(self) -> None:
|
|
pass
|
|
|
|
def disconnect(self) -> None:
|
|
pass
|
|
|
|
def is_connected(self) -> bool:
|
|
return True
|
|
|
|
|
|
class FakeContextManager:
|
|
"""Plain class that satisfies ContextAwareInterface without inheritance."""
|
|
|
|
def __enter__(self) -> FakeContextManager:
|
|
return self
|
|
|
|
def __exit__(
|
|
self, exc_type: type | None, exc_val: object | None, exc_tb: object | None
|
|
) -> None:
|
|
pass
|
|
|
|
|
|
def accepts_json_repo(repo: JsonRepositoryInterface) -> None:
|
|
"""Type-checking hook for JsonRepositoryInterface structural subtyping."""
|
|
repo.get("key")
|
|
|
|
|
|
def accepts_object_repo(repo: ObjectRepositoryInterface) -> None:
|
|
"""Type-checking hook for ObjectRepositoryInterface structural subtyping."""
|
|
repo.get("object")
|
|
|
|
|
|
def accepts_connection_aware(connection: ConnectionAwareInterface) -> None:
|
|
"""Type-checking hook for ConnectionAwareInterface structural subtyping."""
|
|
connection.is_connected()
|
|
|
|
|
|
def accepts_context_aware(context: ContextAwareInterface) -> None:
|
|
"""Type-checking hook for ContextAwareInterface structural subtyping."""
|
|
with context:
|
|
pass
|
|
|
|
|
|
def test_json_repository_structural_subtyping() -> None:
|
|
"""Test that a plain class satisfies JsonRepositoryInterface structurally."""
|
|
repo: JsonRepositoryInterface = InMemoryJsonRepo()
|
|
accepts_json_repo(repo)
|
|
assert isinstance(repo, JsonRepositoryInterface)
|
|
|
|
|
|
def test_object_repository_structural_subtyping() -> None:
|
|
"""Test that a plain class satisfies ObjectRepositoryInterface structurally."""
|
|
repo: ObjectRepositoryInterface = InMemoryObjectRepo()
|
|
accepts_object_repo(repo)
|
|
assert isinstance(repo, ObjectRepositoryInterface)
|
|
|
|
|
|
def test_connection_aware_structural_subtyping() -> None:
|
|
"""Test that a plain class satisfies ConnectionAwareInterface structurally."""
|
|
connection: ConnectionAwareInterface = FakeConnection()
|
|
accepts_connection_aware(connection)
|
|
assert isinstance(connection, ConnectionAwareInterface)
|
|
|
|
|
|
def test_context_aware_structural_subtyping() -> None:
|
|
"""Test that a plain class satisfies ContextAwareInterface structurally."""
|
|
context: ContextAwareInterface = FakeContextManager()
|
|
accepts_context_aware(context)
|
|
assert isinstance(context, ContextAwareInterface)
|