test_base_repository_delete.py
1 """ 2 Unit tests for BaseRepository.delete(). 3 4 All tests use an in-memory SQLite database. The behaviours under test 5 (relationship expiry, cascade walking, with_for_update() invocation) are 6 exercised via SQLAlchemy's ORM API. These tests validate ORM calls and 7 high-level behaviour using SQLite; they do not attempt to cover or assert 8 dialect-specific locking or cascade semantics. 9 10 Two concrete behaviours are tested: 11 12 1. with_for_update() is issued on the SELECT that fetches the row to delete. 13 2. uselist=True relationship collections are expired before session.delete() 14 so SQLAlchemy re-fetches them before walking the cascade, preventing the 15 same-session staleness bug where a child added after the collection was 16 first accessed would be invisible to the cascade and block the DELETE. 17 """ 18 19 import uuid 20 from unittest.mock import patch 21 22 import pytest 23 from pydantic import BaseModel 24 from sqlalchemy import Column, ForeignKey, String, create_engine 25 from sqlalchemy.orm import Session, declarative_base, relationship 26 27 from solace_agent_mesh.shared.database.base_repository import BaseRepository 28 from solace_agent_mesh.shared.exceptions.exceptions import EntityNotFoundError 29 30 # --------------------------------------------------------------------------- 31 # Minimal in-memory models 32 # --------------------------------------------------------------------------- 33 34 _Base = declarative_base() 35 36 37 class _ParentModel(_Base): 38 __tablename__ = "parents" 39 id = Column(String, primary_key=True) 40 # uselist=True — should be expired before cascade 41 children = relationship("_ChildModel", cascade="all, delete-orphan", back_populates="parent") 42 # uselist=False (scalar many-to-one) — should NOT be expired 43 owner_id = Column(String, ForeignKey("owners.id"), nullable=True) 44 owner = relationship("_OwnerModel", back_populates="parents") 45 46 47 class _ChildModel(_Base): 48 __tablename__ = "children" 49 id = Column(String, primary_key=True) 50 parent_id = Column(String, ForeignKey("parents.id", ondelete="CASCADE"), nullable=False) 51 parent = relationship("_ParentModel", back_populates="children") 52 53 54 class _OwnerModel(_Base): 55 __tablename__ = "owners" 56 id = Column(String, primary_key=True) 57 parents = relationship("_ParentModel", back_populates="owner") 58 59 60 class _ParentEntity(BaseModel): 61 id: str 62 63 64 class _ParentRepository(BaseRepository[_ParentModel, _ParentEntity]): 65 @property 66 def entity_name(self) -> str: 67 return "Parent" 68 69 70 # --------------------------------------------------------------------------- 71 # Fixtures 72 # --------------------------------------------------------------------------- 73 74 @pytest.fixture(scope="module") 75 def engine(): 76 e = create_engine("sqlite:///:memory:") 77 _Base.metadata.create_all(e) 78 yield e 79 _Base.metadata.drop_all(e) 80 81 82 @pytest.fixture 83 def db(engine): 84 with Session(engine) as session: 85 yield session 86 session.rollback() 87 88 89 @pytest.fixture 90 def repo(): 91 return _ParentRepository(_ParentModel, _ParentEntity) 92 93 94 def _new_parent(): 95 return _ParentModel(id=str(uuid.uuid4())) 96 97 98 def _new_child(parent_id): 99 return _ChildModel(id=str(uuid.uuid4()), parent_id=parent_id) 100 101 102 def _new_owner(): 103 return _OwnerModel(id=str(uuid.uuid4())) 104 105 106 # --------------------------------------------------------------------------- 107 # Tests 108 # --------------------------------------------------------------------------- 109 110 class TestDeleteNotFound: 111 112 def test_raises_for_nonexistent_id(self, repo, db): 113 """EntityNotFoundError is raised when the row does not exist.""" 114 with pytest.raises(EntityNotFoundError): 115 repo.delete(db, "nonexistent-id") 116 117 def test_raises_after_already_deleted(self, repo, db): 118 """A second delete on the same ID raises EntityNotFoundError.""" 119 parent = _new_parent() 120 db.add(parent) 121 db.flush() 122 parent_id = parent.id 123 124 repo.delete(db, parent_id) 125 db.flush() 126 127 with pytest.raises(EntityNotFoundError): 128 repo.delete(db, parent_id) 129 130 131 class TestDeleteCascade: 132 133 def test_deletes_parent_with_no_children(self, repo, db): 134 """A parent with no children is removed from the session after delete.""" 135 parent = _new_parent() 136 db.add(parent) 137 db.flush() 138 parent_id = parent.id 139 140 repo.delete(db, parent_id) 141 db.flush() 142 143 assert db.get(_ParentModel, parent_id) is None 144 145 def test_deletes_parent_and_its_children(self, repo, db): 146 """Children are cascade-deleted along with the parent.""" 147 parent = _new_parent() 148 child1 = _new_child(parent.id) 149 child2 = _new_child(parent.id) 150 db.add_all([parent, child1, child2]) 151 db.flush() 152 parent_id, child1_id, child2_id = parent.id, child1.id, child2.id 153 154 repo.delete(db, parent_id) 155 db.flush() 156 157 assert db.get(_ParentModel, parent_id) is None 158 assert db.get(_ChildModel, child1_id) is None 159 assert db.get(_ChildModel, child2_id) is None 160 161 def test_deletes_parent_when_child_added_after_collection_was_cached(self, repo, db): 162 """Same-session staleness: child added after collection was first accessed is still deleted. 163 164 This is the core regression test. Without session.expire() on the 165 children collection, the ORM would walk the cached empty list during 166 the cascade and miss the child added afterwards, leaving an orphaned FK 167 row that would block the parent DELETE on databases with strict FK 168 enforcement (MySQL). 169 """ 170 parent = _new_parent() 171 db.add(parent) 172 db.flush() 173 174 # Access the collection now — SQLAlchemy caches an empty list. 175 assert parent.children == [] 176 177 # Add a child in the same session AFTER the collection was cached. 178 child = _new_child(parent.id) 179 db.add(child) 180 db.flush() 181 182 # The cached collection still shows [] — but expire() forces a re-fetch 183 # before the cascade walks it, so the child is included and deleted. 184 repo.delete(db, parent.id) 185 db.flush() 186 187 assert db.get(_ParentModel, parent.id) is None 188 assert db.get(_ChildModel, child.id) is None 189 190 191 class TestDeleteOrmBehaviour: 192 193 def test_with_for_update_is_called_on_fetch_query(self, repo, db): 194 """with_for_update() must be called on the query that fetches the row. 195 196 This is a code-path assertion: the lock is issued as part of the 197 SELECT, not separately. A regression here would mean the row is 198 fetched without a lock before the DELETE. 199 """ 200 parent = _new_parent() 201 db.add(parent) 202 db.flush() 203 204 wfu_called = [] 205 original_query = db.query 206 207 def tracking_query(model): 208 q = original_query(model) 209 210 class _TrackingQuery: 211 def filter(self_, *args, **kwargs): 212 fq = q.filter(*args, **kwargs) 213 214 class _TrackingFilter: 215 def with_for_update(self__): 216 wfu_called.append(True) 217 return fq.with_for_update() 218 219 def first(self__): 220 return fq.first() 221 222 return _TrackingFilter() 223 224 return _TrackingQuery() 225 226 with patch.object(db, "query", side_effect=tracking_query): 227 repo.delete(db, parent.id) 228 229 assert wfu_called, "with_for_update() was not called on the fetch query" 230 231 def test_only_collection_relationships_are_expired(self, repo, db): 232 """Only uselist=True collections are expired; scalar refs are not. 233 234 Expiring scalar (many-to-one / one-to-one) relationships would be 235 unnecessary and could trigger unwanted lazy-loads. Only collections 236 can suffer the staleness problem this fix addresses. 237 """ 238 owner = _new_owner() 239 parent = _new_parent() 240 parent.owner_id = owner.id 241 db.add_all([owner, parent]) 242 db.flush() 243 244 expired_keys = [] 245 original_expire = db.expire 246 247 def tracking_expire(instance, attrs=None): 248 if attrs: 249 expired_keys.extend(attrs) 250 return original_expire(instance, attrs) 251 252 with patch.object(db, "expire", side_effect=tracking_expire): 253 repo.delete(db, parent.id) 254 255 # uselist=True — must be expired so the cascade re-fetches it 256 assert "children" in expired_keys, "children collection was not expired before cascade" 257 # uselist=False — must NOT be expired (scalar ref, not a child collection) 258 assert "owner" not in expired_keys, "scalar relationship 'owner' should not be expired"