zemion
d4fa024034
Module Package Release / publish-packages (push) Successful in 14s
Release v0.1.23. Coordinated integrity review: GovOPlaN/govoplan-core#298.
623 lines
27 KiB
Python
623 lines
27 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from io import BytesIO
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
from openpyxl import Workbook
|
|
|
|
from govoplan_addresses.backend.db.models import AddressBook, AddressList, AddressListEntry, Contact, ContactEmail, ContactPhone, ContactPostalAddress
|
|
from govoplan_addresses.backend.import_schemas import (
|
|
AddressImportConfiguration,
|
|
AddressImportPreviewRequest,
|
|
AddressImportProfileCreateRequest,
|
|
AddressImportProfileUpdateRequest,
|
|
AddressImportRollbackRequest,
|
|
)
|
|
from govoplan_addresses.backend.imports import (
|
|
apply_address_import,
|
|
_contact_hash,
|
|
_contact_snapshot,
|
|
create_import_profile,
|
|
get_import_run,
|
|
import_run_payload,
|
|
preview_address_import,
|
|
rollback_address_import,
|
|
update_import_profile,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_addresses.backend.service import delete_contact
|
|
|
|
|
|
class Principal:
|
|
account_id = "account-1"
|
|
group_ids = frozenset({"group-1"})
|
|
|
|
@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",
|
|
}
|
|
|
|
|
|
class OtherTenantPrincipal(Principal):
|
|
@property
|
|
def tenant_id(self) -> str:
|
|
return "tenant-2"
|
|
|
|
|
|
def encoded(value: str) -> str:
|
|
return base64.b64encode(value.encode()).decode()
|
|
|
|
|
|
class AddressTabularImportTests(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="Imported contacts",
|
|
source_kind="local",
|
|
read_only=False,
|
|
)
|
|
self.session.add(self.book)
|
|
self.profile = create_import_profile(
|
|
self.session,
|
|
self.principal,
|
|
AddressImportProfileCreateRequest(
|
|
scope_type="tenant",
|
|
name="Monthly contacts",
|
|
source_format="csv",
|
|
configuration=AddressImportConfiguration(
|
|
delimiter=";",
|
|
field_mappings={
|
|
"source_key": "id",
|
|
"given_name": "first",
|
|
"family_name": "last",
|
|
"email": "email",
|
|
"organization": "organization",
|
|
},
|
|
),
|
|
),
|
|
)
|
|
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,
|
|
filename="contacts.csv",
|
|
content_base64=encoded(
|
|
"id;first;last;email;organization\n"
|
|
"1;Ada;Lovelace;ada@example.test;Analysis Office\n"
|
|
"2;Grace;Hopper;grace@example.test;Computing Office\n"
|
|
),
|
|
)
|
|
run = preview_address_import(self.session, self.principal, self.book.id, payload)
|
|
self.assertEqual(2, run.statistics["create"])
|
|
self.assertFalse(run.diagnostics)
|
|
|
|
applied = apply_address_import(
|
|
self.session,
|
|
self.principal,
|
|
run.id,
|
|
expected_plan_hash=run.plan_hash,
|
|
)
|
|
self.assertEqual("applied", applied.status)
|
|
self.assertEqual(2, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())
|
|
self.assertIs(applied, apply_address_import(self.session, self.principal, run.id, expected_plan_hash=run.plan_hash))
|
|
response_evidence = import_run_payload(applied)["result_evidence"]
|
|
self.assertEqual(2, response_evidence["created_contact_count"])
|
|
self.assertNotIn("created_contact_ids", response_evidence)
|
|
self.assertNotIn("updated_contacts", response_evidence)
|
|
|
|
repeated = preview_address_import(self.session, self.principal, self.book.id, payload)
|
|
self.assertEqual(2, repeated.statistics["unchanged"])
|
|
apply_address_import(self.session, self.principal, repeated.id, expected_plan_hash=repeated.plan_hash)
|
|
self.assertEqual(2, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())
|
|
|
|
rolled_back = rollback_address_import(
|
|
self.session,
|
|
self.principal,
|
|
run.id,
|
|
AddressImportRollbackRequest(
|
|
expected_plan_hash=run.plan_hash,
|
|
reason="The operator selected the wrong monthly file.",
|
|
),
|
|
)
|
|
self.assertEqual("rolled_back", rolled_back.status)
|
|
self.assertEqual(0, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count())
|
|
|
|
def test_rollback_rejects_a_stale_review_hash(self) -> None:
|
|
run = preview_address_import(
|
|
self.session,
|
|
self.principal,
|
|
self.book.id,
|
|
AddressImportPreviewRequest(
|
|
profile_id=self.profile.id,
|
|
filename="contacts.csv",
|
|
content_base64=encoded(
|
|
"id;first;last;email;organization\n"
|
|
"1;Ada;Lovelace;ada@example.test;Analysis Office\n"
|
|
),
|
|
),
|
|
)
|
|
apply_address_import(
|
|
self.session,
|
|
self.principal,
|
|
run.id,
|
|
expected_plan_hash=run.plan_hash,
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "reviewed import plan changed"):
|
|
rollback_address_import(
|
|
self.session,
|
|
self.principal,
|
|
run.id,
|
|
AddressImportRollbackRequest(
|
|
expected_plan_hash="0" * 64,
|
|
reason="The operator selected the wrong monthly file.",
|
|
),
|
|
)
|
|
|
|
def test_persisted_run_read_is_tenant_bounded_and_source_safe(self) -> None:
|
|
source = "id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;Analysis Office\n"
|
|
run = preview_address_import(
|
|
self.session,
|
|
self.principal,
|
|
self.book.id,
|
|
AddressImportPreviewRequest(
|
|
profile_id=self.profile.id,
|
|
filename="contacts.csv",
|
|
content_base64=encoded(source),
|
|
),
|
|
)
|
|
self.session.flush()
|
|
|
|
payload = import_run_payload(get_import_run(self.session, self.principal, run.id))
|
|
self.assertEqual("previewed", payload["status"])
|
|
self.assertEqual(run.plan_hash, payload["plan_hash"])
|
|
self.assertNotIn("plan_data", payload)
|
|
self.assertNotIn(source, repr(payload))
|
|
|
|
with self.assertRaisesRegex(ValueError, "not found"):
|
|
get_import_run(self.session, OtherTenantPrincipal(), run.id)
|
|
|
|
def test_duplicate_keys_and_changed_targets_block_apply(self) -> None:
|
|
duplicate = preview_address_import(
|
|
self.session,
|
|
self.principal,
|
|
self.book.id,
|
|
AddressImportPreviewRequest(
|
|
profile_id=self.profile.id,
|
|
filename="duplicates.csv",
|
|
content_base64=encoded(
|
|
"id;first;last;email;organization\n"
|
|
"1;Ada;Lovelace;ada@example.test;One\n"
|
|
"1;Ada;Lovelace;ada@example.test;Two\n"
|
|
),
|
|
),
|
|
)
|
|
self.assertEqual(2, duplicate.statistics["conflict"])
|
|
with self.assertRaisesRegex(ValueError, "error diagnostics"):
|
|
apply_address_import(self.session, self.principal, duplicate.id, expected_plan_hash=duplicate.plan_hash)
|
|
|
|
initial = preview_address_import(
|
|
self.session,
|
|
self.principal,
|
|
self.book.id,
|
|
AddressImportPreviewRequest(
|
|
profile_id=self.profile.id,
|
|
filename="one.csv",
|
|
content_base64=encoded("id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;One\n"),
|
|
),
|
|
)
|
|
apply_address_import(self.session, self.principal, initial.id, expected_plan_hash=initial.plan_hash)
|
|
changed = preview_address_import(
|
|
self.session,
|
|
self.principal,
|
|
self.book.id,
|
|
AddressImportPreviewRequest(
|
|
profile_id=self.profile.id,
|
|
filename="one.csv",
|
|
content_base64=encoded("id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;Two\n"),
|
|
),
|
|
)
|
|
contact = self.session.query(Contact).filter(Contact.deleted_at.is_(None)).one()
|
|
contact.organization = "Concurrent edit"
|
|
self.session.flush()
|
|
with self.assertRaisesRegex(ValueError, "changed after preview"):
|
|
apply_address_import(self.session, self.principal, changed.id, expected_plan_hash=changed.plan_hash)
|
|
|
|
def test_profile_updates_create_immutable_versions(self) -> None:
|
|
next_profile = update_import_profile(
|
|
self.session,
|
|
self.principal,
|
|
self.profile.id,
|
|
payload=AddressImportProfileUpdateRequest(name="Monthly contacts v2"),
|
|
)
|
|
self.assertFalse(self.profile.is_current)
|
|
self.assertTrue(next_profile.is_current)
|
|
self.assertEqual(self.profile.profile_key, next_profile.profile_key)
|
|
self.assertEqual(2, next_profile.version)
|
|
|
|
def test_xlsx_sheet_selection_and_formula_rejection(self) -> None:
|
|
workbook = Workbook()
|
|
workbook.active.title = "Ignore"
|
|
sheet = workbook.create_sheet("Contacts")
|
|
sheet.append(["id", "first", "last", "email", "organization"])
|
|
sheet.append(["1", "Ada", "Lovelace", "ada@example.test", "Analysis Office"])
|
|
content = BytesIO()
|
|
workbook.save(content)
|
|
xlsx_profile = create_import_profile(
|
|
self.session,
|
|
self.principal,
|
|
AddressImportProfileCreateRequest(
|
|
scope_type="tenant",
|
|
name="Workbook contacts",
|
|
source_format="xlsx",
|
|
configuration=AddressImportConfiguration(
|
|
sheet_name="Contacts",
|
|
field_mappings={
|
|
"source_key": "id",
|
|
"given_name": "first",
|
|
"family_name": "last",
|
|
"email": "email",
|
|
"organization": "organization",
|
|
},
|
|
),
|
|
),
|
|
)
|
|
self.session.flush()
|
|
run = preview_address_import(
|
|
self.session,
|
|
self.principal,
|
|
self.book.id,
|
|
AddressImportPreviewRequest(
|
|
profile_id=xlsx_profile.id,
|
|
filename="contacts.xlsx",
|
|
content_base64=base64.b64encode(content.getvalue()).decode(),
|
|
),
|
|
)
|
|
self.assertEqual(1, run.statistics["create"])
|
|
|
|
sheet["E2"] = "=CONCAT(\"Analysis\", \" Office\")"
|
|
content = BytesIO()
|
|
workbook.save(content)
|
|
with self.assertRaisesRegex(ValueError, "formulas are never evaluated"):
|
|
preview_address_import(
|
|
self.session,
|
|
self.principal,
|
|
self.book.id,
|
|
AddressImportPreviewRequest(
|
|
profile_id=xlsx_profile.id,
|
|
filename="contacts.xlsx",
|
|
content_base64=base64.b64encode(content.getvalue()).decode(),
|
|
),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|