271 lines
8.6 KiB
Python
271 lines
8.6 KiB
Python
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()
|