from __future__ import annotations import base64 import unittest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from govoplan_addresses.backend.db.models import ( AddressBook, AddressList, AddressListEntry, Contact, ) from govoplan_addresses.backend.schemas import ContactCreateRequest from govoplan_addresses.backend.service import AddressBookError, create_contact from govoplan_addresses.backend.vcard import ( MAX_VCARD_UNFOLDED_LINE_CHARS, parse_vcards_with_issues, ) from govoplan_addresses.backend.vcard_batch_schemas import ( VCardBatchCancelRequest, VCardBatchCommitRequest, VCardBatchFilePayload, VCardBatchPreviewRequest, VCardBatchSelection, VCardExportRequest, ) from govoplan_addresses.backend.vcard_batches import ( apply_vcard_batch, cancel_vcard_batch, export_vcards, preview_vcard_batch, vcard_batch_payload, ) from govoplan_core.db.base import Base class Principal: account_id = "account-1" group_ids = frozenset() @property def tenant_id(self) -> str: return "tenant-1" def has(self, scope: str) -> bool: return scope in { "addresses:address_book:read", "addresses:address_book:write", "addresses:contact:read", "addresses:contact:write", } def vcard(uid: str, name: str, email: str) -> str: return ( "BEGIN:VCARD\r\n" "VERSION:4.0\r\n" f"UID:{uid}\r\n" f"FN:{name}\r\n" f"EMAIL:{email}\r\n" "END:VCARD\r\n" ) def batch_file(filename: str, content: str) -> VCardBatchFilePayload: return VCardBatchFilePayload( filename=filename, content_base64=base64.b64encode(content.encode()).decode(), ) class VCardBatchTests(unittest.TestCase): def setUp(self) -> None: engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) self.session = sessionmaker(bind=engine, expire_on_commit=False)() self.principal = Principal() self.book = AddressBook( tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Batch contacts", source_kind="local", read_only=False, ) self.session.add(self.book) self.session.flush() def test_multifile_preview_selective_apply_and_repeat_are_idempotent(self) -> None: run = preview_vcard_batch( self.session, self.principal, self.book.id, VCardBatchPreviewRequest( files=[ batch_file( "ada.vcf", vcard("ada-1", "Ada Lovelace", "ada@example.test") ), batch_file( "grace.vcf", vcard("grace-1", "Grace Hopper", "grace@example.test"), ), ] ), ) self.assertEqual(0, self.session.query(Contact).count()) self.assertEqual("previewed", run.status) self.assertEqual(2, run.row_count) self.assertEqual("govoplan-vcard/2", run.result_evidence["parser_version"]) self.assertNotIn("ada@example.test", repr(vcard_batch_payload(run))) selections = [ VCardBatchSelection( source_key=run.plan_data[0]["source_key"], action="create" ), VCardBatchSelection( source_key=run.plan_data[1]["source_key"], action="ignore" ), ] request = VCardBatchCommitRequest( expected_plan_hash=run.plan_hash, selections=selections ) applied = apply_vcard_batch(self.session, self.principal, run.id, request) repeated = apply_vcard_batch(self.session, self.principal, run.id, request) self.assertIs(applied, repeated) self.assertEqual(1, self.session.query(Contact).count()) self.assertEqual("Ada Lovelace", self.session.query(Contact).one().display_name) self.assertEqual(1, applied.result_evidence["progress"]["ignored"]) with self.assertRaisesRegex(AddressBookError, "different selection"): apply_vcard_batch( self.session, self.principal, run.id, VCardBatchCommitRequest( expected_plan_hash=run.plan_hash, selections=[ VCardBatchSelection( source_key=run.plan_data[1]["source_key"], action="create" ) ], ), ) def test_duplicate_uid_policy_and_cancellation(self) -> None: run = preview_vcard_batch( self.session, self.principal, self.book.id, VCardBatchPreviewRequest( files=[ batch_file( "duplicates.vcf", vcard("same", "First", "first@example.test") + vcard("same", "Last", "last@example.test"), ) ], duplicate_card_policy="reject", ), ) self.assertEqual( ["conflict", "conflict"], [item["action"] for item in run.plan_data] ) cancelled = cancel_vcard_batch( self.session, self.principal, run.id, VCardBatchCancelRequest( expected_plan_hash=run.plan_hash, reason="Operator rejected duplicate source UIDs.", ), ) self.assertEqual("cancelled", cancelled.status) self.assertEqual(0, self.session.query(Contact).count()) last = preview_vcard_batch( self.session, self.principal, self.book.id, VCardBatchPreviewRequest( files=[ batch_file( "duplicates.vcf", vcard("same", "First", "first@example.test") + vcard("same", "Last", "last@example.test"), ) ], duplicate_card_policy="last", ), ) self.assertEqual( ["ignore", "create"], [item["action"] for item in last.plan_data] ) def test_deterministic_scoped_export_supports_vcard_versions(self) -> None: grace = create_contact( self.session, self.principal, self.book.id, ContactCreateRequest(display_name="Grace Hopper"), ) ada = create_contact( self.session, self.principal, self.book.id, ContactCreateRequest(display_name="Ada Lovelace"), ) address_list = AddressList( tenant_id="tenant-1", address_book_id=self.book.id, name="Selected", source_kind="local", read_only=False, ) self.session.add(address_list) self.session.flush() self.session.add( AddressListEntry( address_list_id=address_list.id, contact_id=grace.id, order_index=0 ) ) self.session.flush() selected = export_vcards( self.session, self.principal, self.book.id, VCardExportRequest( scope="contacts", contact_ids=[grace.id, ada.id], version="3.0" ), ) repeated = export_vcards( self.session, self.principal, self.book.id, VCardExportRequest( scope="contacts", contact_ids=[ada.id, grace.id], version="3.0" ), ) listed = export_vcards( self.session, self.principal, self.book.id, VCardExportRequest(scope="address_list", address_list_id=address_list.id), ) self.assertEqual(selected["content_hash"], repeated["content_hash"]) self.assertLess( selected["content"].index("Ada Lovelace"), selected["content"].index("Grace Hopper"), ) self.assertIn("VERSION:3.0", selected["content"]) self.assertEqual(1, listed["contact_count"]) self.assertIn("Grace Hopper", listed["content"]) def test_parser_rejects_pathological_unfolded_lines(self) -> None: result = parse_vcards_with_issues( "BEGIN:VCARD\nFN:" + ("a" * (MAX_VCARD_UNFOLDED_LINE_CHARS + 1)) + "\nEND:VCARD" ) self.assertEqual([], result.cards) self.assertIn("unfolded lines", result.issues[0].message) if __name__ == "__main__": unittest.main()