""" 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"