"""Additive adversarial Phase 2B integration and freeze checks.""" from datetime import date, timedelta from unittest.mock import patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.core.exceptions import ValidationError from django.db import IntegrityError, transaction from django.db.models.deletion import Collector, ProtectedError from django.test import Client, TransactionTestCase from django.utils import timezone from apps.access.authorization import is_authorized from apps.customers import services as customers, queries as customer_queries from apps.customers.models import Customer, CustomerAddress, CustomerContact, CustomerNumberSequence from apps.customers.tests import create from apps.customers.test_details import address, contact from apps.devices import queries from apps.devices.models import Device, DeviceIdentifier, DevicePurchaseEvidence as Purchase, DeviceWarrantyCoverage as Coverage, CustomerDeviceRelationship as Ownership from apps.devices.test_evidence import coverage from apps.organization import test_assignment_concurrency as concurrency from django.contrib.auth import get_user_model from django.test import TestCase from django.urls import reverse from apps.devices import services as devices from apps.devices.test_relationships import setup from apps.devices.tests import synthetic_imei class IdentifierAdminAuditTests(TestCase): def setUp(self): setup(self) self.user = get_user_model().objects.create_superuser(username="synthetic-audit-admin") self.client.force_login(self.user) devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(30)) self.url = reverse("admin:devices_device_change", args=[self.device.pk]) def test_stale_explicit_correction_cannot_replace_newer_identifier(self): page = self.client.get(self.url) token = page.context["adminform"].form.initial.get("identifier_revision", "") devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(31)) self.client.post(self.url, {"identifier_revision": token, "correction_type": "IMEI1", "correction_value": synthetic_imei(32), "_save": "Save"}) self.assertEqual(self.device.identifiers.get(is_active=True).normalized_value, synthetic_imei(31)) self.assertEqual(self.device.identifiers.count(), 2) def test_correction_without_revision_is_rejected(self): self.client.post(self.url, {"correction_type": "IMEI1", "correction_value": synthetic_imei(32), "_save": "Save"}) self.assertEqual(self.device.identifiers.get(is_active=True).normalized_value, synthetic_imei(30)) def test_fresh_revision_allows_correction(self): token = self.client.get(self.url).context["adminform"].form.initial["identifier_revision"] response = self.client.post(self.url, {"identifier_revision": token, "correction_type": "IMEI1", "correction_value": synthetic_imei(31), "_save": "Save"}) self.assertEqual(response.status_code, 302) self.assertEqual(self.device.identifiers.get(is_active=True).normalized_value, synthetic_imei(31)) def test_tampered_revision_rejected(self): self.client.post(self.url, {"identifier_revision": "tampered", "correction_type": "IMEI1", "correction_value": synthetic_imei(31), "_save": "Save"}) self.assertEqual(self.device.identifiers.count(), 1) def test_service_precondition_detects_restore_of_same_identifier_row(self): row = self.device.identifiers.get(is_active=True) revision = devices._identifier_revision(row) devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(31)) devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(30)) with self.assertRaises(ValidationError): devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(32), expected_identifier_revision=revision) self.assertEqual(self.device.identifiers.get(is_active=True).pk, row.pk) def test_admin_service_precondition_closes_post_validation_race(self): token = self.client.get(self.url).context["adminform"].form.initial["identifier_revision"] original = devices.replace_device_identifier def interleave(**kwargs): original(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(31)) return original(**kwargs) with patch("apps.devices.admin.services.replace_device_identifier", side_effect=interleave): response = self.client.post(self.url, {"identifier_revision": token, "correction_type": "IMEI1", "correction_value": synthetic_imei(32), "_save": "Save"}) self.assertEqual(response.status_code, 302) # Injected operation shares the outer Admin transaction, so both roll back. self.assertEqual(self.device.identifiers.count(), 1) def test_form_revision_matches_the_identifiers_displayed_on_page(self): from apps.devices.admin import DeviceForm displayed = Device.objects.prefetch_related("identifiers").get(pk=self.device.pk) old = next(row for row in displayed.identifiers.all() if row.is_active) devices.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(31)) form = DeviceForm(instance=displayed) self.assertEqual(form.identifier_snapshot["IMEI1"], devices._identifier_revision(old)) class Phase2BIntegrityAuditTests(TestCase): def setUp(self): setup(self) self.user = get_user_model().objects.create_user(username="synthetic-auditor") def test_two_company_four_customer_isolation_matrix(self): b2 = create(self.other_company) owners = [self.a, self.b, self.outsider, b2] units = [] for index, owner in enumerate(owners): unit = devices.register_device(product_model=self.model, serial=f"SYNTHETIC-AUDIT-{index}") devices.assign_device_owner(device=unit, customer=owner) units.append(unit) for company, expected_customers, expected_devices in [ (self.company, [self.a, self.b, self.c], units[:2]), (self.other_company, [self.outsider, b2], units[2:]), ]: with self.assertNumQueries(0): cs = customer_queries.active_customers_for_company(company) ds = queries.currently_owned_devices_for_company(company) self.assertCountEqual(cs, expected_customers) self.assertCountEqual(ds, expected_devices) self.assertCountEqual(customer_queries.find_customers(company=company, query="Synthetic"), expected_customers) for owner, unit in zip(owners, units): self.assertEqual(list(queries.currently_owned_devices(owner)), [unit]) for foreign in [candidate for candidate in owners if candidate.company_id != owner.company_id]: with self.assertRaises(ValidationError): devices.transfer_device_ownership(device=unit, new_customer=foreign) self.assertEqual(queries.find_device_by_identifier("SYNTHETIC-AUDIT-2"), units[2]) def test_stale_forged_customer_company_cannot_bypass_affinity(self): devices.assign_device_owner(device=self.device, customer=self.a) self.outsider.company = self.company with self.assertRaises(ValidationError): devices.transfer_device_ownership(device=self.device, new_customer=self.outsider) devices.end_device_ownership(device=self.device) with self.assertRaises(ValidationError): devices.assign_device_owner(device=self.device, customer=self.outsider) def test_corrupted_counter_fails_closed_without_duplicate_or_advance(self): CustomerNumberSequence.objects.filter(company=self.company).update(next_value=1) before = Customer.objects.filter(company=self.company).count() with self.assertRaises(ValidationError): create(self.company) self.assertEqual(Customer.objects.filter(company=self.company).count(), before) self.assertEqual(CustomerNumberSequence.objects.get(company=self.company).next_value, 1) def test_counter_exhaustion_rolls_back(self): from apps.customers.models import MAX_NUMBER CustomerNumberSequence.objects.filter(company=self.company).update(next_value=MAX_NUMBER+1) with self.assertRaises(ValidationError): create(self.company) self.assertEqual(CustomerNumberSequence.objects.get(company=self.company).next_value, MAX_NUMBER+1) def test_mobile_email_other_duplicates_and_quick_fields_independence(self): for kind, value, equivalent in [("MOBILE", "+1 (202) 555-0140", "+12025550140"), ("EMAIL", "Audit@EXAMPLE.INVALID", "Audit@example.invalid"), ("OTHER", " SYNTHETIC-REFERENCE ", "SYNTHETIC-REFERENCE")]: first = customers.create_customer_contact(customer=self.a, contact_type=kind, value=value, is_primary=True) with self.assertRaises(ValidationError): customers.create_customer_contact(customer=self.a, contact_type=kind, value=equivalent) other = customers.create_customer_contact(customer=self.outsider, contact_type=kind, value=equivalent, is_primary=True) self.assertEqual(list(customer_queries.active_contacts_for_customer(self.a, contact_type=kind)), [first]) self.assertEqual(other.normalized_value, first.normalized_value) self.a.refresh_from_db() self.assertEqual((self.a.primary_mobile, self.a.primary_email), ("", "")) def test_postgres_adjacent_closed_and_open_intervals(self): start = timezone.now()-timedelta(days=30) first = Ownership(device=self.device, customer=self.a, started_at=start, ended_at=start+timedelta(days=10)) second = Ownership(device=self.device, customer=self.b, started_at=first.ended_at, ended_at=start+timedelta(days=20)) current = Ownership(device=self.device, customer=self.a, started_at=second.ended_at) Ownership.objects.bulk_create([first, second, current]) for lower, upper in [(start+timedelta(days=5), start+timedelta(days=15)), (start+timedelta(days=21), start+timedelta(days=22))]: with self.assertRaises(IntegrityError), transaction.atomic(): Ownership.objects.bulk_create([Ownership(device=self.device, customer=self.b, started_at=lower, ended_at=upper)]) self.assertEqual(Ownership.objects.count(), 3) def test_raw_affinity_bypass_is_application_only_and_documented(self): devices.assign_device_owner(device=self.device, customer=self.a, started_at=self.start) devices.end_device_ownership(device=self.device, ended_at=self.start+timedelta(days=1)) # Deliberately trusted-writer bypass: DB constraints cannot join Customer.company. Ownership.objects.bulk_create([Ownership(device=self.device, customer=self.outsider)]) self.assertEqual(queries.current_device_owner(self.device), self.outsider) with self.assertRaises(ValidationError): devices.transfer_device_ownership(device=self.device, new_customer=self.b) def test_all_protected_registry_records_survive_deletion_attempts(self): devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-DELETE-AUDIT") purchase = devices.set_device_purchase_evidence(device=self.device) warranty = coverage(self.device) owner = devices.assign_device_owner(device=self.device, customer=self.a) for row in [self.device, self.device.identifiers.get(), purchase, warranty, owner]: with self.subTest(model=type(row).__name__), self.assertRaises(ValidationError): type(row).objects.filter(pk=row.pk).delete() for row in [self.a, self.device]: with self.assertRaises(ProtectedError): Collector(using="default").collect([row]) def test_unreferenced_customer_and_details_deliberately_allow_orm_delete(self): record = address(self.c) with self.assertRaises(ProtectedError): self.c.delete() record.delete() number = self.c.customer_number self.c.delete() self.assertNotEqual(create(self.company).customer_number, number) def test_five_warranty_boundary_dates(self): warranty = coverage(self.device) start, end = warranty.coverage_start_date, warranty.coverage_end_date for value, expected in [(start-timedelta(days=1), False), (start, True), (start+(end-start)//2, True), (end, True), (end+timedelta(days=1), False)]: with self.subTest(date=value): self.assertEqual(queries.device_has_recorded_warranty_coverage(device=self.device, on_date=value), expected) def test_ownership_preserves_all_evidence_and_coverage_columns(self): evidence = devices.set_device_purchase_evidence(device=self.device, purchase_date=date(2025, 1, 1), seller_name="Synthetic Seller", seller_reference="SYNTHETIC", invoice_number="SYNTHETIC-INVOICE") devices.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user, note="Synthetic review") coverage(self.device) before = (list(Purchase.objects.values()), list(Coverage.objects.values())) devices.assign_device_owner(device=self.device, customer=self.a) devices.transfer_device_ownership(device=self.device, new_customer=self.b) devices.end_device_ownership(device=self.device) self.assertEqual((list(Purchase.objects.values()), list(Coverage.objects.values())), before) def test_material_edit_and_rejection_preserve_independent_coverage(self): for source in ["MANUFACTURER_RECORD", "MANUAL_OVERRIDE"]: evidence = devices.set_device_purchase_evidence(device=self.device, invoice_number=source) devices.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) current = coverage(self.device, coverage_source=source, note="Synthetic justification") snapshot = Coverage.objects.filter(pk=current.pk).values().get() evidence = devices.set_device_purchase_evidence(device=self.device, seller_name=source) devices.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) self.assertEqual(Coverage.objects.filter(pk=current.pk).values().get(), snapshot) def test_rollback_matrix_restores_complete_database_state(self): addr1, addr2 = address(self.a, is_primary=True), address(self.a) evidence = devices.set_device_purchase_evidence(device=self.device) coverage(self.device) owner = devices.assign_device_owner(device=self.device, customer=self.a) models = [Customer, CustomerNumberSequence, CustomerAddress, Device, DeviceIdentifier, Purchase, Coverage, Ownership] def snapshot(): return [list(model.objects.order_by("pk").values()) for model in models] before = snapshot() operations = [lambda: create(self.company), lambda: customers.set_primary_customer_address(address=addr2), lambda: devices.register_device(product_model=self.model, serial="SYNTHETIC-ROLLBACK"), lambda: devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-ROLLBACK"), lambda: devices.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user), lambda: devices.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-ROLLBACK"), lambda: coverage(self.device), lambda: devices.transfer_device_ownership(device=self.device, new_customer=self.b), lambda: devices.end_device_ownership(device=self.device)] for index, operation in enumerate(operations): with self.subTest(operation=index): with self.assertRaises(RuntimeError), transaction.atomic(): operation() raise RuntimeError("Synthetic audit rollback") self.assertEqual(snapshot(), before) def test_staff_group_and_direct_permissions_do_not_create_business_scope(self): self.user.is_staff = True self.user.save(update_fields=["is_staff"]) group = Group.objects.create(name="synthetic-audit-permissions") for record in [self.a, self.device, devices.assign_device_owner(device=self.device, customer=self.a)]: permission = Permission.objects.get(content_type__app_label=record._meta.app_label, codename="view_"+record._meta.model_name) self.user.user_permissions.add(permission) group.permissions.add(permission) self.user.groups.add(group) self.assertFalse(is_authorized(user=self.user, permission=f"{record._meta.app_label}.{permission.codename}", target=record)) def test_all_phase2b_admin_posts_require_csrf(self): self.user.is_staff = self.user.is_superuser = True self.user.save(update_fields=["is_staff", "is_superuser"]) client = Client(enforce_csrf_checks=True) client.force_login(self.user) for model in [Customer, CustomerAddress, CustomerContact, Device, DeviceIdentifier, Purchase, Coverage, Ownership]: url = reverse(f"admin:{model._meta.app_label}_{model._meta.model_name}_add") with self.subTest(model=model.__name__): self.assertEqual(client.post(url, {}).status_code, 403) def test_owned_device_list_and_history_have_bounded_queries(self): for _ in range(4): unit = devices.register_device(product_model=self.model) devices.assign_device_owner(device=unit, customer=self.a) with self.assertNumQueries(1): self.assertEqual(len([str(unit) for unit in queries.currently_owned_devices(self.a)]), 4) with self.assertNumQueries(1): self.assertEqual(len(list(customer_queries.find_customers(company=self.company, query="Synthetic"))), 3) def test_purchase_admin_reviewer_display_avoids_n_plus_one(self): from django.contrib import admin from django.test import RequestFactory for _ in range(4): unit = devices.register_device(product_model=self.model) evidence = devices.set_device_purchase_evidence(device=unit) devices.verify_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) request = RequestFactory().get("/admin/devices/devicepurchaseevidence/") self.user.is_staff = self.user.is_superuser = True request.user = self.user rows = admin.site._registry[Purchase].get_changelist_instance(request).queryset with self.assertNumQueries(1): self.assertEqual([row.reviewed_by.username for row in rows], [self.user.username] * 4) class Phase2BConcurrencyAuditTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): setup(self) self.user = get_user_model().objects.create_user(username="synthetic-race-auditor") def test_ownership_transfer_then_identifier_correction(self): devices.assign_device_owner(device=self.device, customer=self.a) self.run_concurrent(lambda: devices.transfer_device_ownership(device=self.device, new_customer=self.b), lambda: devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-RACE"), expected="success") self.assertEqual(queries.current_device_owner(self.device), self.b) self.assertEqual(queries.find_device_by_identifier("SYNTHETIC-RACE"), self.device) def test_identifier_correction_then_ownership_transfer(self): devices.assign_device_owner(device=self.device, customer=self.a) self.run_concurrent(lambda: devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-RACE"), lambda: devices.transfer_device_ownership(device=self.device, new_customer=self.b), expected="success") self.assertEqual(queries.current_device_owner(self.device), self.b) def test_identifier_correction_then_stale_admin_explicit_correction(self): self.user.is_staff = self.user.is_superuser = True self.user.save(update_fields=["is_staff", "is_superuser"]) devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-OLD") client = Client(); client.force_login(self.user) url = reverse("admin:devices_device_change", args=[self.device.pk]) token = client.get(url).context["adminform"].form.initial["identifier_revision"] def stale_post(): response = client.post(url, {"identifier_revision": token, "correction_type": "SERIAL", "correction_value": "SYNTHETIC-STALE", "_save": "Save"}) self.assertEqual(response.status_code, 302) self.run_concurrent(lambda: devices.replace_device_identifier(device=self.device, identifier_type="SERIAL", new_value="SYNTHETIC-NEW"), stale_post, expected="success") self.assertEqual(self.device.identifiers.get(is_active=True).value, "SYNTHETIC-NEW") self.assertEqual(self.device.identifiers.count(), 2) def test_simultaneous_active_primary_address_creation(self): self.run_concurrent(lambda: address(self.a, is_primary=True), lambda: address(self.a, is_primary=True), expected="success") self.assertEqual(CustomerAddress.objects.filter(customer=self.a, is_primary=True).count(), 1) self.assertEqual(CustomerAddress.objects.filter(customer=self.a).count(), 2) def test_email_duplicate_creation(self): self.run_concurrent(lambda: customers.create_customer_contact(customer=self.a, contact_type="EMAIL", value="Audit@EXAMPLE.INVALID"), lambda: customers.create_customer_contact(customer=self.a, contact_type="EMAIL", value="Audit@example.invalid"), expected="validation") def test_raw_temporal_overlap_race(self): def insert(customer): Ownership.objects.bulk_create([Ownership(device=self.device, customer=customer, started_at=self.start, ended_at=self.start+timedelta(days=1))]) self.run_concurrent(lambda: insert(self.a), lambda: insert(self.b), expected="integrity") self.assertEqual(Ownership.objects.count(), 1) def test_material_edit_then_manufacturer_coverage(self): devices.set_device_purchase_evidence(device=self.device) self.run_concurrent(lambda: devices.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-EDIT"), lambda: coverage(self.device), expected="success") self.assertIsNotNone(queries.current_warranty_coverage(self.device)) def test_manufacturer_coverage_then_material_edit(self): devices.set_device_purchase_evidence(device=self.device) self.run_concurrent(lambda: coverage(self.device), lambda: devices.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-EDIT"), expected="success") self.assertIsNotNone(queries.current_warranty_coverage(self.device))