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

Implement address quality and reversible contact merges

Browse Source
This commit is contained in:
zemion
2026-08-02 07:03:27 +02:00
parent 2e78b9ae50
commit 19e9096572
15 changed files with 4369 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_addresses_service.py
+428 -1
View File
@@ -51,9 +51,13 @@ from govoplan_addresses.backend.db.models import (
Contact,
ContactChannelRule,
ContactEmail,
ContactFieldProvenance,
ContactMergeRecord,
ContactPhone,
ContactPointSnapshot,
ContactPointQualityDecision,
ContactPostalAddress,
ContactRedirect,
)
from govoplan_addresses.backend.schemas import (
AddressBookCreateRequest,
@@ -71,13 +75,27 @@ from govoplan_addresses.backend.schemas import (
ContactCreateRequest,
ContactChannelRuleCreateRequest,
ContactEmailPayload,
ContactMergeRecoveryRequest,
ContactMergeRequest,
ContactPhonePayload,
ContactPointQualityDecisionCreateRequest,
ContactPostalAddressPayload,
ContactPointSnapshotResponse,
ContactUpdateRequest,
)
from govoplan_addresses.backend.manifest import manifest
from govoplan_addresses.backend.router import _sync_source_response
from govoplan_addresses.backend.router import (
_sync_source_response,
api_create_address_list_entry,
api_create_contact,
api_delete_address_list_entry,
api_delete_contact,
api_restore_contact,
api_update_contact,
)
from govoplan_addresses.backend.service import (
AddressBookError,
address_quality_summary,
address_book_contact_counts,
address_list_entry_counts,
create_address_book,
@@ -86,6 +104,7 @@ from govoplan_addresses.backend.service import (
create_carddav_sync_source,
create_contact,
create_contact_channel_rule,
create_contact_quality_decision,
create_sync_source,
count_contacts,
delete_address_list_entry,
@@ -100,6 +119,8 @@ from govoplan_addresses.backend.service import (
list_address_books,
list_contacts,
list_contact_channel_rules,
list_contact_field_provenance,
list_contact_merges,
list_sync_conflicts,
list_sync_diagnostics,
list_sync_sources,
@@ -109,11 +130,16 @@ from govoplan_addresses.backend.service import (
record_sync_tombstone,
run_sync_source,
preview_sync_source,
merge_contacts,
recover_contact_merge,
restore_contact,
resolve_contact_redirect,
resolve_sync_conflict,
start_sync_attempt,
finish_sync_attempt,
update_sync_source,
update_contact,
suggest_duplicate_contacts,
resolve_trusted_deployment_carddav_credential_ref,
_carddav_client_for_source,
)
@@ -207,6 +233,10 @@ class AddressServiceTest(unittest.TestCase):
ContactPostalAddress.__table__,
ContactChannelRule.__table__,
ContactPointSnapshot.__table__,
ContactPointQualityDecision.__table__,
ContactMergeRecord.__table__,
ContactRedirect.__table__,
ContactFieldProvenance.__table__,
AddressListEntry.__table__,
AddressSyncSource.__table__,
AddressSyncTombstone.__table__,
@@ -279,6 +309,75 @@ class AddressServiceTest(unittest.TestCase):
self.session.commit()
self.assertEqual([item.id for item in list_contacts(self.session, self.principal, address_book_id=book.id)], [contact.id])
def test_contact_and_relationship_routes_emit_value_free_audit_evidence(self) -> None:
book = create_address_book(
self.session,
self.principal,
AddressBookCreateRequest(scope_type="user", name="Audited"),
)
self.session.commit()
with patch("govoplan_addresses.backend.router.audit_from_principal") as audit:
created = api_create_contact(
book.id,
ContactCreateRequest(
display_name="Ada Lovelace",
emails=[ContactEmailPayload(email="ada@example.local")],
),
self.principal,
self.session,
)
create_details = audit.call_args.kwargs["details"]
self.assertEqual(audit.call_args.kwargs["action"], "addresses.contact_created")
self.assertEqual(create_details["contact_point_counts"]["email"], 1)
self.assertNotIn("ada@example.local", repr(create_details))
original_email_id = create_details["contact_point_ids"]["email"][0]
updated = api_update_contact(
created.id,
ContactUpdateRequest(
emails=[ContactEmailPayload(email="ada.new@example.local")]
),
self.principal,
self.session,
)
update_details = audit.call_args.kwargs["details"]
self.assertEqual(audit.call_args.kwargs["action"], "addresses.contact_updated")
self.assertEqual(update_details["previous_contact_point_ids"]["email"], [original_email_id])
self.assertNotEqual(update_details["contact_point_ids"]["email"], [original_email_id])
self.assertNotIn("ada.new@example.local", repr(update_details))
api_delete_contact(created.id, self.principal, self.session)
self.assertEqual(audit.call_args.kwargs["action"], "addresses.contact_deleted")
api_restore_contact(created.id, self.principal, self.session)
self.assertEqual(audit.call_args.kwargs["action"], "addresses.contact_restored")
address_list = create_address_list(
self.session,
self.principal,
book.id,
AddressListCreateRequest(name="Audited list"),
)
self.session.commit()
entry = api_create_address_list_entry(
address_list.id,
AddressListEntryCreateRequest(
contact_id=updated.id,
contact_email_id=updated.emails[0].id,
),
self.principal,
self.session,
)
self.assertEqual(
audit.call_args.kwargs["action"],
"addresses.address_list_entry_created",
)
self.assertEqual(audit.call_args.kwargs["details"]["contact_id"], updated.id)
api_delete_address_list_entry(entry.id, self.principal, self.session)
self.assertEqual(
audit.call_args.kwargs["action"],
"addresses.address_list_entry_deleted",
)
def test_contact_windows_report_exact_totals(self) -> None:
book = create_address_book(
self.session,
@@ -605,6 +704,334 @@ END:VCARD
self.assertEqual(expired.explanations[0].code, "addresses.channel_fact.expired")
self.assertEqual(list_contact_channel_rules(self.session, self.principal, contact.id)[0].id, rule.id)
def test_contact_quality_preserves_originals_provenance_and_excludes_invalid_targets(self) -> None:
book = create_address_book(
self.session,
self.principal,
AddressBookCreateRequest(scope_type="user", name="Quality review"),
)
self.session.flush()
contact = create_contact(
self.session,
self.principal,
book.id,
ContactCreateRequest(
display_name="Ada Lovelace",
emails=[ContactEmailPayload(email=" Ada@Example.LOCAL ")],
phones=[ContactPhonePayload(phone="+49 (30) 123 45")],
postal_addresses=[
ContactPostalAddressPayload(
street=" Main Street 1 ",
postal_code=" 10115 ",
locality=" Berlin ",
country=" Germany ",
)
],
provenance={
"field_visibility": {
"organization": "restricted",
}
},
),
)
self.session.commit()
self.session.refresh(contact)
self.assertEqual(contact.emails[0].email, "Ada@Example.LOCAL")
self.assertEqual(contact.emails[0].original_email, " Ada@Example.LOCAL ")
self.assertEqual(contact.emails[0].normalized_email, "ada@example.local")
self.assertEqual(contact.phones[0].original_phone, "+49 (30) 123 45")
self.assertEqual(contact.phones[0].normalized_phone, "+493012345")
self.assertEqual(contact.postal_addresses[0].original_value["street"], " Main Street 1 ")
self.assertEqual(contact.postal_addresses[0].normalized_value["street"], "main street 1")
initial_provenance = list_contact_field_provenance(
self.session,
self.principal,
contact.id,
current_only=True,
)
self.assertTrue(any(item.field_path == "display_name" for item in initial_provenance))
self.assertTrue(any(item.field_path.endswith(".email") for item in initial_provenance))
self.assertEqual(
next(item for item in initial_provenance if item.field_path == "organization").visibility,
"restricted",
)
update_contact(
self.session,
self.principal,
contact.id,
ContactUpdateRequest(organization="Analytical Engine Office"),
)
quality = create_contact_quality_decision(
self.session,
self.principal,
contact.id,
ContactPointQualityDecisionCreateRequest(
channel="email",
contact_point_id=contact.emails[0].id,
state="undeliverable",
reason_code="addresses.quality.smtp_hard_bounce",
reason="The remote server rejected this address permanently.",
evidence_ref="mail:delivery:42",
),
)
self.session.commit()
history = list_contact_field_provenance(
self.session,
self.principal,
contact.id,
)
self.assertTrue(any(not item.selected for item in history))
current_organization = next(
item
for item in history
if item.field_path == "organization" and item.selected
)
self.assertEqual(current_organization.value, "Analytical Engine Office")
self.assertEqual(current_organization.reason_code, "addresses.contact.quality_updated")
facts = AddressesChannelFactsCapability().resolve_channel_facts(
self.session,
self.principal,
request=RecipientChannelFactsRequest(
tenant_id=self.principal.tenant_id,
source=DistributionSourceReference(
provider="addresses",
resource_type="contact",
resource_id=contact.id,
),
recipient_key=f"contact:{contact.id}",
effective_at=utcnow() + timedelta(seconds=1),
purpose="campaign_delivery",
requested_channels=("email",),
),
)
self.assertEqual(facts.candidates[0].status, "invalid")
self.assertEqual(facts.candidates[0].reason_code, "addresses.quality.smtp_hard_bounce")
self.assertEqual(facts.candidates[0].decision_provenance["quality_decision_id"], quality.id)
snapshot = AddressesRecipientSourceCapability().snapshot_address_book(
self.session,
self.principal,
address_book_id=book.id,
purpose="campaign_delivery",
)
self.assertEqual(snapshot.recipients, ())
self.assertEqual(snapshot.excluded[0].reason_code, "addresses.quality.smtp_hard_bounce")
summary = address_quality_summary(
self.session,
self.principal,
address_book_id=book.id,
)
self.assertEqual(summary.contact_count, 1)
self.assertEqual(summary.contact_point_count, 3)
self.assertEqual(summary.quality_counts["undeliverable"], 1)
self.assertEqual(summary.correction_count, 1)
self.assertEqual(summary.corrections[0].contact_id, contact.id)
def test_duplicate_merge_recovery_preserves_references_and_rejects_tampering(self) -> None:
book = create_address_book(
self.session,
self.principal,
AddressBookCreateRequest(scope_type="user", name="Duplicate review"),
)
self.session.flush()
winner = create_contact(
self.session,
self.principal,
book.id,
ContactCreateRequest(
display_name="Ada Lovelace",
organization="Analytical Engine Office",
emails=[ContactEmailPayload(email="ada@example.local")],
),
)
loser = create_contact(
self.session,
self.principal,
book.id,
ContactCreateRequest(
display_name="Ada Lovelace",
organization="Analytical Engine Office",
role_title="Mathematician",
emails=[
ContactEmailPayload(email="ADA@example.local"),
ContactEmailPayload(email="ada.private@example.local"),
],
),
)
address_list = create_address_list(
self.session,
self.principal,
book.id,
AddressListCreateRequest(name="Recipients"),
)
self.session.flush()
original_loser_email_id = loser.emails[1].id
entry = create_address_list_entry(
self.session,
self.principal,
address_list.id,
AddressListEntryCreateRequest(
contact_id=loser.id,
contact_email_id=original_loser_email_id,
),
)
create_contact_quality_decision(
self.session,
self.principal,
loser.id,
ContactPointQualityDecisionCreateRequest(
channel="email",
contact_point_id=original_loser_email_id,
state="stale",
reason="This private address needs confirmation.",
),
)
self.session.commit()
scan = suggest_duplicate_contacts(
self.session,
self.principal,
address_book_id=book.id,
)
self.assertEqual(scan.scanned_contacts, 2)
self.assertEqual(scan.candidate_pairs, 1)
self.assertEqual(scan.suggestions[0].score, 100)
self.assertEqual(scan.suggestions[0].confidence, "strong")
self.assertEqual(
{feature.code for feature in scan.suggestions[0].features},
{"email_exact", "name_organization_exact"},
)
merge = merge_contacts(
self.session,
self.principal,
ContactMergeRequest(
winner_contact_id=winner.id,
duplicate_contact_ids=[loser.id],
reason="Confirmed duplicate record.",
field_sources={"role_title": loser.id},
contact_point_strategy="union",
),
)
merge_id = merge.id
after_hash = merge.after_hash
winner_id = winner.id
loser_id = loser.id
entry_id = entry.id
self.session.commit()
self.session.expire_all()
resolved = resolve_contact_redirect(self.session, self.principal, loser_id)
self.assertTrue(resolved.redirected)
self.assertEqual(resolved.resolved_contact_id, winner_id)
merged_winner = self.session.get(Contact, winner_id)
merged_loser = self.session.get(Contact, loser_id)
assert merged_winner is not None
assert merged_loser is not None
self.assertEqual(merged_winner.role_title, "Mathematician")
self.assertEqual(
{item.normalized_email for item in merged_winner.emails},
{"ada@example.local", "ada.private@example.local"},
)
self.assertIsNotNone(merged_loser.deleted_at)
merged_entry = self.session.get(AddressListEntry, entry_id)
assert merged_entry is not None
self.assertEqual(merged_entry.contact_id, winner_id)
self.assertNotEqual(merged_entry.contact_email_id, original_loser_email_id)
self.assertTrue(
any(
item.state == "stale"
and item.contact_point_id == merged_entry.contact_email_id
for item in merged_winner.quality_decisions
)
)
retained_role_title = next(
item
for item in list_contact_field_provenance(
self.session,
self.principal,
winner_id,
current_only=True,
)
if item.field_path == "role_title"
)
self.assertEqual(retained_role_title.source_ref, f"addresses:contact:{loser_id}")
self.assertEqual(retained_role_title.metadata_["source_contact_id"], loser_id)
self.assertEqual(list_contact_merges(self.session, self.principal)[0].id, merge_id)
merged_winner.note = "Changed after merge"
self.session.commit()
with self.assertRaisesRegex(AddressBookError, "changed after this merge"):
recover_contact_merge(
self.session,
self.principal,
merge_id,
ContactMergeRecoveryRequest(
reason="Correct the duplicate decision.",
expected_after_hash=after_hash,
),
action="undo",
)
self.session.rollback()
merged_winner = self.session.get(Contact, winner_id)
assert merged_winner is not None
merged_winner.note = None
self.session.commit()
recovered = recover_contact_merge(
self.session,
self.principal,
merge_id,
ContactMergeRecoveryRequest(
reason="Correct the duplicate decision.",
expected_after_hash=after_hash,
),
action="undo",
)
self.session.commit()
self.assertEqual(recovered.status, "undone")
self.session.expire_all()
restored_winner = self.session.get(Contact, winner_id)
restored_loser = self.session.get(Contact, loser_id)
restored_entry = self.session.get(AddressListEntry, entry_id)
assert restored_winner is not None
assert restored_loser is not None
assert restored_entry is not None
self.assertIsNone(restored_winner.role_title)
self.assertIsNone(restored_loser.deleted_at)
self.assertEqual(restored_entry.contact_id, loser_id)
self.assertEqual(restored_entry.contact_email_id, original_loser_email_id)
self.assertFalse(resolve_contact_redirect(self.session, self.principal, loser_id).redirected)
second_merge = merge_contacts(
self.session,
self.principal,
ContactMergeRequest(
winner_contact_id=winner_id,
duplicate_contact_ids=[loser_id],
reason="Re-run duplicate decision.",
),
)
self.session.commit()
split = recover_contact_merge(
self.session,
self.principal,
second_merge.id,
ContactMergeRecoveryRequest(
reason="Split records after review.",
expected_after_hash=second_merge.after_hash,
),
action="split",
)
self.session.commit()
self.assertEqual(split.status, "split")
def test_address_lists_group_contacts_and_expose_recipient_sources(self) -> None:
book = create_address_book(self.session, self.principal, AddressBookCreateRequest(scope_type="user", name="Personal"))
other_book = create_address_book(self.session, self.principal, AddressBookCreateRequest(scope_type="user", name="Other"))