|
| 1 | +"""Test MemoryCaseDao.upsert() with a real SQLite database — verify that inserting |
| 2 | +new records does NOT delete existing ones. |
| 3 | +""" |
| 4 | + |
| 5 | +import pytest |
| 6 | +from sqlalchemy import create_engine |
| 7 | + |
| 8 | +from derisk.storage.metadata.db_manager import db as global_db |
| 9 | +from derisk.storage.metadata import Model as GlobalModel |
| 10 | + |
| 11 | +from derisk_ext.plugin.memory_case.models import CandidateCase, CandidateCaseLifecycle |
| 12 | +from derisk_ext.plugin.memory_case.sqlalchemy_dao import MemoryCaseDao, MemoryCaseEntity |
| 13 | + |
| 14 | + |
| 15 | +@pytest.fixture |
| 16 | +def dao(): |
| 17 | + """Create a MemoryCaseDao backed by a throwaway SQLite in-memory database. |
| 18 | +
|
| 19 | + Since MemoryCaseEntity inherits from the global db.Model, we must |
| 20 | + temporarily repoint the global db to a test engine and create tables there. |
| 21 | + """ |
| 22 | + old_engine = global_db._engine |
| 23 | + old_session = global_db._session |
| 24 | + |
| 25 | + test_engine = create_engine("sqlite:///:memory:") |
| 26 | + global_db._engine = test_engine |
| 27 | + global_db._session = None # force re-init below |
| 28 | + |
| 29 | + from sqlalchemy.orm import sessionmaker, Session |
| 30 | + from derisk.storage.metadata.db_manager import BaseQuery |
| 31 | + session_factory = sessionmaker( |
| 32 | + bind=test_engine, class_=Session, query_cls=BaseQuery |
| 33 | + ) |
| 34 | + global_db._session = session_factory |
| 35 | + |
| 36 | + # Create the table on the test engine |
| 37 | + GlobalModel.metadata.create_all(test_engine) |
| 38 | + |
| 39 | + try: |
| 40 | + yield MemoryCaseDao() |
| 41 | + finally: |
| 42 | + global_db._engine = old_engine |
| 43 | + global_db._session = old_session |
| 44 | + test_engine.dispose() |
| 45 | + |
| 46 | + |
| 47 | +def _make_case(case_id: str, symptom: str, confidence: float = 0.5) -> CandidateCase: |
| 48 | + return CandidateCase( |
| 49 | + case_id=case_id, |
| 50 | + symptom_summary=symptom, |
| 51 | + confidence=confidence, |
| 52 | + ) |
| 53 | + |
| 54 | + |
| 55 | +def test_upsert_five_records_keeps_previous_four(dao): |
| 56 | + # --- insert 4 records --- |
| 57 | + for i in range(1, 5): |
| 58 | + case = _make_case(case_id=f"case-{i}", symptom=f"issue #{i}", confidence=0.6) |
| 59 | + saved = dao.upsert(case) |
| 60 | + assert saved.case_id == f"case-{i}" |
| 61 | + |
| 62 | + # verify 4 exist |
| 63 | + assert dao.get_by_case_id("case-1") is not None |
| 64 | + assert dao.get_by_case_id("case-2") is not None |
| 65 | + assert dao.get_by_case_id("case-3") is not None |
| 66 | + assert dao.get_by_case_id("case-4") is not None |
| 67 | + |
| 68 | + # --- insert 5th record --- |
| 69 | + case5 = _make_case(case_id="case-5", symptom="new issue #5", confidence=0.7) |
| 70 | + dao.upsert(case5) |
| 71 | + |
| 72 | + # verify all 5 still exist (old 4 are NOT deleted) |
| 73 | + for i in range(1, 6): |
| 74 | + found = dao.get_by_case_id(f"case-{i}") |
| 75 | + assert found is not None, f"case-{i} should still exist after inserting case-5" |
| 76 | + |
| 77 | + # also spot-check field values of an old record |
| 78 | + c1 = dao.get_by_case_id("case-1") |
| 79 | + assert c1.symptom_summary == "issue #1" |
| 80 | + assert c1.confidence == 0.6 |
| 81 | + |
| 82 | + |
| 83 | +def test_upsert_existing_record_does_not_delete_others(dao): |
| 84 | + # same as above but simulate a merge (same case_id) |
| 85 | + for i in range(1, 4): |
| 86 | + dao.upsert(_make_case(case_id=f"case-{i}", symptom=f"old #{i}")) |
| 87 | + |
| 88 | + # merge case-2: update resolution only |
| 89 | + merged = CandidateCase( |
| 90 | + case_id="case-2", |
| 91 | + symptom_summary="", # should NOT overwrite |
| 92 | + resolution="merged resolution", |
| 93 | + confidence=0.8, |
| 94 | + ) |
| 95 | + dao.upsert(merged) |
| 96 | + |
| 97 | + # old records still exist |
| 98 | + assert dao.get_by_case_id("case-1") is not None |
| 99 | + assert dao.get_by_case_id("case-3") is not None |
| 100 | + |
| 101 | + # merged record: symptom preserved, resolution updated |
| 102 | + c2 = dao.get_by_case_id("case-2") |
| 103 | + assert c2.symptom_summary == "old #2", "existing symptom should be preserved" |
| 104 | + assert c2.resolution == "merged resolution", "new resolution should be set" |
| 105 | + assert c2.confidence == 0.8, "confidence should be updated" |
| 106 | + |
| 107 | + |
| 108 | +def test_upsert_concurrent_like_scenario(dao): |
| 109 | + """Simulate the Agent writing many cases in rapid succession with different metadata.""" |
| 110 | + import json |
| 111 | + |
| 112 | + base_data = [ |
| 113 | + {"case_id": "case-a", "symptom": "CPU 飙升", "metadata": {"case_context": {"app_code": "app1", "environment": "prod"}}}, |
| 114 | + {"case_id": "case-b", "symptom": "OOM Kill", "metadata": {"case_context": {"app_code": "app1", "environment": "prod"}}}, |
| 115 | + {"case_id": "case-c", "symptom": "Disk full", "metadata": {"case_context": {"app_code": "app2", "environment": "prod"}}}, |
| 116 | + {"case_id": "case-d", "symptom": "Latency spike", "metadata": {"case_context": {"app_code": "app1", "environment": "staging"}}}, |
| 117 | + ] |
| 118 | + |
| 119 | + for item in base_data: |
| 120 | + case = CandidateCase(**item) |
| 121 | + dao.upsert(case) |
| 122 | + |
| 123 | + # verify 4 records |
| 124 | + results = dao.search(scope={"app_code": "default"}, limit=50) |
| 125 | + assert len(results) == 4, f"Expected 4 records before new insert, got {len(results)}" |
| 126 | + |
| 127 | + # insert 5th |
| 128 | + new_case = CandidateCase( |
| 129 | + case_id="case-e", |
| 130 | + symptom_summary="Network timeout", |
| 131 | + metadata={"case_context": {"app_code": "app1", "environment": "prod"}}, |
| 132 | + ) |
| 133 | + dao.upsert(new_case) |
| 134 | + |
| 135 | + # verify all 5 still exist |
| 136 | + results_after = dao.search(scope={"app_code": "default"}, limit=50) |
| 137 | + assert len(results_after) == 5, ( |
| 138 | + f"Expected 5 records after insert, got {len(results_after)}. " |
| 139 | + f"Found case_ids: {[r.case_id for r in results_after]}" |
| 140 | + ) |
| 141 | + |
| 142 | + # verify individual lookup |
| 143 | + for cid in ["case-a", "case-b", "case-c", "case-d", "case-e"]: |
| 144 | + assert dao.get_by_case_id(cid) is not None, f"{cid} should exist" |
| 145 | + |
| 146 | + |
| 147 | +def test_stress_many_upserts_never_delete_old_records(dao): |
| 148 | + """Stress-test: insert 20 records one by one, each time verify all previous exist.""" |
| 149 | + ids_inserted = [] |
| 150 | + for i in range(20): |
| 151 | + cid = f"stress-{i}" |
| 152 | + dao.upsert(_make_case(case_id=cid, symptom=f"stress symptom {i}")) |
| 153 | + ids_inserted.append(cid) |
| 154 | + # verify all previously inserted still exist |
| 155 | + for prev_cid in ids_inserted: |
| 156 | + assert dao.get_by_case_id(prev_cid) is not None, ( |
| 157 | + f"{prev_cid} disappeared after inserting {cid}" |
| 158 | + ) |
| 159 | + # final verification: all 20 exist |
| 160 | + results = dao.search(scope={"app_code": "default"}, limit=100) |
| 161 | + assert len(results) == 20, f"Expected 20, got {len(results)}" |
| 162 | + |
| 163 | + |
| 164 | +def test_full_service_upsert_flow_with_real_dao(dao): |
| 165 | + """Exercise the complete service._upsert → dao path (no vector store).""" |
| 166 | + from derisk_ext.plugin.memory_case.service import MemoryCasePluginService |
| 167 | + |
| 168 | + class _DummySystemApp: |
| 169 | + config = {} |
| 170 | + |
| 171 | + class _FakeVector: |
| 172 | + async def upsert(self, case): pass |
| 173 | + async def search(self, query, scope, top_k): return [] |
| 174 | + async def invalidate(self, case_id): pass |
| 175 | + |
| 176 | + service = MemoryCasePluginService( |
| 177 | + system_app=_DummySystemApp(), |
| 178 | + dao=dao, |
| 179 | + vector_index=_FakeVector(), |
| 180 | + ) |
| 181 | + |
| 182 | + # Write 4 cases through the service (same path the Agent uses) |
| 183 | + import asyncio |
| 184 | + for i in range(1, 5): |
| 185 | + result = asyncio.get_event_loop().run_until_complete( |
| 186 | + service.call_tool("memory_case_upsert", { |
| 187 | + "case": { |
| 188 | + "case_id": f"svc-{i}", |
| 189 | + "symptom_summary": f"service issue {i}", |
| 190 | + "resolution": f"fix {i}", |
| 191 | + "metadata": {"case_context": {"app_code": "demo", "environment": "prod"}}, |
| 192 | + } |
| 193 | + }) |
| 194 | + ) |
| 195 | + assert result["code"] == "OK" |
| 196 | + assert result["case"]["case_id"] == f"svc-{i}" |
| 197 | + |
| 198 | + # Verify 4 exist |
| 199 | + results = dao.search(scope={"app_code": "default"}, limit=50) |
| 200 | + assert len(results) == 4, f"Expected 4, got {len(results)}" |
| 201 | + |
| 202 | + # Write 5th |
| 203 | + result = asyncio.get_event_loop().run_until_complete( |
| 204 | + service.call_tool("memory_case_upsert", { |
| 205 | + "case": { |
| 206 | + "case_id": "svc-5", |
| 207 | + "symptom_summary": "new service issue", |
| 208 | + "metadata": {"case_context": {"app_code": "demo", "environment": "prod"}}, |
| 209 | + } |
| 210 | + }) |
| 211 | + ) |
| 212 | + assert result["code"] == "OK" |
| 213 | + |
| 214 | + # Verify ALL 5 still exist |
| 215 | + results_after = dao.search(scope={"app_code": "default"}, limit=50) |
| 216 | + assert len(results_after) == 5, ( |
| 217 | + f"CRITICAL: Expected 5 records, got {len(results_after)}. " |
| 218 | + f"Found: {[r.case_id for r in results_after]}" |
| 219 | + ) |
| 220 | + for i in range(1, 6): |
| 221 | + found = dao.get_by_case_id(f"svc-{i}") |
| 222 | + assert found is not None, f"svc-{i} disappeared after 5th insert" |
| 223 | + |
| 224 | + # Spot-check: old record field values preserved |
| 225 | + c1 = dao.get_by_case_id("svc-1") |
| 226 | + assert c1.symptom_summary == "service issue 1" |
| 227 | + assert c1.resolution == "fix 1" |
| 228 | + |
| 229 | + |
| 230 | +def test_search_with_narrow_scope_does_not_mistake_scoping_for_deletion(dao): |
| 231 | + """Scope filters narrow results — this is expected, not data loss.""" |
| 232 | + cases = [ |
| 233 | + CandidateCase(case_id="c1", symptom_summary="A", metadata={"case_context": {"app_code": "team-x"}}), |
| 234 | + CandidateCase(case_id="c2", symptom_summary="B", metadata={"case_context": {"app_code": "team-x"}}), |
| 235 | + CandidateCase(case_id="c3", symptom_summary="C", metadata={"case_context": {"app_code": "team-y"}}), |
| 236 | + ] |
| 237 | + for c in cases: |
| 238 | + dao.upsert(c) |
| 239 | + |
| 240 | + # search with team-x scope → returns 2 |
| 241 | + r_x = dao.search(scope={"app_code": "team-x"}, limit=10) |
| 242 | + assert len(r_x) == 2, f"team-x should see 2 cases, got {len(r_x)}" |
| 243 | + |
| 244 | + # search with team-y scope → returns 1 |
| 245 | + r_y = dao.search(scope={"app_code": "team-y"}, limit=10) |
| 246 | + assert len(r_y) == 1 |
| 247 | + |
| 248 | + # search with default scope → returns ALL 3 |
| 249 | + r_all = dao.search(scope={"app_code": "default"}, limit=10) |
| 250 | + assert len(r_all) == 3, f"default scope should see all 3, got {len(r_all)}" |
0 commit comments