GovOPlaN/govoplan-addresses
2
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Release v0.1.23. Coordinated integrity review: GovOPlaN/govoplan-core#298.
This commit is contained in:
zemion
2026-09-08 12:19:36 +02:00
parent 53490e7be7
commit cf80afa42a
7 changed files with 677 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
package.json
+1 -1
View File
@@ -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",
pyproject.toml
+1 -1
View File
@@ -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"
src/govoplan_addresses/backend/imports.py
+327 -35
View File
@@ -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:
src/govoplan_addresses/backend/manifest.py
+27 -2
View File
@@ -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 "
src/govoplan_addresses/backend/service.py
+5 -2
View File
@@ -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
tests/test_tabular_imports.py
+315 -2
View File
@@ -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,
webui/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@govoplan/addresses-webui",
"version": "0.1.22",
"version": "0.1.23",
"private": true,
"type": "module",
"main": "src/index.ts",