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

Add governed contact point snapshots

Browse Source
This commit is contained in:
zemion
2026-08-02 06:20:24 +02:00
parent 67392f620f
commit 2e78b9ae50
10 changed files with 1620 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_addresses_service.py
+178
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import unittest
from dataclasses import asdict
from datetime import timedelta
from unittest.mock import patch
@@ -13,6 +14,12 @@ from govoplan_core.core.distribution_lists import (
DistributionSourceReference,
RecipientChannelFactsRequest,
)
from govoplan_core.core.contact_points import (
CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION,
CONTACT_POINT_CONTRACT_VERSION,
ContactPointResolutionRequest,
ContactPointSourceRequest,
)
from govoplan_core.core.people import CAPABILITY_ADDRESSES_PEOPLE_SEARCH, PeopleSearchProvider
from govoplan_core.db.base import Base
from govoplan_core.db.base import utcnow
@@ -26,6 +33,7 @@ from govoplan_addresses.backend.capabilities import (
CAPABILITY_ADDRESSES_LOOKUP,
CAPABILITY_ADDRESSES_RECIPIENT_SOURCE,
AddressesChannelFactsCapability,
AddressesContactPointResolutionCapability,
AddressesContactWriterCapability,
AddressesLookupCapability,
AddressesPeopleSearchProvider,
@@ -44,6 +52,7 @@ from govoplan_addresses.backend.db.models import (
ContactChannelRule,
ContactEmail,
ContactPhone,
ContactPointSnapshot,
ContactPostalAddress,
)
from govoplan_addresses.backend.schemas import (
@@ -63,6 +72,7 @@ from govoplan_addresses.backend.schemas import (
ContactChannelRuleCreateRequest,
ContactEmailPayload,
ContactPostalAddressPayload,
ContactPointSnapshotResponse,
)
from govoplan_addresses.backend.manifest import manifest
from govoplan_addresses.backend.router import _sync_source_response
@@ -196,6 +206,7 @@ class AddressServiceTest(unittest.TestCase):
ContactPhone.__table__,
ContactPostalAddress.__table__,
ContactChannelRule.__table__,
ContactPointSnapshot.__table__,
AddressListEntry.__table__,
AddressSyncSource.__table__,
AddressSyncTombstone.__table__,
@@ -397,6 +408,11 @@ END:VCARD
self.assertIn(CAPABILITY_ADDRESSES_RECIPIENT_SOURCE, provided)
self.assertIn(CAPABILITY_ADDRESSES_CONTACT_WRITER, provided)
self.assertIn(CAPABILITY_RECIPIENT_CHANNEL_FACTS, provided)
self.assertIn(CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION, provided)
self.assertIn(
CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION,
manifest.capability_factories,
)
book = create_address_book(self.session, self.principal, AddressBookCreateRequest(scope_type="user", name="Recipients"))
self.session.commit()
@@ -691,6 +707,168 @@ END:VCARD
self.session.commit()
self.assertEqual(list_address_list_entries(self.session, self.principal, address_list.id), [])
def test_contact_point_resolution_supports_external_refs_and_frozen_postal_snapshots(self) -> None:
book = create_address_book(
self.session,
self.principal,
AddressBookCreateRequest(scope_type="user", name="Official contacts"),
)
self.session.commit()
self.session.refresh(book)
contact = create_contact(
self.session,
self.principal,
book.id,
ContactCreateRequest(
display_name="Ada Lovelace",
emails=[
ContactEmailPayload(
label="private",
email="ada.private@example.local",
is_primary=True,
)
],
postal_addresses=[
ContactPostalAddressPayload(
label="official",
street="Main Street 1",
postal_code="10115",
locality="Berlin",
country="Germany",
is_primary=True,
),
ContactPostalAddressPayload(
label="private",
street="Side Street 2",
postal_code="10117",
locality="Berlin",
country="Germany",
),
],
),
)
contact.source_kind = "idm"
contact.source_ref = "idm:identity:identity-1"
self.session.flush()
create_contact_channel_rule(
self.session,
self.principal,
contact.id,
ContactChannelRuleCreateRequest(
channel="postal",
purpose="official_notice",
contact_point_id=contact.postal_addresses[0].id,
decision="preferred",
legal_basis="public_task",
evidence_ref="idm:function-assignment:17",
preference_rank=1,
locale="de-DE",
),
)
address_list = create_address_list(
self.session,
self.principal,
book.id,
AddressListCreateRequest(name="Postal recipients"),
)
self.session.flush()
postal_entry = create_address_list_entry(
self.session,
self.principal,
address_list.id,
AddressListEntryCreateRequest(
contact_id=contact.id,
contact_postal_address_id=contact.postal_addresses[0].id,
),
)
self.session.commit()
capability = AddressesContactPointResolutionCapability()
direct = capability.resolve_contact_points(
self.session,
self.principal,
request=ContactPointResolutionRequest(
tenant_id=self.principal.tenant_id,
subject=DistributionSourceReference(
provider="idm",
resource_type="identity",
resource_id="identity-1",
),
effective_at=utcnow(),
purpose="official_notice",
requested_channels=("postal",),
address_purpose="official",
fallback_rule="none",
locale="de-DE",
postal_format="international",
),
)
self.assertEqual(CONTACT_POINT_CONTRACT_VERSION, direct.contract_version)
self.assertEqual("usable", direct.status)
self.assertEqual(contact.id, direct.contact_id)
self.assertEqual(1, len(direct.candidates))
self.assertEqual(contact.postal_addresses[0].id, direct.candidates[0].contact_point_id)
self.assertEqual("official", direct.candidates[0].address_purpose)
self.assertIn("Ada Lovelace", direct.candidates[0].target)
self.assertIn("Germany", direct.candidates[0].target)
self.assertEqual("de-DE", direct.candidates[0].locale)
self.assertTrue(direct.candidates[0].preference_revision)
self.assertEqual(1, len(direct.excluded))
self.assertEqual(
"addresses.address_purpose.not_selected",
direct.excluded[0].reason_code,
)
source_request = ContactPointSourceRequest(
tenant_id=self.principal.tenant_id,
source_id=f"addresses:address_list:{address_list.id}",
effective_at=utcnow(),
purpose="official_notice",
requested_channels=("email", "postal"),
address_purpose="official",
fallback_rule="none",
locale="de-DE",
postal_format="international",
)
preview = capability.preview_source(
self.session,
self.principal,
request=source_request,
limit=1,
)
self.assertEqual(1, preview.total_count)
self.assertEqual(1, preview.usable_count)
self.assertFalse(preview.has_more)
self.assertEqual(postal_entry.id, preview.resolutions[0].provenance["address_list_entry_ids"][0])
self.assertEqual("postal", preview.resolutions[0].candidates[0].channel)
snapshot = capability.freeze_source(
self.session,
self.principal,
request=source_request,
)
self.session.commit()
original_target = snapshot.resolutions[0].candidates[0].target
contact.postal_addresses[0].street = "Changed Street 99"
contact.postal_addresses[0].country = "France"
self.session.commit()
frozen = capability.get_snapshot(
self.session,
self.principal,
snapshot_id=snapshot.id,
)
self.assertIsNotNone(frozen)
assert frozen is not None
self.assertEqual(original_target, frozen.resolutions[0].candidates[0].target)
self.assertIn("Main Street 1", frozen.resolutions[0].candidates[0].target)
self.assertNotIn("Changed Street 99", frozen.resolutions[0].candidates[0].target)
self.assertEqual(snapshot.snapshot_hash, frozen.snapshot_hash)
self.assertEqual(1, frozen.recipient_count)
response = ContactPointSnapshotResponse.model_validate(asdict(frozen))
self.assertEqual(snapshot.id, response.id)
self.assertEqual("postal", response.resolutions[0].candidates[0].channel)
def test_sync_source_marks_read_only_books_and_can_be_made_writable(self) -> None:
book = create_address_book(self.session, self.principal, AddressBookCreateRequest(scope_type="user", name="CardDAV"))
self.session.commit()