"""Integration tests for Phase 2B cutover approval and action ledger.""" from __future__ import annotations import hashlib import json import uuid from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import Connection, text from osm_lead_source_service.errors import InvalidConfiguration from osm_lead_source_service.noco.write import DuplicateRecord from osm_lead_source_service.sidecar.writer_ledger import PostgresInsertLedger from osm_lead_source_service.writer.models import ShadowReportDocument from osm_lead_source_service.writer.service import ReservationState def _shadow_document(*, osm_id: int = 42) -> ShadowReportDocument: report: dict[str, object] = { "category_counts": {"Hospitality/Restaurant or cafe": 1}, "definition_digest": "d" * 64, "leads": [ { "address": "1 fixture road", "category": "Hospitality", "city": "chiang mai", "country": "TH", "domain": "fixture.example", "evidence": ["amenity=restaurant"], "identity_hash": "b" * 64, "latitude_e7": 188000000, "longitude_e7": 989000000, "matched_rule_ids": ["hospitality.restaurant"], "name": "fixture cafe", "osm_id": osm_id, "osm_type": "node", "phone": "+6621234567", "postcode": "50000", "source_tags_hash": "c" * 64, "state": "chiang mai", "subcategory": "Restaurant or cafe", } ], "object_counts": {"nodes": 1, "relations": 0, "total": 1, "ways": 0}, "profile": {"id": "luxeillum", "version": "0.1.0-draft"}, "region_id": "thailand", "review_items": [], "source_sha256": "a" * 64, "source_size_bytes": 123, "state_counts": {"active": 1}, "total_tags": 2, } canonical = json.dumps(report, sort_keys=True, separators=(",", ":")).encode() report_hash = hashlib.sha256(canonical).hexdigest() report["report_hash"] = report_hash return ShadowReportDocument.from_json( json.dumps(report), expected_report_hash=report_hash, max_candidates=10, ) def _approval( db: Connection, document: ShadowReportDocument, *, approval_id: str = "approval-fixture", table_id: str = "table-fixture", max_inserts: int = 5, expires_delta: timedelta = timedelta(hours=1), ) -> None: now = datetime.now(UTC) db.execute( text( """ INSERT INTO osm_lead_source.writer_cutover_approval (approval_id, report_hash, region_id, profile_id, profile_version, noco_table_id, approved_max_inserts, legacy_writer_disabled_at, approved_by, approved_at, expires_at, evidence_json) VALUES (:approval_id, :report_hash, :region_id, :profile_id, :profile_version, :table_id, :max_inserts, :legacy_disabled, 'integration-test', :approved_at, :expires_at, CAST(:evidence AS jsonb)) """ ), { "approval_id": approval_id, "report_hash": document.report_hash, "region_id": document.region_id, "profile_id": document.profile_id, "profile_version": document.profile_version, "table_id": table_id, "max_inserts": max_inserts, "legacy_disabled": now - timedelta(minutes=10), "approved_at": now - timedelta(minutes=5), "expires_at": now + expires_delta, "evidence": json.dumps({"ticket": "fixture"}), }, ) def _database_url(db: Connection) -> str: return str(db.engine.url) def _isolated_identity(prefix: str) -> tuple[str, int]: token = uuid.uuid4().hex return f"{prefix}-{token}", int(token[:15], 16) + 1 @pytest.mark.database @pytest.mark.integration def test_cutover_approval_must_match_exact_run(db: Connection) -> None: approval_id, osm_id = _isolated_identity("approval-exact") document = _shadow_document(osm_id=osm_id) _approval(db, document, approval_id=approval_id) db.commit() ledger = PostgresInsertLedger( database_url=_database_url(db), approval_id=approval_id, ) ledger.assert_cutover_approved(document, table_id="table-fixture", max_inserts=5) with pytest.raises(InvalidConfiguration, match="table"): ledger.assert_cutover_approved(document, table_id="different", max_inserts=5) with pytest.raises(InvalidConfiguration, match="cap"): ledger.assert_cutover_approved(document, table_id="table-fixture", max_inserts=6) @pytest.mark.database @pytest.mark.integration def test_ledger_reservation_finalization_and_replay(db: Connection) -> None: approval_id, osm_id = _isolated_identity("approval-duplicate") document = _shadow_document(osm_id=osm_id) candidate = document.candidates[0] _approval(db, document, approval_id=approval_id) db.commit() ledger = PostgresInsertLedger( database_url=_database_url(db), approval_id=approval_id, ) assert ledger.reserve(candidate) is ReservationState.CLAIMED assert ledger.reserve(candidate) is ReservationState.IN_PROGRESS ledger.mark_duplicate( candidate, duplicates=(DuplicateRecord(record_id=77, match_paths=("osm_identity",)),), ) assert ledger.reserve(candidate) is ReservationState.ALREADY_APPLIED with db.engine.connect() as inspection: row = ( inspection.execute( text( "SELECT state, attempt_count, duplicate_evidence_json " "FROM osm_lead_source.insert_action_ledger " "WHERE idempotency_key = :key" ), {"key": candidate.idempotency_key}, ) .mappings() .one() ) assert row["state"] == "duplicate" assert row["attempt_count"] == 1 assert row["duplicate_evidence_json"][0]["record_id"] == 77 @pytest.mark.database @pytest.mark.integration def test_safe_pre_post_failure_can_be_reclaimed(db: Connection) -> None: approval_id, osm_id = _isolated_identity("approval-reclaim") document = _shadow_document(osm_id=osm_id) candidate = document.candidates[0] _approval(db, document, approval_id=approval_id) db.commit() ledger = PostgresInsertLedger( database_url=_database_url(db), approval_id=approval_id, ) assert ledger.reserve(candidate) is ReservationState.CLAIMED ledger.mark_failed(candidate, reason="NocoLookupFailure") assert ledger.reserve(candidate) is ReservationState.CLAIMED ledger.mark_applied(candidate, table_id="table-fixture", record_id=9001) assert ledger.reserve(candidate) is ReservationState.ALREADY_APPLIED with db.engine.connect() as inspection: row = ( inspection.execute( text( "SELECT state, attempt_count, noco_record_id " "FROM osm_lead_source.insert_action_ledger " "WHERE idempotency_key = :key" ), {"key": candidate.idempotency_key}, ) .mappings() .one() ) assert row == {"state": "applied", "attempt_count": 2, "noco_record_id": 9001} @pytest.mark.database @pytest.mark.integration def test_global_writer_lease_rejects_concurrent_holder(db: Connection) -> None: approval_id, osm_id = _isolated_identity("approval-lease") document = _shadow_document(osm_id=osm_id) _approval(db, document, approval_id=approval_id) db.commit() first = PostgresInsertLedger( database_url=_database_url(db), approval_id=approval_id, ) second = PostgresInsertLedger( database_url=_database_url(db), approval_id=approval_id, ) with first.run_lease(owner="first"): with pytest.raises(InvalidConfiguration, match="holds the lease"): with second.run_lease(owner="second"): raise AssertionError("unreachable")