| 1 | """ |
| 2 | Tests for ingest.watcher - ADDED / UPDATED / REMOVED classification. |
| 3 | |
| 4 | All tests are purely synthetic (no network calls, no subprocess). |
| 5 | They exercise: |
| 6 | - _classify_added_updated() : seen-table diff logic |
| 7 | - _detect_full_set_changes() : manifest-diff + removal guards |
| 8 | - Guard: non-empty pull (prevents false removals on empty result) |
| 9 | - Guard: first-run (no prior manifest → no removals) |
| 10 | - source_url non-null invariant on every change_event |
| 11 | """ |
| 12 | |
| 13 | import os |
| 14 | import tempfile |
| 15 | |
| 16 | import pytest |
| 17 | |
| 18 | from ingest.db import get_connection, init_schema |
| 19 | from ingest.watcher import ( |
| 20 | FULL_SET_CONFIGS, |
| 21 | _classify_added_updated, |
| 22 | _detect_full_set_changes, |
| 23 | _ensure_watcher_tables, |
| 24 | _load_manifest, |
| 25 | _snapshot_seen, |
| 26 | ) |
| 27 | |
| 28 | |
| 29 | |
| 30 | def _make_db(): |
| 31 | """Return (conn, tmp_path) for a fresh in-memory-equivalent temp DB.""" |
| 32 | tmp = tempfile.mktemp(suffix=".db") |
| 33 | conn = get_connection(tmp) |
| 34 | init_schema(conn) |
| 35 | _ensure_watcher_tables(conn) |
| 36 | return conn, tmp |
| 37 | |
| 38 | |
| 39 | def _cleanup(conn, tmp): |
| 40 | conn.close() |
| 41 | try: |
| 42 | os.unlink(tmp) |
| 43 | except OSError: |
| 44 | pass |
| 45 | |
| 46 | |
| 47 | def _seed_seen(conn, rows): |
| 48 | """Seed seen table with list of (source, external_id, update_date).""" |
| 49 | for source, ext_id, update_date in rows: |
| 50 | conn.execute( |
| 51 | "INSERT OR REPLACE INTO seen (source, external_id, update_date)" |
| 52 | " VALUES (?,?,?)", |
| 53 | (source, ext_id, update_date), |
| 54 | ) |
| 55 | conn.commit() |
| 56 | |
| 57 | |
| 58 | def _seed_members_table(conn, members): |
| 59 | """Create and populate the members table with minimal columns.""" |
| 60 | conn.executescript( |
| 61 | """ |
| 62 | CREATE TABLE IF NOT EXISTS members ( |
| 63 | id INTEGER PRIMARY KEY AUTOINCREMENT, |
| 64 | bioguide_id TEXT NOT NULL UNIQUE, |
| 65 | full_name TEXT NOT NULL DEFAULT '', |
| 66 | party TEXT, state TEXT, district INTEGER, chamber TEXT, |
| 67 | lis_member_id TEXT, govtrack_id INTEGER, icpsr_id INTEGER, |
| 68 | source_url TEXT NOT NULL |
| 69 | ); |
| 70 | """ |
| 71 | ) |
| 72 | for m in members: |
| 73 | bio = m["bioguide_id"] |
| 74 | conn.execute( |
| 75 | "INSERT OR REPLACE INTO members (bioguide_id, full_name, source_url)" |
| 76 | " VALUES (?,?,?)", |
| 77 | (bio, m.get("full_name", "Test Member"), |
| 78 | m.get("source_url", f"https://bioguide.congress.gov/search/bio/{bio}")), |
| 79 | ) |
| 80 | conn.commit() |
| 81 | |
| 82 | |
| 83 | TS = "2026-06-26T00:00:00+00:00" |
| 84 | |
| 85 | |
| 86 | |
| 87 | class TestClassifyAddedUpdated: |
| 88 | |
| 89 | def test_added_single_new_entry(self): |
| 90 | conn, tmp = _make_db() |
| 91 | before = {} |
| 92 | after = {("federal_register", "2026-12345"): "2026-06-25"} |
| 93 | added, updated = _classify_added_updated(before, after, TS, conn) |
| 94 | assert added == 1 |
| 95 | assert updated == 0 |
| 96 | row = conn.execute( |
| 97 | "SELECT * FROM change_events WHERE ref_id='2026-12345'" |
| 98 | ).fetchone() |
| 99 | assert row["change"] == "added" |
| 100 | assert row["source"] == "federal_register" |
| 101 | _cleanup(conn, tmp) |
| 102 | |
| 103 | def test_updated_same_id_different_date(self): |
| 104 | conn, tmp = _make_db() |
| 105 | before = {("openomb", "FY2026_Agency.json"): "2026-01-15"} |
| 106 | after = {("openomb", "FY2026_Agency.json"): "2026-06-20"} |
| 107 | added, updated = _classify_added_updated(before, after, TS, conn) |
| 108 | assert added == 0 |
| 109 | assert updated == 1 |
| 110 | row = conn.execute( |
| 111 | "SELECT * FROM change_events WHERE ref_id='FY2026_Agency.json'" |
| 112 | ).fetchone() |
| 113 | assert row["change"] == "updated" |
| 114 | assert "2026-01-15" in row["summary"] |
| 115 | assert "2026-06-20" in row["summary"] |
| 116 | _cleanup(conn, tmp) |
| 117 | |
| 118 | def test_no_change_when_state_identical(self): |
| 119 | conn, tmp = _make_db() |
| 120 | state = {("congress", "BILLS-119hr1"): "2026-06-01"} |
| 121 | added, updated = _classify_added_updated(state, state, TS, conn) |
| 122 | assert added == 0 |
| 123 | assert updated == 0 |
| 124 | count = conn.execute( |
| 125 | "SELECT COUNT(*) FROM change_events" |
| 126 | ).fetchone()[0] |
| 127 | assert count == 0 |
| 128 | _cleanup(conn, tmp) |
| 129 | |
| 130 | def test_mixed_added_updated_unchanged(self): |
| 131 | conn, tmp = _make_db() |
| 132 | before = { |
| 133 | ("federal_register", "DOC-001"): "2026-01-01", |
| 134 | ("federal_register", "DOC-002"): "2026-01-01", |
| 135 | } |
| 136 | after = { |
| 137 | ("federal_register", "DOC-001"): "2026-06-01", |
| 138 | ("federal_register", "DOC-002"): "2026-01-01", |
| 139 | ("federal_register", "DOC-003"): "2026-06-26", |
| 140 | } |
| 141 | added, updated = _classify_added_updated(before, after, TS, conn) |
| 142 | assert added == 1 |
| 143 | assert updated == 1 |
| 144 | _cleanup(conn, tmp) |
| 145 | |
| 146 | def test_multiple_sources_in_single_diff(self): |
| 147 | conn, tmp = _make_db() |
| 148 | before = {} |
| 149 | after = { |
| 150 | ("usaspending", "CONT_AWD_001"): "2026-06-01", |
| 151 | ("treasury", "debt_2026-06-01"): "2026-06-01", |
| 152 | ("openomb", "FY2026_DOD.json"): "2026-06-01", |
| 153 | } |
| 154 | added, updated = _classify_added_updated(before, after, TS, conn) |
| 155 | assert added == 3 |
| 156 | assert updated == 0 |
| 157 | _cleanup(conn, tmp) |
| 158 | |
| 159 | def test_source_url_non_empty_on_every_event(self): |
| 160 | """Every inserted change_event must have a non-empty source_url.""" |
| 161 | conn, tmp = _make_db() |
| 162 | before = {} |
| 163 | after = { |
| 164 | ("usaspending", "CONT_AWD_XTEST"): "2026-06-20", |
| 165 | ("congress", "BILLS-119hr99"): "2026-06-20", |
| 166 | } |
| 167 | _classify_added_updated(before, after, TS, conn) |
| 168 | rows = conn.execute( |
| 169 | "SELECT source_url FROM change_events" |
| 170 | ).fetchall() |
| 171 | assert len(rows) == 2 |
| 172 | for r in rows: |
| 173 | assert r["source_url"], "source_url must be non-empty" |
| 174 | _cleanup(conn, tmp) |
| 175 | |
| 176 | def test_ref_type_mapped_correctly(self): |
| 177 | conn, tmp = _make_db() |
| 178 | before = {} |
| 179 | after = { |
| 180 | ("bills_bulk", "B001"): "2026-01-01", |
| 181 | ("federal_register", "FR001"): "2026-01-01", |
| 182 | ("usaspending", "USA001"): "2026-01-01", |
| 183 | ("votes_house", "VH001"): "2026-01-01", |
| 184 | } |
| 185 | _classify_added_updated(before, after, TS, conn) |
| 186 | rows = { |
| 187 | r["ref_id"]: r["ref_type"] |
| 188 | for r in conn.execute("SELECT ref_id, ref_type FROM change_events").fetchall() |
| 189 | } |
| 190 | assert rows["B001"] == "bill" |
| 191 | assert rows["FR001"] == "executive_action" |
| 192 | assert rows["USA001"] == "award" |
| 193 | assert rows["VH001"] == "vote" |
| 194 | _cleanup(conn, tmp) |
| 195 | |
| 196 | def test_change_field_constraint_enforced(self): |
| 197 | """SQLite CHECK constraint on change column must be honoured.""" |
| 198 | conn, tmp = _make_db() |
| 199 | with pytest.raises(Exception): |
| 200 | conn.execute( |
| 201 | "INSERT INTO change_events" |
| 202 | " (ts,source,ref_type,ref_id,change,summary,source_url)" |
| 203 | " VALUES (?,?,?,?,?,?,?)", |
| 204 | (TS, "test", "bill", "X", "deleted", "bad value", |
| 205 | "https://congress.gov/test"), |
| 206 | ) |
| 207 | conn.commit() |
| 208 | _cleanup(conn, tmp) |
| 209 | |
| 210 | |
| 211 | |
| 212 | class TestFullSetChanges: |
| 213 | |
| 214 | |
| 215 | def test_removal_detected_when_member_disappears(self): |
| 216 | conn, tmp = _make_db() |
| 217 | cfg = FULL_SET_CONFIGS["members"] |
| 218 | _seed_members_table(conn, [{"bioguide_id": "A000001"}]) |
| 219 | for bio in ("A000001", "A000002"): |
| 220 | conn.execute( |
| 221 | "INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)" |
| 222 | " VALUES (?,?,?,?)", |
| 223 | ("members", bio, 1, |
| 224 | f"https://bioguide.congress.gov/search/bio/{bio}"), |
| 225 | ) |
| 226 | conn.commit() |
| 227 | |
| 228 | added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS) |
| 229 | |
| 230 | assert removed == 1 |
| 231 | row = conn.execute( |
| 232 | "SELECT * FROM change_events WHERE ref_id='A000002'" |
| 233 | ).fetchone() |
| 234 | assert row is not None |
| 235 | assert row["change"] == "removed" |
| 236 | assert row["source_url"] |
| 237 | _cleanup(conn, tmp) |
| 238 | |
| 239 | def test_no_removal_when_set_is_stable(self): |
| 240 | conn, tmp = _make_db() |
| 241 | cfg = FULL_SET_CONFIGS["members"] |
| 242 | _seed_members_table(conn, [ |
| 243 | {"bioguide_id": "A000001"}, |
| 244 | {"bioguide_id": "A000002"}, |
| 245 | ]) |
| 246 | for bio in ("A000001", "A000002"): |
| 247 | conn.execute( |
| 248 | "INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)" |
| 249 | " VALUES (?,?,?,?)", |
| 250 | ("members", bio, 1, |
| 251 | f"https://bioguide.congress.gov/search/bio/{bio}"), |
| 252 | ) |
| 253 | conn.commit() |
| 254 | |
| 255 | added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS) |
| 256 | assert removed == 0 |
| 257 | _cleanup(conn, tmp) |
| 258 | |
| 259 | |
| 260 | def test_removal_guard_empty_pull(self): |
| 261 | """ |
| 262 | Non-empty guard: if current pull returns 0 items (table missing or empty), |
| 263 | no REMOVED events must be written. |
| 264 | """ |
| 265 | conn, tmp = _make_db() |
| 266 | cfg = FULL_SET_CONFIGS["members"] |
| 267 | conn.execute( |
| 268 | "INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)" |
| 269 | " VALUES (?,?,?,?)", |
| 270 | ("members", "A000001", 1, |
| 271 | "https://bioguide.congress.gov/search/bio/A000001"), |
| 272 | ) |
| 273 | conn.commit() |
| 274 | |
| 275 | added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS) |
| 276 | |
| 277 | assert removed == 0 |
| 278 | count = conn.execute( |
| 279 | "SELECT COUNT(*) FROM change_events WHERE change='removed'" |
| 280 | ).fetchone()[0] |
| 281 | assert count == 0 |
| 282 | _cleanup(conn, tmp) |
| 283 | |
| 284 | def test_manifest_not_overwritten_on_empty_pull(self): |
| 285 | """ |
| 286 | When the empty-pull guard fires, the existing manifest must be preserved |
| 287 | so the next cycle can still detect removals correctly. |
| 288 | """ |
| 289 | conn, tmp = _make_db() |
| 290 | cfg = FULL_SET_CONFIGS["members"] |
| 291 | conn.execute( |
| 292 | "INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)" |
| 293 | " VALUES (?,?,?,?)", |
| 294 | ("members", "A000001", 1, |
| 295 | "https://bioguide.congress.gov/search/bio/A000001"), |
| 296 | ) |
| 297 | conn.commit() |
| 298 | |
| 299 | _detect_full_set_changes(conn, "members", cfg, 2, TS) |
| 300 | |
| 301 | manifest = _load_manifest(conn, "members") |
| 302 | assert "A000001" in manifest |
| 303 | _cleanup(conn, tmp) |
| 304 | |
| 305 | |
| 306 | def test_removal_guard_first_run(self): |
| 307 | """ |
| 308 | First-run guard: no prior manifest → no REMOVED events, even when |
| 309 | the current pull has many items. |
| 310 | """ |
| 311 | conn, tmp = _make_db() |
| 312 | cfg = FULL_SET_CONFIGS["members"] |
| 313 | _seed_members_table(conn, [ |
| 314 | {"bioguide_id": "A000001"}, |
| 315 | {"bioguide_id": "A000002"}, |
| 316 | ]) |
| 317 | |
| 318 | added, removed = _detect_full_set_changes(conn, "members", cfg, 1, TS) |
| 319 | |
| 320 | assert removed == 0 |
| 321 | count = conn.execute( |
| 322 | "SELECT COUNT(*) FROM change_events WHERE change='removed'" |
| 323 | ).fetchone()[0] |
| 324 | assert count == 0 |
| 325 | _cleanup(conn, tmp) |
| 326 | |
| 327 | |
| 328 | def test_added_detected_for_members_source(self): |
| 329 | """ |
| 330 | Members poller does not call mark_seen, so manifest diff must detect ADDs. |
| 331 | """ |
| 332 | conn, tmp = _make_db() |
| 333 | cfg = FULL_SET_CONFIGS["members"] |
| 334 | _seed_members_table(conn, [ |
| 335 | {"bioguide_id": "A000001"}, |
| 336 | {"bioguide_id": "A000003"}, |
| 337 | ]) |
| 338 | conn.execute( |
| 339 | "INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)" |
| 340 | " VALUES (?,?,?,?)", |
| 341 | ("members", "A000001", 1, |
| 342 | "https://bioguide.congress.gov/search/bio/A000001"), |
| 343 | ) |
| 344 | conn.commit() |
| 345 | |
| 346 | added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS) |
| 347 | |
| 348 | assert added == 1 |
| 349 | row = conn.execute( |
| 350 | "SELECT * FROM change_events WHERE ref_id='A000003'" |
| 351 | ).fetchone() |
| 352 | assert row is not None |
| 353 | assert row["change"] == "added" |
| 354 | assert row["source_url"] |
| 355 | _cleanup(conn, tmp) |
| 356 | |
| 357 | |
| 358 | def test_manifest_updated_after_successful_cycle(self): |
| 359 | conn, tmp = _make_db() |
| 360 | cfg = FULL_SET_CONFIGS["members"] |
| 361 | _seed_members_table(conn, [ |
| 362 | {"bioguide_id": "A000001"}, |
| 363 | {"bioguide_id": "A000002"}, |
| 364 | ]) |
| 365 | |
| 366 | _detect_full_set_changes(conn, "members", cfg, 1, TS) |
| 367 | |
| 368 | manifest = _load_manifest(conn, "members") |
| 369 | assert "A000001" in manifest |
| 370 | assert "A000002" in manifest |
| 371 | _cleanup(conn, tmp) |
| 372 | |
| 373 | def test_manifest_reflects_removal_next_cycle(self): |
| 374 | """ |
| 375 | After A000002 is removed from the current pull, the manifest must |
| 376 | no longer contain it so the next cycle doesn't re-flag it. |
| 377 | """ |
| 378 | conn, tmp = _make_db() |
| 379 | cfg = FULL_SET_CONFIGS["members"] |
| 380 | |
| 381 | _seed_members_table(conn, [ |
| 382 | {"bioguide_id": "A000001"}, |
| 383 | {"bioguide_id": "A000002"}, |
| 384 | ]) |
| 385 | _detect_full_set_changes(conn, "members", cfg, 1, TS) |
| 386 | |
| 387 | conn.execute("DELETE FROM members WHERE bioguide_id='A000002'") |
| 388 | conn.commit() |
| 389 | _detect_full_set_changes(conn, "members", cfg, 2, TS) |
| 390 | |
| 391 | manifest = _load_manifest(conn, "members") |
| 392 | assert "A000001" in manifest |
| 393 | assert "A000002" not in manifest |
| 394 | |
| 395 | events_before = conn.execute( |
| 396 | "SELECT COUNT(*) FROM change_events WHERE change='removed'" |
| 397 | ).fetchone()[0] |
| 398 | _detect_full_set_changes(conn, "members", cfg, 3, TS) |
| 399 | events_after = conn.execute( |
| 400 | "SELECT COUNT(*) FROM change_events WHERE change='removed'" |
| 401 | ).fetchone()[0] |
| 402 | assert events_after == events_before, "A000002 must not be flagged twice" |
| 403 | _cleanup(conn, tmp) |
| 404 | |
| 405 | |
| 406 | def test_removed_event_has_source_url(self): |
| 407 | """Every REMOVED event must carry a non-empty source_url.""" |
| 408 | conn, tmp = _make_db() |
| 409 | cfg = FULL_SET_CONFIGS["members"] |
| 410 | _seed_members_table(conn, [{"bioguide_id": "A000001"}]) |
| 411 | for bio in ("A000001", "A000002"): |
| 412 | conn.execute( |
| 413 | "INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)" |
| 414 | " VALUES (?,?,?,?)", |
| 415 | ("members", bio, 1, |
| 416 | f"https://bioguide.congress.gov/search/bio/{bio}"), |
| 417 | ) |
| 418 | conn.commit() |
| 419 | |
| 420 | _detect_full_set_changes(conn, "members", cfg, 2, TS) |
| 421 | |
| 422 | rows = conn.execute( |
| 423 | "SELECT source_url FROM change_events WHERE change='removed'" |
| 424 | ).fetchall() |
| 425 | assert len(rows) == 1 |
| 426 | assert rows[0]["source_url"] |
| 427 | _cleanup(conn, tmp) |