fix(addresses): preserve complete import rollback evidence and batch previews

Release v0.1.23. Coordinated integrity review: GovOPlaN/govoplan-core#298.

tests/test_tabular_imports.py

@@ -3,12 +3,13 @@ from __future__ import annotations
 import base64
 from io import BytesIO
 import unittest
+from unittest.mock import patch
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, event
 from sqlalchemy.orm import sessionmaker
 from openpyxl import Workbook
 
-from govoplan_addresses.backend.db.models import AddressBook, Contact
+from govoplan_addresses.backend.db.models import AddressBook, AddressList, AddressListEntry, Contact, ContactEmail, ContactPhone, ContactPostalAddress
 from govoplan_addresses.backend.import_schemas import (
     AddressImportConfiguration,
     AddressImportPreviewRequest,
@@ -18,6 +19,8 @@ from govoplan_addresses.backend.import_schemas import (
 )
 from govoplan_addresses.backend.imports import (
     apply_address_import,
+    _contact_hash,
+    _contact_snapshot,
     create_import_profile,
     get_import_run,
     import_run_payload,
@@ -26,6 +29,7 @@ from govoplan_addresses.backend.imports import (
     update_import_profile,
 )
 from govoplan_core.db.base import Base
+from govoplan_addresses.backend.service import delete_contact
 
 
 class Principal:
@@ -91,6 +95,315 @@ class AddressTabularImportTests(unittest.TestCase):
         )
         self.session.flush()
 
+    def _preview_rows(self, count: int = 1, *, organization: str = "Office"):
+        return preview_address_import(
+            self.session, self.principal, self.book.id,
+            AddressImportPreviewRequest(
+                profile_id=self.profile.id, filename="fixture.csv",
+                content_base64=encoded(
+                    "id;first;last;email;organization\n"
+                    + "".join(f"{index};Given;Family;u{index}@example.test;{organization}\n" for index in range(count))
+                ),
+            ),
+        )
+
+    def _apply(self, run):
+        apply_address_import(self.session, self.principal, run.id, expected_plan_hash=run.plan_hash)
+        self.session.commit()
+
+    def test_rollback_restores_previously_deleted_state_and_source_fields(self) -> None:
+        self._apply(self._preview_rows())
+        contact = self.session.query(Contact).one()
+        delete_contact(self.session, self.principal, contact.id)
+        self.session.commit()
+        prior_deleted_at = contact.deleted_at
+        prior_source_revision = contact.source_revision
+        run = self._preview_rows(organization="Changed")
+        self._apply(run)
+        self.assertIsNone(contact.deleted_at)
+        before = run.result_evidence["updated_contacts"][0]["before"]
+        self.assertEqual(2, before["version"])
+        self.assertEqual(prior_deleted_at.isoformat(), before["deleted_at"])
+        rollback_address_import(
+            self.session, self.principal, run.id,
+            AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Restore the reviewed previous state."),
+        )
+        self.session.commit()
+        self.assertEqual("rolled_back", run.status)
+        self.assertEqual(prior_deleted_at, contact.deleted_at)
+        self.assertEqual("Office", contact.organization)
+        self.assertEqual(prior_source_revision, contact.source_revision)
+
+    def test_rollback_restores_all_point_evidence_and_identities_after_reload(self) -> None:
+        self._apply(self._preview_rows())
+        contact = self.session.query(Contact).one()
+        contact.emails[0].original_email = " U0@EXAMPLE.TEST "
+        contact.emails[0].provenance = {"nested": {"original": "email evidence"}}
+        contact.note = "  Exact retained note\r\n"
+        contact.tags = ["Exact", "Exact", " padded "]
+        contact.metadata_ = None
+        contact.phones.append(ContactPhone(
+            phone="+49 123", original_phone=" +49 (123) ", normalized_phone="+49123",
+            provenance={"original": "phone evidence"}, label="Office", is_primary=True, order_index=4,
+        ))
+        contact.postal_addresses.append(ContactPostalAddress(
+            street="Main Street", original_value={"street": " Main Street "},
+            normalized_value={"street": "main street"}, provenance={"original": "postal evidence"},
+            is_primary=True, order_index=7,
+        ))
+        self.session.commit()
+        before = _contact_snapshot(contact)["points"]
+        run = self._preview_rows(organization="Changed")
+        self._apply(run)
+        self.assertEqual(before["emails"], _contact_snapshot(contact)["points"]["emails"])
+        self.session.expire_all()
+        rollback_address_import(
+            self.session, self.principal, run.id,
+            AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Restore all original point evidence."),
+        )
+        self.session.commit()
+        self.session.expire_all()
+        self.assertEqual(before, _contact_snapshot(contact)["points"])
+        self.assertEqual("  Exact retained note\r\n", contact.note)
+        self.assertEqual(["Exact", "Exact", " padded "], contact.tags)
+        self.assertIsNone(contact.metadata_)
+
+    def test_post_import_point_provenance_edit_is_guarded(self) -> None:
+        self._apply(self._preview_rows())
+        run = self._preview_rows(organization="Changed")
+        self._apply(run)
+        contact = self.session.query(Contact).one()
+        after_hash = _contact_hash(contact)
+        contact.emails[0].provenance = {"later": "manual evidence"}
+        self.session.commit()
+        self.assertNotEqual(after_hash, _contact_hash(contact))
+        with self.assertRaisesRegex(ValueError, "changed after import"):
+            rollback_address_import(
+                self.session, self.principal, run.id,
+                AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Do not erase a later point edit."),
+            )
+        self.assertEqual({"later": "manual evidence"}, contact.emails[0].provenance)
+
+    def test_point_lock_order_does_not_change_tied_collection_order_or_reviewed_hash(self) -> None:
+        self._apply(self._preview_rows())
+        contact = self.session.query(Contact).one()
+        contact.emails.append(ContactEmail(
+            id="00000000-0000-0000-0000-000000000000", email="extra@example.test",
+            original_email="extra@example.test", normalized_email="extra@example.test",
+            label="Extra", is_primary=False, order_index=0,
+        ))
+        self.session.commit()
+        self.session.expire_all()
+        before = _contact_snapshot(contact)["points"]
+        run = self._preview_rows(organization="Changed")
+        self._apply(run)
+        rollback_address_import(
+            self.session, self.principal, run.id,
+            AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Preserve tied contact-point ordering."),
+        )
+        self.session.commit()
+        self.session.expire_all()
+        self.assertEqual(before, _contact_snapshot(contact)["points"])
+
+    def _link_email(self, contact):
+        address_list = AddressList(address_book_id=self.book.id, tenant_id="tenant-1", name="Recipients")
+        entry = AddressListEntry(address_list=address_list, contact=contact, contact_email=contact.emails[0], target_kind="email")
+        self.session.add(entry)
+        self.session.commit()
+        return entry
+
+    def test_unchanged_point_keeps_address_list_identity_through_apply_and_rollback(self) -> None:
+        self._apply(self._preview_rows())
+        contact = self.session.query(Contact).one()
+        entry = self._link_email(contact)
+        point_id = contact.emails[0].id
+        run = self._preview_rows(organization="Changed")
+        self._apply(run)
+        self.session.expire_all()
+        self.assertEqual(point_id, entry.contact_email_id)
+        self.assertEqual(point_id, contact.emails[0].id)
+        rollback_address_import(
+            self.session, self.principal, run.id,
+            AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Keep the explicit recipient reference."),
+        )
+        self.session.commit()
+        self.session.expire_all()
+        self.assertEqual(point_id, entry.contact_email_id)
+        self.assertEqual(point_id, contact.emails[0].id)
+
+    def test_replacing_a_linked_point_requires_reconciliation_before_contact_mutation(self) -> None:
+        self._apply(self._preview_rows())
+        contact = self.session.query(Contact).one()
+        entry = self._link_email(contact)
+        run = preview_address_import(
+            self.session, self.principal, self.book.id,
+            AddressImportPreviewRequest(profile_id=self.profile.id, filename="fixture.csv", content_base64=encoded(
+                "id;first;last;email;organization\n0;Given;Family;different@example.test;Changed\n"
+            )),
+        )
+        before = _contact_hash(contact)
+        with self.assertRaisesRegex(ValueError, "address-list or governance references"):
+            apply_address_import(self.session, self.principal, run.id, expected_plan_hash=run.plan_hash)
+        self.assertEqual(before, _contact_hash(contact))
+        self.assertEqual(contact.emails[0].id, entry.contact_email_id)
+        self.assertEqual("previewed", run.status)
+
+    def test_new_reference_to_imported_point_blocks_destructive_rollback(self) -> None:
+        self._apply(self._preview_rows())
+        run = preview_address_import(
+            self.session, self.principal, self.book.id,
+            AddressImportPreviewRequest(profile_id=self.profile.id, filename="fixture.csv", content_base64=encoded(
+                "id;first;last;email;organization\n0;Given;Family;different@example.test;Changed\n"
+            )),
+        )
+        self._apply(run)
+        contact = self.session.query(Contact).one()
+        entry = self._link_email(contact)
+        before = _contact_hash(contact)
+        with self.assertRaisesRegex(ValueError, "address-list or governance references"):
+            rollback_address_import(
+                self.session, self.principal, run.id,
+                AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Retain the newly referenced recipient point."),
+            )
+        self.assertEqual(before, _contact_hash(contact))
+        self.assertEqual(contact.emails[0].id, entry.contact_email_id)
+        self.assertEqual("applied", run.status)
+
+    def test_version_one_point_incomplete_evidence_is_not_accepted(self) -> None:
+        import copy
+
+        self._apply(self._preview_rows())
+        run = self._preview_rows(organization="Changed")
+        self._apply(run)
+        evidence = copy.deepcopy(run.result_evidence)
+        evidence["updated_contacts"][0]["before"]["version"] = 1
+        run.result_evidence = evidence
+        self.session.commit()
+        with self.assertRaisesRegex(ValueError, "incomplete legacy rollback evidence"):
+            rollback_address_import(
+                self.session, self.principal, run.id,
+                AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Do not infer missing point evidence."),
+            )
+
+    def test_legacy_or_incomplete_before_images_fail_before_any_rollback_mutation(self) -> None:
+        self._apply(self._preview_rows())
+        run = self._preview_rows(2, organization="Changed")
+        self._apply(run)
+        evidence = dict(run.result_evidence)
+        updates = [dict(item) for item in evidence["updated_contacts"]]
+        updates[0]["before"] = updates[0]["before"]["contact"]
+        evidence["updated_contacts"] = updates
+        run.result_evidence = evidence
+        self.session.commit()
+        with self.assertRaisesRegex(ValueError, "incomplete legacy rollback evidence"):
+            rollback_address_import(
+                self.session, self.principal, run.id,
+                AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Reject an incomplete previous state."),
+            )
+        self.assertEqual("applied", run.status)
+        self.assertEqual(2, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())
+        self.assertTrue(all(contact.organization == "Changed" for contact in self.session.query(Contact)))
+
+    def test_preview_queries_are_batched_and_relationships_are_eager(self) -> None:
+        queries = []
+
+        def capture(conn, cursor, statement, parameters, context, executemany):
+            if statement.lstrip().upper().startswith("SELECT"):
+                queries.append(statement)
+
+        event.listen(self.session.bind, "before_cursor_execute", capture)
+        try:
+            first = self._preview_rows(50)
+            self.assertEqual(50, first.statistics["create"])
+            self.assertLessEqual(len(queries), 3)
+            self._apply(first)
+            self.session.expunge_all()
+            queries.clear()
+            repeat = self._preview_rows(50)
+            self.assertEqual(50, repeat.statistics["unchanged"])
+            self.assertLessEqual(len(queries), 6)
+        finally:
+            event.remove(self.session.bind, "before_cursor_execute", capture)
+
+    def test_preview_batches_preserve_first_source_match_and_book_scope(self) -> None:
+        self._apply(self._preview_rows(5))
+        original = self.session.query(Contact).order_by(Contact.created_at, Contact.id).first()
+        other_book = AddressBook(tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Other", source_kind="local", read_only=False)
+        self.session.add(other_book)
+        self.session.flush()
+        self.session.add_all([
+            Contact(tenant_id="tenant-1", address_book_id=original.address_book_id, display_name="Later duplicate", source_ref=original.source_ref),
+            Contact(tenant_id="tenant-1", address_book_id=other_book.id, display_name="Other book", source_ref=original.source_ref),
+        ])
+        self.session.commit()
+        with patch("govoplan_addresses.backend.imports.CONTACT_LOOKUP_BATCH_SIZE", 2):
+            repeat = self._preview_rows(5)
+        self.assertEqual(5, repeat.statistics["unchanged"])
+
+    def test_missing_created_after_hash_blocks_rollback(self) -> None:
+        run = self._preview_rows()
+        self._apply(run)
+        run.plan_data = [{key: value for key, value in item.items() if key != "after_hash"} for item in run.plan_data]
+        self.session.commit()
+        with self.assertRaisesRegex(ValueError, "evidence is incomplete"):
+            rollback_address_import(
+                self.session, self.principal, run.id,
+                AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Incomplete evidence must not delete contacts."),
+            )
+        self.assertEqual(1, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())
+
+    def test_applied_effect_hashes_survive_commit_and_reload(self) -> None:
+        run = self._preview_rows()
+        self._apply(run)
+        run_id, plan_hash = run.id, run.plan_hash
+        self.session.expunge_all()
+        reloaded = get_import_run(self.session, self.principal, run_id)
+        self.assertTrue(reloaded.plan_data[0]["contact_id"])
+        self.assertEqual(64, len(reloaded.plan_data[0]["after_hash"]))
+        rollback_address_import(
+            self.session, self.principal, run_id,
+            AddressImportRollbackRequest(expected_plan_hash=plan_hash, reason="Durable after-images guard rollback."),
+        )
+        self.session.commit()
+        self.assertEqual(0, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())
+
+    def test_replayed_apply_still_requires_the_reviewed_plan_hash(self) -> None:
+        run = self._preview_rows()
+        self._apply(run)
+        with self.assertRaisesRegex(ValueError, "reviewed import plan changed"):
+            apply_address_import(self.session, self.principal, run.id, expected_plan_hash="0" * 64)
+
+    def test_rollback_rejects_a_moved_target_without_archiving_it(self) -> None:
+        run = self._preview_rows()
+        self._apply(run)
+        other_book = AddressBook(tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Other", source_kind="local", read_only=False)
+        self.session.add(other_book)
+        self.session.flush()
+        contact = self.session.query(Contact).one()
+        contact.address_book_id = other_book.id
+        self.session.commit()
+        with self.assertRaisesRegex(ValueError, "moved to another address book"):
+            rollback_address_import(
+                self.session, self.principal, run.id,
+                AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Moved contacts require manual reconciliation."),
+            )
+        self.assertIsNone(contact.deleted_at)
+        self.assertEqual("applied", run.status)
+
+    def test_rollback_retains_post_import_edits(self) -> None:
+        run = self._preview_rows()
+        self._apply(run)
+        contact = self.session.query(Contact).one()
+        contact.organization = "Later manual edit"
+        self.session.commit()
+        with self.assertRaisesRegex(ValueError, "changed after import"):
+            rollback_address_import(
+                self.session, self.principal, run.id,
+                AddressImportRollbackRequest(expected_plan_hash=run.plan_hash, reason="Later changes must remain untouched."),
+            )
+        self.assertIsNone(contact.deleted_at)
+        self.assertEqual("Later manual edit", contact.organization)
+
     def test_preview_apply_repeat_and_guarded_rollback(self) -> None:
         payload = AddressImportPreviewRequest(
             profile_id=self.profile.id,