245 lines
9.2 KiB
Python
245 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from io import BytesIO
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from openpyxl import Workbook
|
|
|
|
from govoplan_addresses.backend.db.models import AddressBook, Contact
|
|
from govoplan_addresses.backend.import_schemas import (
|
|
AddressImportConfiguration,
|
|
AddressImportPreviewRequest,
|
|
AddressImportProfileCreateRequest,
|
|
AddressImportProfileUpdateRequest,
|
|
AddressImportRollbackRequest,
|
|
)
|
|
from govoplan_addresses.backend.imports import (
|
|
apply_address_import,
|
|
create_import_profile,
|
|
import_run_payload,
|
|
preview_address_import,
|
|
rollback_address_import,
|
|
update_import_profile,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
|
|
|
|
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",
|
|
}
|
|
|
|
|
|
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 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(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_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()
|