From d4fa024034e52924e9573b13a46dd0cee2287d9e Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 8 Sep 2026 12:19:36 +0200 Subject: [PATCH] fix(addresses): preserve complete import rollback evidence and batch previews Release v0.1.23. Coordinated integrity review: GovOPlaN/govoplan-core#298. --- package.json | 2 +- pyproject.toml | 2 +- src/govoplan_addresses/backend/imports.py | 362 +++++++++++++++++++-- src/govoplan_addresses/backend/manifest.py | 29 +- src/govoplan_addresses/backend/service.py | 7 +- tests/test_tabular_imports.py | 317 +++++++++++++++++- webui/package.json | 2 +- 7 files changed, 677 insertions(+), 44 deletions(-) diff --git a/package.json b/package.json index 0405256..54f83d1 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/addresses-webui", - "version": "0.1.22", + "version": "0.1.23", "private": true, "type": "module", "main": "webui/src/index.ts", diff --git a/pyproject.toml b/pyproject.toml index 4c98269..591a45e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "govoplan-addresses" -version = "0.1.22" +version = "0.1.23" description = "GovOPlaN reusable address and recipient-source module." readme = "README.md" requires-python = ">=3.12" diff --git a/src/govoplan_addresses/backend/imports.py b/src/govoplan_addresses/backend/imports.py index 67509ed..8bfa489 100644 --- a/src/govoplan_addresses/backend/imports.py +++ b/src/govoplan_addresses/backend/imports.py @@ -3,19 +3,28 @@ from __future__ import annotations import base64 import binascii import csv +import copy import hashlib import json from collections import Counter +from datetime import UTC, datetime from io import BytesIO, StringIO from typing import Any -from sqlalchemy import and_, false, or_ -from sqlalchemy.orm import Session +from sqlalchemy import and_, false, func, or_, select +from sqlalchemy.orm import Session, selectinload +from sqlalchemy.orm.attributes import set_committed_value from govoplan_addresses.backend.db.models import ( AddressImportProfile, AddressImportRun, + AddressListEntry, Contact, + ContactChannelRule, + ContactEmail, + ContactPhone, + ContactPointQualityDecision, + ContactPostalAddress, ) from govoplan_addresses.backend.import_schemas import ( AddressImportConfiguration, @@ -34,6 +43,12 @@ from govoplan_addresses.backend.schemas import ( ) from govoplan_addresses.backend.service import ( AddressBookError, + _contact_change_payload, + _record_address_contact_change, + _require_mutable_book, + _replace_emails, + _replace_phones, + _replace_postal_addresses, create_contact, delete_contact, get_visible_address_book, @@ -47,6 +62,16 @@ from govoplan_core.db.base import utcnow MAX_IMPORT_BYTES = 10_000_000 MAX_IMPORT_COLUMNS = 200 +CONTACT_LOOKUP_BATCH_SIZE = 250 +ROLLBACK_SNAPSHOT_VERSION = 2 +CONTACT_POINTS = { + "emails": (ContactEmail, ("email",), _replace_emails), + "phones": (ContactPhone, ("phone",), _replace_phones), + "postal_addresses": ( + ContactPostalAddress, ("street", "postal_code", "locality", "region", "country"), + _replace_postal_addresses, + ), +} def _account_id(principal: ApiPrincipal) -> str: @@ -241,15 +266,19 @@ def get_import_run( session: Session, principal: ApiPrincipal, run_id: str, + *, + lock: bool = False, ) -> AddressImportRun: visible_book_ids = [book.id for book in _visible_import_books(session, principal)] if not visible_book_ids: raise AddressBookError("Address import run not found.") - run = ( + query = ( session.query(AddressImportRun) .filter(AddressImportRun.id == run_id, AddressImportRun.address_book_id.in_(visible_book_ids)) - .one_or_none() ) + if lock: + query = query.populate_existing().with_for_update() + run = query.one_or_none() if run is None: raise AddressBookError("Address import run not found.") return run @@ -262,13 +291,13 @@ def apply_address_import( *, expected_plan_hash: str, ) -> AddressImportRun: - run = get_import_run(session, principal, run_id) + run = get_import_run(session, principal, run_id, lock=True) + if run.plan_hash != expected_plan_hash: + raise AddressBookError("The reviewed import plan changed; create a new preview.") if run.status == "applied": return run if run.status != "previewed": raise AddressBookError(f"Import run cannot be applied from status {run.status!r}.") - if run.plan_hash != expected_plan_hash: - raise AddressBookError("The reviewed import plan changed; create a new preview.") if any(item.get("severity") == "error" for item in run.diagnostics or []): raise AddressBookError("Import plans with error diagnostics cannot be applied.") if any(item.get("action") == "conflict" for item in run.plan_data or []): @@ -276,12 +305,15 @@ def apply_address_import( created_ids: list[str] = [] updated: list[dict[str, Any]] = [] - for item in run.plan_data or []: + # SQLAlchemy JSON columns do not track nested mutations. Keep the persisted + # preview untouched until assigning a genuinely changed complete plan. + applied_plan = copy.deepcopy(run.plan_data or []) + for item in applied_plan: action = item.get("action") if action in {"ignored", "unchanged"}: continue source_ref = str(item["source_ref"]) - existing = _contact_by_source_ref(session, run.address_book_id, source_ref) + existing = _contact_by_source_ref(session, run.address_book_id, source_ref, lock=True) if action == "create": if existing is not None and existing.deleted_at is None: raise AddressBookError("A target contact appeared after preview; preview the import again.") @@ -299,18 +331,27 @@ def apply_address_import( elif action == "update": if existing is None: raise AddressBookError("An import target disappeared after preview; preview the import again.") + _lock_contact_points(session, existing) if _contact_hash(existing) != item.get("expected_contact_hash"): raise AddressBookError( f'Contact "{existing.display_name}" changed after preview; preview the import again.' ) - before = _contact_snapshot(existing) + _require_mutable_book(existing.address_book) + before = { + "version": ROLLBACK_SNAPSHOT_VERSION, + "contact": copy.deepcopy(_contact_snapshot(existing)), + "deleted_at": _deleted_at_value(existing.deleted_at), + } + update_payload = ContactUpdateRequest.model_validate(item["payload"]) + points = _prepare_import_points(session, existing, update_payload) if existing.deleted_at is not None: restore_contact(session, principal, existing.id) + _apply_import_points(existing, points) contact = update_contact( session, principal, existing.id, - ContactUpdateRequest.model_validate(item["payload"]), + _without_contact_points(update_payload), ) _stamp_import_contact(contact, run=run, item=item) session.flush() @@ -319,7 +360,7 @@ def apply_address_import( run.status = "applied" run.applied_at = utcnow() - run.plan_data = list(run.plan_data or []) + run.plan_data = applied_plan run.result_evidence = { "input_hash": run.input_hash, "plan_hash": run.plan_hash, @@ -337,7 +378,7 @@ def rollback_address_import( run_id: str, payload: AddressImportRollbackRequest, ) -> AddressImportRun: - run = get_import_run(session, principal, run_id) + run = get_import_run(session, principal, run_id, lock=True) if run.plan_hash != payload.expected_plan_hash: raise AddressBookError("The reviewed import plan changed; reload the import run.") if run.status == "rolled_back": @@ -347,6 +388,12 @@ def rollback_address_import( evidence = dict(run.result_evidence or {}) updated = list(evidence.get("updated_contacts") or []) created_ids = list(evidence.get("created_contact_ids") or []) + # Validate every before-image before touching any contact. Older runs did + # not record deletion state, so automatic recovery cannot infer it safely. + before_images = { + str(item["contact_id"]): _validated_rollback_snapshot(item.get("before")) + for item in updated + } expected_hashes = { str(item["contact_id"]): str(item["after_hash"]) @@ -359,12 +406,20 @@ def rollback_address_import( if item.get("contact_id") in created_ids and item.get("after_hash") } ) - for contact_id, expected_hash in expected_hashes.items(): - contact = get_visible_contact(session, principal, contact_id, include_deleted=True) + if not set(created_ids).issubset(expected_hashes): + raise AddressBookError("Import rollback evidence is incomplete; automatic rollback is unsafe.") + for contact_id, expected_hash in sorted(expected_hashes.items()): + contact = get_visible_contact(session, principal, contact_id, include_deleted=True, lock=True) + _lock_contact_points(session, contact) + if contact.address_book_id != run.address_book_id: + raise AddressBookError("An import target moved to another address book; automatic rollback is unsafe.") + _require_mutable_book(contact.address_book) if _contact_hash(contact) != expected_hash: raise AddressBookError( f'Contact "{contact.display_name}" changed after import; automatic rollback is unsafe.' ) + if contact_id in before_images: + _validate_point_restoration(session, contact, before_images[contact_id][0]["points"]) for contact_id in created_ids: contact = get_visible_contact(session, principal, contact_id, include_deleted=True) @@ -372,22 +427,23 @@ def rollback_address_import( delete_contact(session, principal, contact.id) for item in updated: contact = get_visible_contact(session, principal, str(item["contact_id"]), include_deleted=True) - snapshot = dict(item["before"]) + snapshot, previous_deleted_at = before_images[str(item["contact_id"])] if contact.deleted_at is not None: restore_contact(session, principal, contact.id) - update_contact( - session, - principal, - contact.id, - ContactUpdateRequest.model_validate(snapshot["payload"]), - ) - contact.source_kind = snapshot.get("source_kind") or "local" - contact.source_ref = snapshot.get("source_ref") - contact.source_revision = snapshot.get("source_revision") - contact.source_payload_kind = snapshot.get("source_payload_kind") - contact.source_payload_raw = snapshot.get("source_payload_raw") - contact.provenance = dict(snapshot.get("provenance") or {}) - contact.metadata_ = dict(snapshot.get("metadata") or {}) + previous = _contact_change_payload(contact, prefix="previous_") + _restore_contact_points(contact, snapshot["points"]) + # A validated stored before-image is not a fresh user edit: do not trim, + # normalize or coerce it through the generic update path a second time. + for field in ("display_name", "given_name", "family_name", "organization", "role_title", "note", "tags"): + setattr(contact, field, copy.deepcopy(snapshot["payload"][field])) + for field in ("source_kind", "source_ref", "source_revision", "source_payload_kind", "source_payload_raw", "provenance"): + setattr(contact, field, copy.deepcopy(snapshot[field])) + contact.metadata_ = copy.deepcopy(snapshot["metadata"]) + contact.updated_by_account_id = _account_id(principal) + _record_address_contact_change(session, principal, contact=contact, operation="updated", previous=previous) + if previous_deleted_at is not None: + delete_contact(session, principal, contact.id) + contact.deleted_at = previous_deleted_at run.status = "rolled_back" run.rolled_back_at = utcnow() @@ -701,6 +757,12 @@ def _plan_rows( first_index.setdefault(key, index) last_index[key] = index + existing_contacts = _contacts_by_source_refs( + session, + book_id, + [f"import:{profile.profile_key}:{key}" for key in key_counts], + ) + for index, (row_number, row, key) in enumerate(keyed_rows): if key_counts[key] > 1: if config.duplicate_source_key_policy == "reject": @@ -721,7 +783,7 @@ def _plan_rows( ) diagnostics.extend(row_diagnostics) source_ref = f"import:{profile.profile_key}:{key}" - existing = _contact_by_source_ref(session, book_id, source_ref) + existing = existing_contacts.get(source_ref) payload = _payload_from_mapped( mapped, profile=profile, @@ -951,13 +1013,79 @@ def _diagnostic( } -def _contact_by_source_ref(session: Session, book_id: str, source_ref: str) -> Contact | None: - return ( +def _contacts_by_source_refs(session: Session, book_id: str, source_refs: list[str]) -> dict[str, Contact]: + contacts: dict[str, Contact] = {} + for offset in range(0, len(source_refs), CONTACT_LOOKUP_BATCH_SIZE): + candidates = ( + select( + Contact.id, + func.row_number().over( + partition_by=Contact.source_ref, + order_by=(Contact.created_at.asc(), Contact.id.asc()), + ).label("source_position"), + ) + .where( + Contact.address_book_id == book_id, + Contact.source_ref.in_(source_refs[offset : offset + CONTACT_LOOKUP_BATCH_SIZE]), + ) + .subquery() + ) + rows = ( + session.query(Contact) + .join(candidates, candidates.c.id == Contact.id) + .filter(candidates.c.source_position == 1) + .options(selectinload(Contact.emails), selectinload(Contact.phones), selectinload(Contact.postal_addresses)) + .populate_existing() + .order_by(Contact.created_at.asc(), Contact.id.asc()) + .all() + ) + for contact in rows: + # Preserve the historical first-match choice for duplicate stored + # source references, independently of batch/database row order. + contacts.setdefault(str(contact.source_ref), contact) + return contacts + + +def _contact_by_source_ref(session: Session, book_id: str, source_ref: str, *, lock: bool = False) -> Contact | None: + query = ( session.query(Contact) .filter(Contact.address_book_id == book_id, Contact.source_ref == source_ref) .order_by(Contact.created_at.asc(), Contact.id.asc()) - .first() ) + if lock: + query = query.populate_existing().with_for_update() + return query.first() + + +def _validated_rollback_snapshot(value: object) -> tuple[dict[str, Any], datetime | None]: + if ( + not isinstance(value, dict) + or value.get("version") != ROLLBACK_SNAPSHOT_VERSION + or "deleted_at" not in value + or not isinstance(value.get("contact"), dict) + ): + raise AddressBookError( + "This import has incomplete legacy rollback evidence; automatic rollback is unsafe. " + "Review and reconcile the affected contacts manually." + ) + snapshot = value["contact"] + required_fields = { + "display_name", "given_name", "family_name", "organization", "role_title", + "note", "tags", "emails", "phones", "postal_addresses", "provenance", + } + if ( + not isinstance(snapshot.get("payload"), dict) + or not required_fields.issubset(snapshot["payload"]) + or not {"source_kind", "source_ref", "source_revision", "source_payload_kind", "source_payload_raw", "provenance", "metadata"}.issubset(snapshot) + ): + raise AddressBookError("Import rollback evidence is incomplete; automatic rollback is unsafe.") + try: + deleted_at = datetime.fromisoformat(value["deleted_at"]) if value["deleted_at"] is not None else None + ContactUpdateRequest.model_validate(snapshot["payload"]) + _validate_point_snapshot(snapshot.get("points")) + except (TypeError, ValueError) as exc: + raise AddressBookError("Import rollback evidence is invalid; automatic rollback is unsafe.") from exc + return snapshot, deleted_at def _stamp_import_contact(contact: Contact, *, run: AddressImportRun, item: dict[str, Any]) -> None: @@ -1007,12 +1135,176 @@ def _contact_snapshot(contact: Contact) -> dict[str, Any]: "source_payload_kind": contact.source_payload_kind, "source_payload_raw": contact.source_payload_raw, "provenance": dict(contact.provenance or {}), - "metadata": dict(contact.metadata_ or {}), + "metadata": copy.deepcopy(contact.metadata_), + "points": { + name: [_point_snapshot(point) for point in getattr(contact, name)] + for name in CONTACT_POINTS + }, } def _contact_hash(contact: Contact) -> str: - return _hash_json({**_contact_snapshot(contact), "deleted_at": contact.deleted_at.isoformat() if contact.deleted_at else None}) + return _hash_json({**_contact_snapshot(contact), "deleted_at": _deleted_at_value(contact.deleted_at)}) + + +def _point_snapshot(point: ContactEmail | ContactPhone | ContactPostalAddress) -> dict[str, Any]: + # Include every persisted evidence field, including identity, originals, + # normalization, provenance, ordering and timestamps, but never the parent FK. + return { + column.key: ( + _deleted_at_value(getattr(point, column.key)) + if column.key in {"created_at", "updated_at"} + else copy.deepcopy(getattr(point, column.key)) + ) + for column in point.__table__.columns if column.key != "contact_id" + } + + +def _lock_contact_points(session: Session, contact: Contact) -> None: + # The parent is locked by the caller. Point-only evidence updates need their + # own row locks; they need not update the parent row on PostgreSQL. + for name, (model, _identity, _replace) in CONTACT_POINTS.items(): + positions = {point.id: index for index, point in enumerate(getattr(contact, name))} + points = session.query(model).filter(model.contact_id == contact.id).order_by( + model.id.asc(), + ).populate_existing().with_for_update().all() + # Lock acquisition order is not presentation order. Preserve the + # relationship's existing order for tied indexes; new identities still + # enter the hash and therefore cannot evade the reviewed-state guard. + points.sort(key=lambda point: (point.order_index, positions.get(point.id, len(positions)), point.id)) + set_committed_value(contact, name, points) + + +def _validate_point_snapshot(value: object) -> None: + if not isinstance(value, dict) or set(value) != set(CONTACT_POINTS): + raise ValueError("Missing complete contact-point evidence") + for name, (model, _identity, _replace) in CONTACT_POINTS.items(): + points = value[name] + columns = {column.key: column for column in model.__table__.columns if column.key != "contact_id"} + ids: set[str] = set() + if not isinstance(points, list): + raise ValueError("Invalid contact-point collection") + for point in points: + if not isinstance(point, dict) or set(point) != set(columns): + raise ValueError("Incomplete contact-point evidence") + for key, column in columns.items(): + field = point[key] + if field is None and column.nullable: + continue + if key in {"created_at", "updated_at"}: + datetime.fromisoformat(field) + elif type(field) is not column.type.python_type: + raise ValueError("Invalid contact-point evidence type") + if not point["id"] or point["id"] in ids: + raise ValueError("Invalid contact-point identity") + ids.add(point["id"]) + + +def _without_contact_points(payload: ContactUpdateRequest) -> ContactUpdateRequest: + return ContactUpdateRequest.model_validate( + payload.model_dump(exclude=set(CONTACT_POINTS), exclude_unset=True) + ) + + +def _require_unreferenced_points(session: Session, contact: Contact, removed: set[str]) -> None: + if not removed: + return + # Do not detach address-list selections or silently retarget point-specific + # consent/quality decisions when an import removes or replaces a value. + referenced = session.query(AddressListEntry.id).filter( + AddressListEntry.contact_id == contact.id, + or_(AddressListEntry.contact_email_id.in_(removed), AddressListEntry.contact_postal_address_id.in_(removed)), + ).first() + for model in (ContactChannelRule, ContactPointQualityDecision): + if referenced is not None: + break + referenced = session.query(model.id).filter( + model.contact_id == contact.id, model.contact_point_id.in_(removed), + ).first() + if referenced is not None: + raise AddressBookError( + "An affected contact point has address-list or governance references; " + "review and reconcile those references before applying or rolling back this import." + ) + + +def _prepare_import_points(session: Session, contact: Contact, payload: ContactUpdateRequest): + staged = Contact( + source_kind=contact.source_kind, source_ref=contact.source_ref, + source_revision=contact.source_revision, + created_by_account_id=contact.created_by_account_id, + updated_by_account_id=contact.updated_by_account_id, + ) + result = {} + removed: set[str] = set() + for name, (_model, identity, replace) in CONTACT_POINTS.items(): + if name not in payload.model_fields_set: + continue + replace(staged, getattr(payload, name) or []) + available = list(getattr(contact, name)) + matches = [] + for candidate in list(getattr(staged, name)): + # Detach the normalization-only parent before a new point enters the + # persistent collection; otherwise save-update cascade can enlist it. + candidate.contact = None + original = next((point for point in available if all( + getattr(point, key) == getattr(candidate, key) for key in identity + )), None) + if original is not None: + available.remove(original) + matches.append((original, candidate)) + removed.update(point.id for point in available) + result[name] = matches + _require_unreferenced_points(session, contact, removed) + return result + + +def _apply_import_points(contact: Contact, prepared) -> None: + for name, matches in prepared.items(): + points = [] + for original, candidate in matches: + if original is None: + points.append(candidate) + else: + # Same value retains its exact original/normalized evidence and + # identity; only explicitly imported presentation fields change. + for field in ("label", "is_primary", "order_index"): + setattr(original, field, getattr(candidate, field)) + points.append(original) + setattr(contact, name, points) + + +def _validate_point_restoration(session: Session, contact: Contact, points: dict[str, Any]) -> None: + removed: set[str] = set() + for name, (model, _identity, _replace) in CONTACT_POINTS.items(): + desired = {point["id"] for point in points[name]} + removed.update(point.id for point in getattr(contact, name) if point.id not in desired) + # A deleted original identity must never be reclaimed from another contact. + if desired and session.query(model.id).filter( + model.id.in_(desired), model.contact_id != contact.id, + ).first() is not None: + raise AddressBookError("Contact-point identity changed; automatic rollback is unsafe.") + _require_unreferenced_points(session, contact, removed) + + +def _restore_contact_points(contact: Contact, snapshots: dict[str, Any]) -> None: + for name, (model, _identity, _replace) in CONTACT_POINTS.items(): + existing = {point.id: point for point in getattr(contact, name)} + restored = [] + for snapshot in snapshots[name]: + point = existing.get(snapshot["id"]) + if point is None: + point = model() + for key, value in snapshot.items(): + setattr(point, key, datetime.fromisoformat(value) if key in {"created_at", "updated_at"} else copy.deepcopy(value)) + restored.append(point) + setattr(contact, name, restored) + + +def _deleted_at_value(value: datetime | None) -> str | None: + if value is None: + return None + return (value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)).isoformat() def _hash_json(value: object) -> str: diff --git a/src/govoplan_addresses/backend/manifest.py b/src/govoplan_addresses/backend/manifest.py index 5e01df9..104b212 100644 --- a/src/govoplan_addresses/backend/manifest.py +++ b/src/govoplan_addresses/backend/manifest.py @@ -438,7 +438,7 @@ LDAP_PROVIDER = ExternalProviderDeclaration( manifest = ModuleManifest( id="addresses", name="Addresses", - version="0.1.22", + version="0.1.23", required_capabilities=( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, @@ -818,7 +818,18 @@ manifest = ModuleManifest( "CSV, XLSX, and LDIF files can be mapped with scoped, reusable profile versions. Each preview validates headers or attributes, " "encodings, source keys, duplicates, blank values, format limits, and contact identity before any mutation. " "The reviewed input hash and plan hash are retained with row-level effects and diagnostics. Apply is idempotent, " - "rejects contacts changed after preview, and records sufficient evidence for a guarded rollback. A persisted run " + "rejects contacts changed after preview, and records sufficient evidence for a guarded rollback. Each new update " + "uses version-2 before-images for editable contact values, source metadata, complete contact-point identities, " + "original and normalized values, provenance, order and timestamps, and prior deletion state: rolling back an " + "import that restored a deleted contact archives it again. Older update runs without complete before-images require " + "manual reconciliation; automatic rollback stops before changing any contacts. Created-contact identities and " + "after-hashes are persisted with the applied plan and required for rollback; missing older guards or contacts " + "moved to another book also require reconciliation. Point-evidence edits are included in change guards. " + "Unchanged point values retain their identities and original evidence. Removing or replacing a point referenced " + "by an address list or point-specific consent/quality decision requires explicit reconciliation, including when " + "a new reference would otherwise be detached by rollback. Current book visibility and change " + "guards still apply. Preview source lookups and contact collections are loaded in bounded batches, without changing " + "duplicate policies, reviewed hashes, or apply-time validation. A persisted run " "can be reopened with its run link after navigation or reload; previewed, applied, rolled-back, expired, and " "unavailable states remain explicit. Both apply and rollback submit the reviewed plan hash. Missing, expired, " "hidden, and cross-tenant runs disclose no source payload. XLSX formulas, macros, and legacy workbook formats " @@ -850,6 +861,20 @@ manifest = ModuleManifest( "Kontaktidentität vor jeder Änderung. Der geprüfte Eingabe- und Planhash wird mit zeilenbezogenen Wirkungen und Diagnosen " "aufbewahrt. Die Anwendung ist idempotent, verwirft seit der Vorschau geänderte Kontakte und zeichnet ausreichende " "Nachweise für eine gesicherte Rücknahme auf. Ein gespeicherter Lauf kann nach Navigation oder Neuladen über seinen Link " + "erneut geöffnet werden. Vorher-Bilder der Version 2 enthalten bearbeitbare Kontaktwerte, Quellmetadaten, " + "vollständige Kontaktpunktkennungen, Original- und normalisierte Werte, Herkunft, Reihenfolge, Zeitstempel und den " + "vorherigen Löschzustand: Die Rücknahme archiviert einen durch den Import wiederhergestellten Kontakt erneut. " + "Ältere Änderungsläufe ohne vollständige Vorher-Bilder erfordern einen manuellen Abgleich; die automatische " + "Rücknahme stoppt vor jeder Kontaktänderung. Kennungen neu angelegter Kontakte und Nachher-Hashes werden mit dem " + "angewendeten Plan gespeichert und sind für die Rücknahme erforderlich; fehlende ältere Sicherungen oder in ein " + "anderes Buch verschobene Kontakte erfordern ebenfalls einen Abgleich. Änderungen an Punktnachweisen werden vom " + "Änderungsschutz erfasst. Unveränderte Punktwerte behalten Kennung und Originalnachweise. Das Entfernen oder " + "Ersetzen eines in Adresslisten oder punktspezifischen Einwilligungs-/Qualitätsentscheidungen referenzierten " + "Punkts erfordert einen ausdrücklichen Abgleich; dies gilt auch für neue Referenzen, die eine Rücknahme sonst " + "lösen würde. Aktuelle Adressbuchsichtbarkeit und " + "Änderungsschutz bleiben wirksam. " + "Quellzuordnungen und Kontaktpunkte werden für die Vorschau in begrenzten Stapeln geladen, ohne Dublettenregeln, " + "geprüfte Hashes oder die erneute Prüfung bei Anwendung zu ändern. Ein gespeicherter Lauf kann über seinen Link " "erneut geöffnet werden; Vorschau-, Anwendungs-, Rücknahme-, Ablauf- und Nichtverfügbarkeitszustände bleiben eindeutig. " "Anwendung und Rücknahme übermitteln den geprüften Planhash. Fehlende, abgelaufene, verborgene und mandantenfremde Läufe " "legen keine Quelldaten offen. XLSX-Formeln, Makros und ältere Arbeitsmappenformate werden niemals ausgeführt oder " diff --git a/src/govoplan_addresses/backend/service.py b/src/govoplan_addresses/backend/service.py index fa2a19a..f8af11a 100644 --- a/src/govoplan_addresses/backend/service.py +++ b/src/govoplan_addresses/backend/service.py @@ -3278,8 +3278,11 @@ def _filtered_contact_query( return contact_query -def get_visible_contact(session: Session, principal: ApiPrincipal, contact_id: str, *, include_deleted: bool = False) -> Contact: - contact = _visible_contact_query(session, principal, include_deleted=include_deleted, include_deleted_books=include_deleted).filter(Contact.id == contact_id).one_or_none() +def get_visible_contact(session: Session, principal: ApiPrincipal, contact_id: str, *, include_deleted: bool = False, lock: bool = False) -> Contact: + query = _visible_contact_query(session, principal, include_deleted=include_deleted, include_deleted_books=include_deleted).filter(Contact.id == contact_id) + if lock: + query = query.populate_existing().with_for_update(of=Contact) + contact = query.one_or_none() if contact is None: raise AddressBookError("Contact not found.") return contact diff --git a/tests/test_tabular_imports.py b/tests/test_tabular_imports.py index 79d298b..4d5a5d6 100644 --- a/tests/test_tabular_imports.py +++ b/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, diff --git a/webui/package.json b/webui/package.json index 9d07e25..8718cb3 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/addresses-webui", - "version": "0.1.22", + "version": "0.1.23", "private": true, "type": "module", "main": "src/index.ts",