Files
wehub-resource-sync 7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:08:55 +08:00

857 lines
28 KiB
Python

"""
Comprehensive tests for local_deep_research/news/core/storage.py
Tests cover:
- BaseStorage abstract interface and generate_id()
- CardStorage abstract interface
- RatingStorage abstract interface
- PreferenceStorage abstract interface
- Concrete subclass instantiation and method dispatch
- Edge cases: empty data, missing fields, duplicate IDs
"""
import uuid
from typing import Any, Dict, List, Optional
import pytest
from local_deep_research.news.core.storage import (
BaseStorage,
CardStorage,
PreferenceStorage,
RatingStorage,
)
# ---------------------------------------------------------------------------
# Helpers: minimal concrete implementations for testing
# ---------------------------------------------------------------------------
class ConcreteBaseStorage(BaseStorage):
"""Minimal concrete implementation of BaseStorage for testing."""
def __init__(self):
self._store: Dict[str, Dict[str, Any]] = {}
def create(self, data: Dict[str, Any]) -> str:
rid = self.generate_id()
self._store[rid] = data
return rid
def get(self, id: str) -> Optional[Dict[str, Any]]:
return self._store.get(id)
def update(self, id: str, data: Dict[str, Any]) -> bool:
if id not in self._store:
return False
self._store[id].update(data)
return True
def delete(self, id: str) -> bool:
if id in self._store:
del self._store[id]
return True
return False
def list(
self,
filters: Optional[Dict[str, Any]] = None,
limit: int = 100,
offset: int = 0,
) -> List[Dict[str, Any]]:
items = list(self._store.values())
if filters:
for key, val in filters.items():
items = [i for i in items if i.get(key) == val]
return items[offset : offset + limit]
class ConcreteCardStorage(CardStorage):
"""Minimal concrete implementation of CardStorage."""
def __init__(self):
self._store: Dict[str, Dict[str, Any]] = {}
self._versions: Dict[str, List[Dict[str, Any]]] = {}
def create(self, data: Dict[str, Any]) -> str:
rid = self.generate_id()
self._store[rid] = {**data, "archived": False, "pinned": False}
self._versions[rid] = []
return rid
def get(self, id: str) -> Optional[Dict[str, Any]]:
return self._store.get(id)
def update(self, id: str, data: Dict[str, Any]) -> bool:
if id not in self._store:
return False
self._store[id].update(data)
return True
def delete(self, id: str) -> bool:
if id in self._store:
del self._store[id]
self._versions.pop(id, None)
return True
return False
def list(
self,
filters: Optional[Dict[str, Any]] = None,
limit: int = 100,
offset: int = 0,
) -> List[Dict[str, Any]]:
items = list(self._store.values())
return items[offset : offset + limit]
def get_by_user(
self, user_id: str, limit: int = 50, offset: int = 0
) -> List[Dict[str, Any]]:
return [v for v in self._store.values() if v.get("user_id") == user_id][
offset : offset + limit
]
def get_latest_version(self, card_id: str) -> Optional[Dict[str, Any]]:
versions = self._versions.get(card_id, [])
return versions[-1] if versions else None
def add_version(self, card_id: str, version_data: Dict[str, Any]) -> str:
if card_id not in self._store:
return ""
vid = self.generate_id()
version_data["version_id"] = vid
self._versions.setdefault(card_id, []).append(version_data)
return vid
def update_latest_info(
self, card_id: str, version_data: Dict[str, Any]
) -> bool:
if card_id not in self._store:
return False
self._store[card_id]["latest_version"] = version_data
return True
def archive_card(self, card_id: str) -> bool:
if card_id not in self._store:
return False
self._store[card_id]["archived"] = True
return True
def pin_card(self, card_id: str, pinned: bool = True) -> bool:
if card_id not in self._store:
return False
self._store[card_id]["pinned"] = pinned
return True
class ConcreteRatingStorage(RatingStorage):
"""Minimal concrete implementation of RatingStorage."""
def __init__(self):
self._store: Dict[str, Dict[str, Any]] = {}
def create(self, data: Dict[str, Any]) -> str:
rid = self.generate_id()
self._store[rid] = data
return rid
def get(self, id: str) -> Optional[Dict[str, Any]]:
return self._store.get(id)
def update(self, id: str, data: Dict[str, Any]) -> bool:
if id not in self._store:
return False
self._store[id].update(data)
return True
def delete(self, id: str) -> bool:
if id in self._store:
del self._store[id]
return True
return False
def list(
self,
filters: Optional[Dict[str, Any]] = None,
limit: int = 100,
offset: int = 0,
) -> List[Dict[str, Any]]:
return list(self._store.values())[offset : offset + limit]
def get_user_rating(
self, user_id: str, item_id: str, rating_type: str
) -> Optional[Dict[str, Any]]:
for v in self._store.values():
if (
v.get("user_id") == user_id
and v.get("item_id") == item_id
and v.get("rating_type") == rating_type
):
return v
return None
def upsert_rating(
self,
user_id: str,
item_id: str,
rating_type: str,
rating_value: str,
item_type: str = "card",
) -> str:
for rid, v in self._store.items():
if (
v.get("user_id") == user_id
and v.get("item_id") == item_id
and v.get("rating_type") == rating_type
):
v["rating_value"] = rating_value
return rid
return self.create(
{
"user_id": user_id,
"item_id": item_id,
"rating_type": rating_type,
"rating_value": rating_value,
"item_type": item_type,
}
)
def get_ratings_summary(
self, item_id: str, item_type: str = "card"
) -> Dict[str, Any]:
ratings = [
v
for v in self._store.values()
if v.get("item_id") == item_id and v.get("item_type") == item_type
]
return {"item_id": item_id, "count": len(ratings), "ratings": ratings}
def get_user_ratings(
self, user_id: str, rating_type: Optional[str] = None, limit: int = 100
) -> List[Dict[str, Any]]:
result = [
v for v in self._store.values() if v.get("user_id") == user_id
]
if rating_type:
result = [r for r in result if r.get("rating_type") == rating_type]
return result[:limit]
class ConcretePreferenceStorage(PreferenceStorage):
"""Minimal concrete implementation of PreferenceStorage."""
def __init__(self):
self._store: Dict[str, Dict[str, Any]] = {}
self._user_prefs: Dict[str, Dict[str, Any]] = {}
def create(self, data: Dict[str, Any]) -> str:
rid = self.generate_id()
self._store[rid] = data
return rid
def get(self, id: str) -> Optional[Dict[str, Any]]:
return self._store.get(id)
def update(self, id: str, data: Dict[str, Any]) -> bool:
if id not in self._store:
return False
self._store[id].update(data)
return True
def delete(self, id: str) -> bool:
if id in self._store:
del self._store[id]
return True
return False
def list(
self,
filters: Optional[Dict[str, Any]] = None,
limit: int = 100,
offset: int = 0,
) -> List[Dict[str, Any]]:
return list(self._store.values())[offset : offset + limit]
def get_user_preferences(self, user_id: str) -> Optional[Dict[str, Any]]:
return self._user_prefs.get(user_id)
def upsert_preferences(
self, user_id: str, preferences: Dict[str, Any]
) -> str:
if user_id in self._user_prefs:
self._user_prefs[user_id].update(preferences)
# Find existing id
for rid, v in self._store.items():
if v.get("user_id") == user_id:
return rid
return ""
rid = self.generate_id()
self._user_prefs[user_id] = {**preferences, "user_id": user_id}
self._store[rid] = self._user_prefs[user_id]
return rid
def add_liked_item(
self, user_id: str, item_id: str, item_type: str = "news"
) -> bool:
prefs = self._user_prefs.setdefault(
user_id, {"liked": [], "disliked": []}
)
prefs.setdefault("liked", []).append(
{"item_id": item_id, "item_type": item_type}
)
return True
def add_disliked_item(
self, user_id: str, item_id: str, item_type: str = "news"
) -> bool:
prefs = self._user_prefs.setdefault(
user_id, {"liked": [], "disliked": []}
)
prefs.setdefault("disliked", []).append(
{"item_id": item_id, "item_type": item_type}
)
return True
def update_preference_embedding(
self, user_id: str, embedding: List[float]
) -> bool:
if user_id not in self._user_prefs:
return False
self._user_prefs[user_id]["embedding"] = embedding
return True
# ---------------------------------------------------------------------------
# Test: cannot instantiate abstract classes directly
# ---------------------------------------------------------------------------
class TestAbstractInstantiation:
"""Verify that abstract classes cannot be instantiated directly."""
def test_base_storage_cannot_be_instantiated(self):
with pytest.raises(TypeError):
BaseStorage()
def test_card_storage_cannot_be_instantiated(self):
with pytest.raises(TypeError):
CardStorage()
def test_rating_storage_cannot_be_instantiated(self):
with pytest.raises(TypeError):
RatingStorage()
def test_preference_storage_cannot_be_instantiated(self):
with pytest.raises(TypeError):
PreferenceStorage()
class TestPartialAbstractImplementation:
"""Verify that partial implementations still raise TypeError."""
def test_base_storage_missing_one_method(self):
"""Omitting 'list' should prevent instantiation."""
class Incomplete(BaseStorage):
def create(self, data):
pass
def get(self, id):
pass
def update(self, id, data):
pass
def delete(self, id):
pass
# list intentionally missing
with pytest.raises(TypeError):
Incomplete()
def test_card_storage_missing_card_method(self):
"""CardStorage requires all BaseStorage + card-specific methods."""
class IncompleteCard(CardStorage):
def create(self, data):
pass
def get(self, id):
pass
def update(self, id, data):
pass
def delete(self, id):
pass
def list(self, filters=None, limit=100, offset=0):
pass
# Missing get_by_user, get_latest_version, add_version, etc.
with pytest.raises(TypeError):
IncompleteCard()
# ---------------------------------------------------------------------------
# Test: generate_id (the only concrete method on BaseStorage)
# ---------------------------------------------------------------------------
class TestGenerateId:
"""Test BaseStorage.generate_id via concrete subclass."""
def test_generate_id_returns_string(self):
storage = ConcreteBaseStorage()
generated = storage.generate_id()
assert isinstance(generated, str)
def test_generate_id_is_valid_uuid4(self):
storage = ConcreteBaseStorage()
generated = storage.generate_id()
parsed = uuid.UUID(generated, version=4)
assert str(parsed) == generated
def test_generate_id_uniqueness(self):
storage = ConcreteBaseStorage()
ids = {storage.generate_id() for _ in range(200)}
assert len(ids) == 200
def test_generate_id_available_on_all_subclasses(self):
"""Every storage subclass inherits generate_id."""
for cls in [
ConcreteCardStorage,
ConcreteRatingStorage,
ConcretePreferenceStorage,
]:
instance = cls()
gid = instance.generate_id()
assert isinstance(gid, str)
uuid.UUID(gid, version=4) # validates format
# ---------------------------------------------------------------------------
# Test: BaseStorage CRUD via ConcreteBaseStorage
# ---------------------------------------------------------------------------
class TestBaseStorageCRUD:
@pytest.fixture
def storage(self):
return ConcreteBaseStorage()
def test_create_and_get(self, storage):
rid = storage.create({"name": "test"})
assert isinstance(rid, str)
record = storage.get(rid)
assert record == {"name": "test"}
def test_get_nonexistent(self, storage):
assert storage.get("nonexistent-id") is None
def test_update_existing(self, storage):
rid = storage.create({"name": "old"})
assert storage.update(rid, {"name": "new"}) is True
assert storage.get(rid)["name"] == "new"
def test_update_nonexistent(self, storage):
assert storage.update("nonexistent", {"x": 1}) is False
def test_delete_existing(self, storage):
rid = storage.create({"a": 1})
assert storage.delete(rid) is True
assert storage.get(rid) is None
def test_delete_nonexistent(self, storage):
assert storage.delete("nonexistent") is False
def test_list_empty(self, storage):
assert storage.list() == []
def test_list_with_records(self, storage):
storage.create({"x": 1})
storage.create({"x": 2})
assert len(storage.list()) == 2
def test_list_with_filters(self, storage):
storage.create({"color": "red"})
storage.create({"color": "blue"})
assert len(storage.list(filters={"color": "red"})) == 1
def test_list_limit_and_offset(self, storage):
for i in range(10):
storage.create({"idx": i})
result = storage.list(limit=3, offset=2)
assert len(result) == 3
def test_create_empty_data(self, storage):
rid = storage.create({})
assert storage.get(rid) == {}
def test_update_with_empty_data(self, storage):
rid = storage.create({"a": 1})
assert storage.update(rid, {}) is True
assert storage.get(rid) == {"a": 1}
def test_delete_twice(self, storage):
rid = storage.create({"a": 1})
assert storage.delete(rid) is True
assert storage.delete(rid) is False
# ---------------------------------------------------------------------------
# Test: CardStorage
# ---------------------------------------------------------------------------
class TestCardStorage:
@pytest.fixture
def storage(self):
return ConcreteCardStorage()
def test_create_card(self, storage):
cid = storage.create({"user_id": "u1", "title": "Card 1"})
card = storage.get(cid)
assert card["user_id"] == "u1"
assert card["archived"] is False
assert card["pinned"] is False
def test_get_by_user(self, storage):
storage.create({"user_id": "u1", "title": "C1"})
storage.create({"user_id": "u2", "title": "C2"})
storage.create({"user_id": "u1", "title": "C3"})
result = storage.get_by_user("u1")
assert len(result) == 2
def test_get_by_user_empty(self, storage):
assert storage.get_by_user("nobody") == []
def test_get_by_user_limit_offset(self, storage):
for i in range(5):
storage.create({"user_id": "u1", "title": f"C{i}"})
result = storage.get_by_user("u1", limit=2, offset=1)
assert len(result) == 2
def test_get_latest_version_no_versions(self, storage):
cid = storage.create({"title": "empty"})
assert storage.get_latest_version(cid) is None
def test_add_and_get_latest_version(self, storage):
cid = storage.create({"title": "card"})
vid1 = storage.add_version(cid, {"content": "v1"})
vid2 = storage.add_version(cid, {"content": "v2"})
latest = storage.get_latest_version(cid)
assert latest["content"] == "v2"
assert vid1 != vid2
def test_add_version_nonexistent_card(self, storage):
result = storage.add_version("nonexistent", {"content": "x"})
assert result == ""
def test_update_latest_info(self, storage):
cid = storage.create({"title": "card"})
assert storage.update_latest_info(cid, {"summary": "s"}) is True
card = storage.get(cid)
assert card["latest_version"] == {"summary": "s"}
def test_update_latest_info_nonexistent(self, storage):
assert storage.update_latest_info("no-card", {}) is False
def test_archive_card(self, storage):
cid = storage.create({"title": "card"})
assert storage.archive_card(cid) is True
assert storage.get(cid)["archived"] is True
def test_archive_nonexistent_card(self, storage):
assert storage.archive_card("missing") is False
def test_pin_card(self, storage):
cid = storage.create({"title": "card"})
assert storage.pin_card(cid) is True
assert storage.get(cid)["pinned"] is True
def test_unpin_card(self, storage):
cid = storage.create({"title": "card"})
storage.pin_card(cid, pinned=True)
storage.pin_card(cid, pinned=False)
assert storage.get(cid)["pinned"] is False
def test_pin_nonexistent_card(self, storage):
assert storage.pin_card("missing") is False
def test_delete_card_removes_versions(self, storage):
cid = storage.create({"title": "card"})
storage.add_version(cid, {"v": 1})
storage.delete(cid)
assert storage.get(cid) is None
assert storage.get_latest_version(cid) is None
# ---------------------------------------------------------------------------
# Test: RatingStorage
# ---------------------------------------------------------------------------
class TestRatingStorage:
@pytest.fixture
def storage(self):
return ConcreteRatingStorage()
def test_upsert_creates_rating(self, storage):
rid = storage.upsert_rating("u1", "item1", "quality", "good")
assert isinstance(rid, str)
rating = storage.get_user_rating("u1", "item1", "quality")
assert rating["rating_value"] == "good"
def test_upsert_updates_existing(self, storage):
rid1 = storage.upsert_rating("u1", "item1", "quality", "good")
rid2 = storage.upsert_rating("u1", "item1", "quality", "excellent")
assert rid1 == rid2
rating = storage.get_user_rating("u1", "item1", "quality")
assert rating["rating_value"] == "excellent"
def test_get_user_rating_not_found(self, storage):
assert storage.get_user_rating("u1", "item99", "quality") is None
def test_get_ratings_summary_empty(self, storage):
summary = storage.get_ratings_summary("item99")
assert summary["count"] == 0
assert summary["ratings"] == []
def test_get_ratings_summary_with_data(self, storage):
storage.upsert_rating("u1", "item1", "quality", "good")
storage.upsert_rating("u2", "item1", "quality", "bad")
summary = storage.get_ratings_summary("item1")
assert summary["count"] == 2
def test_get_user_ratings(self, storage):
storage.upsert_rating("u1", "item1", "quality", "good")
storage.upsert_rating("u1", "item2", "relevance", "high")
storage.upsert_rating("u2", "item1", "quality", "bad")
result = storage.get_user_ratings("u1")
assert len(result) == 2
def test_get_user_ratings_filtered_by_type(self, storage):
storage.upsert_rating("u1", "item1", "quality", "good")
storage.upsert_rating("u1", "item2", "relevance", "high")
result = storage.get_user_ratings("u1", rating_type="quality")
assert len(result) == 1
def test_get_user_ratings_empty(self, storage):
assert storage.get_user_ratings("nobody") == []
def test_upsert_with_custom_item_type(self, storage):
storage.upsert_rating(
"u1", "item1", "quality", "good", item_type="article"
)
rating = storage.get_user_rating("u1", "item1", "quality")
assert rating["item_type"] == "article"
def test_get_ratings_summary_filters_by_item_type(self, storage):
storage.upsert_rating("u1", "i1", "q", "good", item_type="card")
storage.upsert_rating("u2", "i1", "q", "bad", item_type="article")
summary = storage.get_ratings_summary("i1", item_type="card")
assert summary["count"] == 1
# ---------------------------------------------------------------------------
# Test: PreferenceStorage
# ---------------------------------------------------------------------------
class TestPreferenceStorage:
@pytest.fixture
def storage(self):
return ConcretePreferenceStorage()
def test_get_user_preferences_none(self, storage):
assert storage.get_user_preferences("u1") is None
def test_upsert_creates_preferences(self, storage):
rid = storage.upsert_preferences("u1", {"theme": "dark"})
assert isinstance(rid, str)
prefs = storage.get_user_preferences("u1")
assert prefs["theme"] == "dark"
def test_upsert_updates_preferences(self, storage):
storage.upsert_preferences("u1", {"theme": "dark"})
storage.upsert_preferences("u1", {"theme": "light"})
prefs = storage.get_user_preferences("u1")
assert prefs["theme"] == "light"
def test_add_liked_item(self, storage):
assert storage.add_liked_item("u1", "item1") is True
prefs = storage.get_user_preferences("u1")
assert any(
item["item_id"] == "item1" for item in prefs.get("liked", [])
)
def test_add_disliked_item(self, storage):
assert storage.add_disliked_item("u1", "item1") is True
prefs = storage.get_user_preferences("u1")
assert any(
item["item_id"] == "item1" for item in prefs.get("disliked", [])
)
def test_add_liked_item_custom_type(self, storage):
storage.add_liked_item("u1", "item1", item_type="article")
prefs = storage.get_user_preferences("u1")
liked = prefs["liked"]
assert liked[0]["item_type"] == "article"
def test_update_preference_embedding(self, storage):
storage.upsert_preferences("u1", {"theme": "dark"})
emb = [0.1, 0.2, 0.3]
assert storage.update_preference_embedding("u1", emb) is True
prefs = storage.get_user_preferences("u1")
assert prefs["embedding"] == emb
def test_update_preference_embedding_no_user(self, storage):
assert storage.update_preference_embedding("nobody", [0.1]) is False
def test_update_preference_embedding_empty(self, storage):
storage.upsert_preferences("u1", {})
assert storage.update_preference_embedding("u1", []) is True
assert storage.get_user_preferences("u1")["embedding"] == []
# ---------------------------------------------------------------------------
# Test: Inheritance hierarchy
# ---------------------------------------------------------------------------
class TestInheritanceHierarchy:
def test_card_storage_is_base_storage(self):
assert issubclass(CardStorage, BaseStorage)
def test_rating_storage_is_base_storage(self):
assert issubclass(RatingStorage, BaseStorage)
def test_preference_storage_is_base_storage(self):
assert issubclass(PreferenceStorage, BaseStorage)
def test_concrete_card_isinstance(self):
s = ConcreteCardStorage()
assert isinstance(s, BaseStorage)
assert isinstance(s, CardStorage)
# ---------------------------------------------------------------------------
# Test: Abstract method signatures
# ---------------------------------------------------------------------------
class TestAbstractMethodSignatures:
"""Ensure the abstract classes define the expected methods."""
def test_base_storage_abstract_methods(self):
expected = {"create", "get", "update", "delete", "list"}
actual = set(BaseStorage.__abstractmethods__)
assert expected == actual
def test_card_storage_abstract_methods(self):
expected = {
"create",
"get",
"update",
"delete",
"list",
"get_by_user",
"get_latest_version",
"add_version",
"update_latest_info",
"archive_card",
"pin_card",
}
actual = set(CardStorage.__abstractmethods__)
assert expected == actual
def test_rating_storage_abstract_methods(self):
expected = {
"create",
"get",
"update",
"delete",
"list",
"get_user_rating",
"upsert_rating",
"get_ratings_summary",
"get_user_ratings",
}
actual = set(RatingStorage.__abstractmethods__)
assert expected == actual
def test_preference_storage_abstract_methods(self):
expected = {
"create",
"get",
"update",
"delete",
"list",
"get_user_preferences",
"upsert_preferences",
"add_liked_item",
"add_disliked_item",
"update_preference_embedding",
}
actual = set(PreferenceStorage.__abstractmethods__)
assert expected == actual
# ---------------------------------------------------------------------------
# Test: Default parameters
# ---------------------------------------------------------------------------
class TestDefaultParameters:
"""Ensure default parameter values work correctly through implementations."""
def test_list_defaults(self):
storage = ConcreteBaseStorage()
for i in range(150):
storage.create({"idx": i})
# default limit=100, offset=0
result = storage.list()
assert len(result) == 100
def test_get_by_user_defaults(self):
storage = ConcreteCardStorage()
for i in range(60):
storage.create({"user_id": "u1", "title": f"C{i}"})
# default limit=50, offset=0
result = storage.get_by_user("u1")
assert len(result) == 50
def test_pin_card_default_pinned_true(self):
storage = ConcreteCardStorage()
cid = storage.create({"title": "card"})
storage.pin_card(cid) # default pinned=True
assert storage.get(cid)["pinned"] is True
def test_upsert_rating_default_item_type(self):
storage = ConcreteRatingStorage()
storage.upsert_rating("u1", "i1", "q", "good")
rating = storage.get_user_rating("u1", "i1", "q")
assert rating["item_type"] == "card"
def test_get_ratings_summary_default_item_type(self):
storage = ConcreteRatingStorage()
storage.upsert_rating("u1", "i1", "q", "good", item_type="card")
summary = storage.get_ratings_summary("i1")
assert summary["count"] == 1
def test_add_liked_item_default_type(self):
storage = ConcretePreferenceStorage()
storage.add_liked_item("u1", "item1")
prefs = storage.get_user_preferences("u1")
assert prefs["liked"][0]["item_type"] == "news"
def test_add_disliked_item_default_type(self):
storage = ConcretePreferenceStorage()
storage.add_disliked_item("u1", "item1")
prefs = storage.get_user_preferences("u1")
assert prefs["disliked"][0]["item_type"] == "news"