"""Synthetic purchase review and recorded-coverage integrity tests.""" import uuid from datetime import date, datetime, timedelta from unittest.mock import patch from django.contrib import admin from django.contrib.auth import get_user_model from django.contrib.auth.models import 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, TestCase from django.urls import reverse from apps.catalog import services as catalog_services from apps.catalog.tests import make_catalog from apps.access.authorization import is_authorized from . import queries, services from .evidence_admin import revision_token from .models import DevicePurchaseEvidence as Purchase, DeviceWarrantyCoverage as Coverage from .tests import policy, synthetic_imei TODAY = date(2026, 1, 15) START = date(2025, 1, 1) END = date(2025, 12, 31) def coverage(device, **kwargs): return services.set_device_warranty_coverage(**{ "device": device, "coverage_start_date": START, "coverage_end_date": END, "coverage_source": "MANUFACTURER_RECORD", "reference": "SYNTHETIC-REFERENCE", **kwargs, }) class EvidenceTests(TestCase): def setUp(self): self.brand, self.category, self.model, self.variant = make_catalog("EVIDENCE") policy(self.model) self.device = services.register_device(product_model=self.model) self.user = get_user_model().objects.create_user(username="synthetic-reviewer") self.addCleanup(patch.stopall) patch("apps.devices.evidence.timezone.localdate", return_value=TODAY).start() def evidence(self, **kwargs): return services.set_device_purchase_evidence(device=self.device, **kwargs) def verify(self, evidence=None): return services.verify_device_purchase_evidence(evidence=evidence or self.evidence(), reviewed_by=self.user, note="Synthetic review") def test_purchase_uuid_timestamps_optional_date_and_default_review(self): evidence = self.evidence() self.assertIsInstance(evidence.pk, uuid.UUID) self.assertEqual(evidence.device_id, self.device.pk) self.assertIsNone(evidence.purchase_date) self.assertIsNotNone(evidence.created_at) self.assertIsNotNone(evidence.updated_at) self.assertEqual(evidence.evidence_status, "UNVERIFIED") self.assertIsNone(evidence.reviewed_by_id) self.assertIsNone(evidence.reviewed_at) self.assertFalse(evidence.verification_note) def test_dates_and_conservative_text_normalization(self): for value in (TODAY - timedelta(days=1), TODAY, TODAY.isoformat()): evidence = self.evidence(purchase_date=value, seller_name=" Synthetic Seller ", seller_reference=" dealer/aB ", invoice_number=" Inv-aB/001 ") self.assertIn(evidence.purchase_date, (TODAY, TODAY - timedelta(days=1))) self.assertEqual((evidence.seller_name, evidence.seller_reference, evidence.invoice_number), ("Synthetic Seller", "dealer/aB", "Inv-aB/001")) for value in (TODAY + timedelta(days=1), "not-a-date"): with self.assertRaises(ValidationError): self.evidence(purchase_date=value) self.assertEqual(queries.get_device_purchase_evidence(self.device).purchase_date, TODAY) def test_one_to_one_upsert_and_patch_preserve_omitted_values(self): original = self.evidence(seller_name="Synthetic", invoice_number="INV-SYNTHETIC") updated = self.evidence(seller_reference="reference") self.assertEqual(original.pk, updated.pk) self.assertEqual(updated.invoice_number, "INV-SYNTHETIC") self.assertEqual(updated.seller_name, "Synthetic") self.assertEqual(Purchase.objects.count(), 1) other = services.register_device(product_model=self.model) services.set_device_purchase_evidence(device=other, invoice_number="INV-SYNTHETIC") self.assertEqual(Purchase.objects.count(), 2) def test_verify_and_reject_record_complete_review_metadata(self): evidence = self.verify() self.assertEqual(evidence.evidence_status, "VERIFIED") self.assertEqual(evidence.reviewed_by_id, self.user.pk) self.assertIsNotNone(evidence.reviewed_at) self.assertEqual(evidence.verification_note, "Synthetic review") rejected = services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user, note=" Synthetic rejection ") self.assertEqual(rejected.evidence_status, "REJECTED") self.assertEqual(rejected.verification_note, "Synthetic rejection") self.assertGreaterEqual(rejected.reviewed_at, evidence.reviewed_at) def test_reviewer_must_be_persisted_active_user_but_not_necessarily_staff(self): evidence = self.evidence() self.assertFalse(self.user.is_staff) evidence = self.verify(evidence) get_user_model().objects.filter(pk=self.user.pk).update(is_active=False) for reviewer in (self.user, get_user_model()(username="unsaved"), None): with self.assertRaises(ValidationError): services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=reviewer) self.assertEqual(queries.get_device_purchase_evidence(self.device).evidence_status, "VERIFIED") def test_each_material_fact_change_invalidates_review(self): for field, value in (("purchase_date", START), ("seller_name", "New Synthetic Seller"), ("seller_reference", "New reference"), ("invoice_number", "INV-NEW")): evidence = self.verify(queries.get_device_purchase_evidence(self.device)) updated = self.evidence(**{field: value}) self.assertEqual(updated.evidence_status, "UNVERIFIED") self.assertIsNone(updated.reviewed_by_id) self.assertIsNone(updated.reviewed_at) self.assertEqual(updated.verification_note, "") def test_material_edit_resets_rejection_and_noop_does_not_change_review(self): evidence = self.evidence(seller_name="Synthetic Seller") rejected = services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user, note="Synthetic") unchanged = self.evidence(seller_name=" Synthetic Seller ") self.assertEqual(unchanged.updated_at, rejected.updated_at) self.assertEqual(unchanged.evidence_status, "REJECTED") self.assertEqual(self.evidence(seller_name="Corrected Seller").evidence_status, "UNVERIFIED") with self.assertRaises(TypeError): self.evidence(verification_note="Not an ordinary editable field") def test_stale_review_and_optimistic_edit_cannot_restore_old_state(self): old = self.evidence(invoice_number="SYNTHETIC-OLD") current = self.evidence(invoice_number="SYNTHETIC-NEW") with self.assertRaises(ValidationError): self.verify(old) current = services.reject_device_purchase_evidence(evidence=current, reviewed_by=self.user) with self.assertRaises(ValidationError): self.evidence(invoice_number="SYNTHETIC-OLD", expected_updated_at=old.updated_at) self.assertEqual(queries.get_device_purchase_evidence(self.device).evidence_status, "REJECTED") self.assertEqual(current.invoice_number, "SYNTHETIC-NEW") def test_failed_purchase_or_review_edit_rolls_back(self): evidence = self.verify() with self.assertRaises(ValidationError): self.evidence(purchase_date=TODAY + timedelta(days=1), seller_name="Must not persist") with patch.object(Purchase, "_persist", side_effect=ValidationError("Synthetic persistence failure")): with self.assertRaises(ValidationError): services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) current = queries.get_device_purchase_evidence(self.device) self.assertEqual(current.updated_at, evidence.updated_at) self.assertEqual(current.evidence_status, "VERIFIED") self.assertEqual(current.seller_name, "") def test_database_purchase_status_review_and_one_to_one_constraints(self): evidence = self.evidence() for changes in ({"evidence_status": "INVALID"}, {"evidence_status": "VERIFIED"}, {"reviewed_by_id": self.user.pk}, {"verification_note": "Old review"}): with self.assertRaises(IntegrityError), transaction.atomic(): Purchase.objects.filter(pk=evidence.pk).update(**changes) with self.assertRaises(IntegrityError), transaction.atomic(): Purchase.objects.bulk_create([Purchase(device=self.device)]) reviewed = self.verify(evidence) with self.assertRaises(IntegrityError), transaction.atomic(): Purchase.objects.filter(pk=reviewed.pk).update(reviewed_at=None) with self.assertRaises(ProtectedError): self.user.delete() def test_model_paths_reject_ownership_and_history_bypasses(self): evidence, record = self.evidence(), coverage(self.device) other = services.register_device(product_model=self.model) for obj in (evidence, record): obj.device = other with self.assertRaises(ValidationError): obj.full_clean() with self.assertRaises(ValidationError): obj.save() with self.assertRaises(ValidationError): obj.delete() with self.assertRaises(ValidationError): type(obj).objects.filter(pk=obj.pk).delete() with self.assertRaises(ProtectedError): Collector(using="default").collect([self.device]) def test_coverage_dates_uuid_source_and_explicit_future_period(self): record = coverage(self.device, coverage_start_date=date(2030, 1, 1), coverage_end_date=date(2031, 1, 1)) self.assertIsInstance(record.pk, uuid.UUID) self.assertTrue(record.is_active) self.assertIsNotNone(record.created_at) self.assertIsNotNone(record.updated_at) for changes in ({"coverage_end_date": START - timedelta(days=1)}, {"coverage_start_date": None}, {"coverage_end_date": "invalid"}, {"coverage_source": "INVALID"}): with self.assertRaises(ValidationError): coverage(self.device, **changes) self.assertEqual(queries.current_warranty_coverage(self.device).pk, record.pk) def test_coverage_replacement_and_clear_preserve_history(self): first = coverage(self.device) second = coverage(self.device, reference="SYNTHETIC-REPLACEMENT") first.refresh_from_db() self.assertFalse(first.is_active) self.assertEqual(first.reference, "SYNTHETIC-REFERENCE") self.assertEqual(queries.current_warranty_coverage(self.device).pk, second.pk) cleared = services.clear_device_warranty_coverage(device=self.device) self.assertFalse(cleared.is_active) self.assertIsNone(services.clear_device_warranty_coverage(device=self.device)) self.assertEqual(queries.warranty_coverage_history(self.device).count(), 2) def test_failed_replacement_restores_previous_active_coverage(self): first = coverage(self.device) original = Coverage._persist def fail_new(record, **kwargs): if record._state.adding: raise ValidationError("Synthetic insertion failure") return original(record, **kwargs) with patch.object(Coverage, "_persist", fail_new): with self.assertRaises(ValidationError): coverage(self.device, reference="Must not survive") first.refresh_from_db() self.assertTrue(first.is_active) self.assertEqual(Coverage.objects.count(), 1) def test_stale_coverage_precondition_cannot_replace_or_clear_new_record(self): first = coverage(self.device) second = coverage(self.device) with self.assertRaises(ValidationError): coverage(self.device, expected_current_id=first.pk) with self.assertRaises(ValidationError): services.clear_device_warranty_coverage(device=self.device, expected_current_id=first.pk) self.assertEqual(queries.current_warranty_coverage(self.device).pk, second.pk) def test_manual_override_requires_meaningful_note_or_reference(self): with self.assertRaises(ValidationError): coverage(self.device, coverage_source="MANUAL_OVERRIDE", reference=" ", note="\t") record = coverage(self.device, coverage_source="MANUAL_OVERRIDE", reference="", note=" Deliberate synthetic exception ") self.assertEqual(record.note, "Deliberate synthetic exception") def test_purchase_source_requires_verified_evidence_and_edit_retires_it(self): with self.assertRaises(ValidationError): coverage(self.device, coverage_source="PURCHASE_EVIDENCE") evidence = self.evidence(invoice_number="SYNTHETIC-ONE") with self.assertRaises(ValidationError): coverage(self.device, coverage_source="PURCHASE_EVIDENCE") self.verify(evidence) record = coverage(self.device, coverage_source="PURCHASE_EVIDENCE") self.evidence(invoice_number="SYNTHETIC-TWO") record.refresh_from_db() self.assertFalse(record.is_active) self.verify(queries.get_device_purchase_evidence(self.device)) self.assertIsNone(queries.current_warranty_coverage(self.device)) def test_rejection_retires_purchase_coverage_but_other_sources_are_independent(self): evidence = self.verify() record = coverage(self.device, coverage_source="PURCHASE_EVIDENCE") services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) record.refresh_from_db() self.assertFalse(record.is_active) for source in ("MANUFACTURER_RECORD", "MANUAL_OVERRIDE"): record = coverage(self.device, coverage_source=source) self.evidence(seller_name=source) self.assertEqual(queries.current_warranty_coverage(self.device).pk, record.pk) def test_evidence_invalidation_and_coverage_retirement_rollback_together(self): evidence = self.verify() record = coverage(self.device, coverage_source="PURCHASE_EVIDENCE") with patch.object(Coverage, "_persist", side_effect=ValidationError("Synthetic retirement failure")): with self.assertRaises(ValidationError): self.evidence(invoice_number="Must not persist") self.assertEqual(queries.get_device_purchase_evidence(self.device).evidence_status, "VERIFIED") self.assertEqual(queries.current_warranty_coverage(self.device).pk, record.pk) def test_verified_evidence_does_not_create_coverage_and_manufacturer_needs_no_evidence(self): self.verify() self.assertIsNone(queries.current_warranty_coverage(self.device)) other = services.register_device(product_model=self.model) self.assertIsNone(queries.get_device_purchase_evidence(other)) self.assertTrue(coverage(other).is_active) def test_inclusive_date_boundaries_and_no_implicit_current_date(self): coverage(self.device) for day, expected in ((START - timedelta(days=1), False), (START, True), (START + timedelta(days=30), True), (END, True), (END + timedelta(days=1), False)): self.assertEqual(queries.device_has_recorded_warranty_coverage(device=self.device, on_date=day), expected) for day in (None, "2025-01-01", datetime(2025, 1, 1)): with self.assertRaises(ValidationError): queries.device_has_recorded_warranty_coverage(device=self.device, on_date=day) def test_database_coverage_constraints(self): record = coverage(self.device) for changes in ({"coverage_source": "INVALID"}, {"coverage_end_date": START - timedelta(days=1)}, {"coverage_source": "MANUAL_OVERRIDE", "reference": "", "note": ""}): with self.assertRaises(IntegrityError), transaction.atomic(): Coverage.objects.filter(pk=record.pk).update(**changes) with self.assertRaises(IntegrityError), transaction.atomic(): Coverage.objects.bulk_create([Coverage(device=self.device, coverage_start_date=START, coverage_end_date=END, coverage_source="MANUFACTURER_RECORD")]) def test_device_and_catalog_lifecycle_preserve_records_and_allow_historical_evidence_edits(self): evidence = self.verify() record = coverage(self.device) services.deactivate_device(device=self.device) evidence.refresh_from_db() record.refresh_from_db() self.assertEqual(evidence.evidence_status, "VERIFIED") self.assertTrue(record.is_active) self.assertFalse(queries.device_has_recorded_warranty_coverage(device=self.device, on_date=START)) with self.assertRaises(ValidationError): coverage(self.device) self.evidence(seller_name="Historical correction") services.reactivate_device(device=self.device) self.assertEqual(queries.current_warranty_coverage(self.device).pk, record.pk) catalog_services.deactivate_product_model(product_model=self.model) self.evidence(seller_reference="Historical catalog correction") self.assertTrue(queries.device_has_recorded_warranty_coverage(device=self.device, on_date=START)) def test_imei_correction_keeps_evidence_and_coverage_on_device_uuid(self): services.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei()) evidence, record = self.verify(), coverage(self.device) services.replace_device_identifier(device=self.device, identifier_type="IMEI1", new_value=synthetic_imei(2)) self.assertEqual(queries.get_device_purchase_evidence(self.device).pk, evidence.pk) self.assertEqual(queries.current_warranty_coverage(self.device).pk, record.pk) self.assertEqual(queries.find_device_by_identifier(synthetic_imei()), self.device) def test_queries_are_current_isolated_and_efficient(self): self.assertIsNone(queries.get_device_purchase_evidence(self.device)) evidence, record = self.evidence(), coverage(self.device) other = services.register_device(product_model=self.model) with self.assertNumQueries(1): self.assertEqual(queries.get_device_purchase_evidence(self.device), evidence) with self.assertNumQueries(1): self.assertEqual(queries.current_warranty_coverage(self.device), record) with self.assertNumQueries(1): self.assertTrue(queries.device_has_recorded_warranty_coverage(device=self.device, on_date=START)) with self.assertNumQueries(0): history = queries.warranty_coverage_history(self.device) with self.assertNumQueries(1): self.assertEqual(list(history), [record]) self.assertIsNone(queries.current_warranty_coverage(other)) self.assertIsNone(queries.get_device_purchase_evidence(other)) self.assertIsNone(queries.current_warranty_coverage(None)) def test_evidence_types_do_not_extend_authorization_engine(self): user = get_user_model().objects.create_superuser(username="scope-test", password="test-only-password") for record in (self.evidence(), coverage(self.device)): permission = f"devices.view_{record._meta.model_name}" self.assertTrue(user.has_perm(permission)) self.assertFalse(is_authorized(user=user, permission=permission, target=record)) class EvidenceAdminTests(TestCase): @classmethod def setUpTestData(cls): _, _, cls.model, _ = make_catalog("EVIDENCEADMIN") policy(cls.model) cls.device = services.register_device(product_model=cls.model) cls.other = services.register_device(product_model=cls.model) cls.user = get_user_model().objects.create_superuser(username="evidence-admin", password="test-only-password") def setUp(self): self.client.force_login(self.user) def url(self, model, action, record=None): return reverse(f"admin:devices_{model._meta.model_name}_{action}", args=[record.pk] if record else None) def evidence(self): return services.set_device_purchase_evidence(device=self.device, invoice_number="SYNTHETIC-INVOICE") def payload(self, record, **kwargs): return {"purchase_date": "", "seller_name": "", "seller_reference": "", "invoice_number": record.invoice_number, "revision": revision_token(record), "_save": "Save", **kwargs} def test_purchase_admin_create_and_review_actions_use_services(self): response = self.client.post(self.url(Purchase, "add"), {"device": str(self.device.pk), "purchase_date": "", "seller_name": "Synthetic", "invoice_number": "SYNTHETIC-INVOICE", "evidence_status": "VERIFIED", "_save": "Save"}) self.assertEqual(response.status_code, 302) record = Purchase.objects.get() self.assertEqual(record.evidence_status, "UNVERIFIED") for action, status in (("verify_selected", "VERIFIED"), ("reject_selected", "REJECTED")): response = self.client.post(self.url(Purchase, "changelist"), {"action": action, "_selected_action": str(record.pk)}) self.assertEqual(response.status_code, 302) record.refresh_from_db() self.assertEqual(record.evidence_status, status) self.assertEqual(record.reviewed_by_id, self.user.pk) def test_material_admin_edit_invalidates_review_and_ignores_owner_status_forgery(self): evidence = services.verify_device_purchase_evidence(evidence=self.evidence(), reviewed_by=self.user) response = self.client.post(self.url(Purchase, "change", evidence), self.payload(evidence, invoice_number="SYNTHETIC-NEW", device=str(self.other.pk), evidence_status="VERIFIED")) self.assertEqual(response.status_code, 302) evidence.refresh_from_db() self.assertEqual(evidence.device_id, self.device.pk) self.assertEqual(evidence.evidence_status, "UNVERIFIED") self.assertIsNone(evidence.reviewed_by_id) def test_stale_purchase_form_cannot_restore_old_values_or_review(self): evidence = self.evidence() old_payload = self.payload(evidence, invoice_number="SYNTHETIC-OLD") services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) response = self.client.post(self.url(Purchase, "change", evidence), old_payload) self.assertEqual(response.status_code, 200) evidence.refresh_from_db() self.assertEqual(evidence.evidence_status, "REJECTED") self.assertEqual(evidence.invoice_number, "SYNTHETIC-INVOICE") def test_review_between_form_validation_and_save_rejects_stale_write(self): evidence = self.evidence() registered = admin.site._registry[Purchase] original = registered.save_model def interleave(request, obj, form, change): services.reject_device_purchase_evidence(evidence=evidence, reviewed_by=self.user) return original(request, obj, form, change) with patch.object(registered, "save_model", side_effect=interleave): response = self.client.post(self.url(Purchase, "change", evidence), self.payload(evidence, invoice_number="Must roll back")) self.assertEqual(response.status_code, 302) evidence.refresh_from_db() self.assertEqual(evidence.invoice_number, "SYNTHETIC-INVOICE") # The injected review shares the admin transaction and also rolls back. self.assertEqual(evidence.evidence_status, "UNVERIFIED") def test_coverage_admin_creation_replacement_and_historical_readonly(self): response = self.client.post(self.url(Coverage, "add"), {"device": str(self.device.pk), "coverage_start_date": START.isoformat(), "coverage_end_date": END.isoformat(), "coverage_source": "MANUFACTURER_RECORD", "reference": "SYNTHETIC", "note": "", "_save": "Save"}) self.assertEqual(response.status_code, 302) first = Coverage.objects.get() response = self.client.post(self.url(Coverage, "change", first), {"revision": revision_token(first), "replacement_start_date": START.isoformat(), "replacement_end_date": END.isoformat(), "replacement_source": "MANUAL_OVERRIDE", "replacement_reference": "SYNTHETIC-REPLACEMENT", "_save": "Save"}) self.assertEqual(response.status_code, 302) first.refresh_from_db() self.assertFalse(first.is_active) self.assertEqual(Coverage.objects.count(), 2) self.assertEqual(self.client.get(self.url(Coverage, "change", first)).status_code, 200) self.assertEqual(self.client.post(self.url(Coverage, "change", first), {"is_active": "on", "_save": "Save"}).status_code, 403) def test_stale_coverage_admin_cannot_clear_or_overwrite_replacement(self): first = coverage(self.device) payload = {"revision": revision_token(first), "replacement_start_date": START.isoformat(), "replacement_end_date": END.isoformat(), "replacement_source": "MANUFACTURER_RECORD", "_save": "Save"} second = coverage(self.device) self.assertEqual(self.client.post(self.url(Coverage, "change", first), payload).status_code, 403) self.client.post(self.url(Coverage, "changelist"), {"action": "clear_selected", "_selected_action": str(first.pk)}) self.assertEqual(queries.current_warranty_coverage(self.device).pk, second.pk) def test_admin_coverage_clear_preserves_history_and_no_deletion(self): record = coverage(self.device) response = self.client.post(self.url(Coverage, "changelist"), {"action": "clear_selected", "_selected_action": str(record.pk)}) self.assertEqual(response.status_code, 302) self.assertIsNone(queries.current_warranty_coverage(self.device)) for obj in (self.evidence(), record): self.assertEqual(self.client.post(self.url(type(obj), "delete", obj), {"post": "yes"}).status_code, 403) def test_admin_view_only_and_csrf_protect_evidence(self): evidence = self.evidence() reader = get_user_model().objects.create_user(username="evidence-reader", is_staff=True) reader.user_permissions.add(Permission.objects.get(content_type__app_label="devices", codename="view_devicepurchaseevidence")) self.client.force_login(reader) self.assertEqual(self.client.post(self.url(Purchase, "change", evidence), self.payload(evidence)).status_code, 403) client = Client(enforce_csrf_checks=True) client.force_login(self.user) self.assertEqual(client.post(self.url(Purchase, "add"), {"device": str(self.device.pk)}).status_code, 403)