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
857 lines
28 KiB
Python
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"
|