Brian Bjarke JensenandCursor
4f58c32dd6
PR Title Check / check-title (pull_request) Successful in 9s
Code Quality Pipeline / code-quality (pull_request) Failing after 53s
Test Python Package / unit-tests (pull_request) Successful in 1m1s
Test Python Package / integration-tests (pull_request) Successful in 1m44s
Test Python Package / coverage-report (pull_request) Successful in 13s
Provide a generic disk-backed FIFO queue with configurable path, retention, and dedup keys so consumers can buffer items across restarts without optional extras. Co-authored-by: Cursor <[email protected]>
101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
"""Unit tests for QueueRepositoryInterface."""
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from python_repositories.interfaces.queue_repository_interface import (
|
|
QueueRepositoryInterface,
|
|
)
|
|
|
|
|
|
class InMemoryQueueRepo:
|
|
"""Plain class that satisfies QueueRepositoryInterface without inheritance."""
|
|
|
|
def enqueue(self, items: list[dict[str, Any]]) -> None:
|
|
pass
|
|
|
|
def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]:
|
|
del max_items
|
|
return []
|
|
|
|
def size(self) -> int:
|
|
return 0
|
|
|
|
def evict_older_than(self, cutoff: datetime) -> int:
|
|
del cutoff
|
|
return 0
|
|
|
|
|
|
def accepts_queue_repo(repo: QueueRepositoryInterface) -> None:
|
|
"""Type-checking hook for QueueRepositoryInterface structural subtyping."""
|
|
repo.size()
|
|
|
|
|
|
def test_instantiation_fails_when_enqueue_not_implemented() -> None:
|
|
class Incomplete(QueueRepositoryInterface):
|
|
def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]:
|
|
return []
|
|
|
|
def size(self) -> int:
|
|
return 0
|
|
|
|
def evict_older_than(self, cutoff: datetime) -> int:
|
|
return 0
|
|
|
|
with pytest.raises(TypeError):
|
|
_ = Incomplete() # type: ignore
|
|
|
|
|
|
def test_instantiation_fails_when_dequeue_batch_not_implemented() -> None:
|
|
class Incomplete(QueueRepositoryInterface):
|
|
def enqueue(self, items: list[dict[str, Any]]) -> None:
|
|
pass
|
|
|
|
def size(self) -> int:
|
|
return 0
|
|
|
|
def evict_older_than(self, cutoff: datetime) -> int:
|
|
return 0
|
|
|
|
with pytest.raises(TypeError):
|
|
_ = Incomplete() # type: ignore
|
|
|
|
|
|
def test_instantiation_fails_when_size_not_implemented() -> None:
|
|
class Incomplete(QueueRepositoryInterface):
|
|
def enqueue(self, items: list[dict[str, Any]]) -> None:
|
|
pass
|
|
|
|
def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]:
|
|
return []
|
|
|
|
def evict_older_than(self, cutoff: datetime) -> int:
|
|
return 0
|
|
|
|
with pytest.raises(TypeError):
|
|
_ = Incomplete() # type: ignore
|
|
|
|
|
|
def test_instantiation_fails_when_evict_older_than_not_implemented() -> None:
|
|
class Incomplete(QueueRepositoryInterface):
|
|
def enqueue(self, items: list[dict[str, Any]]) -> None:
|
|
pass
|
|
|
|
def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]:
|
|
return []
|
|
|
|
def size(self) -> int:
|
|
return 0
|
|
|
|
with pytest.raises(TypeError):
|
|
_ = Incomplete() # type: ignore
|
|
|
|
|
|
def test_structural_subtyping() -> None:
|
|
repo: QueueRepositoryInterface = InMemoryQueueRepo()
|
|
accepts_queue_repo(repo)
|
|
assert isinstance(repo, QueueRepositoryInterface)
|
|
assert repo.evict_older_than(datetime.now(UTC)) == 0
|