"""Synthetic physical-identity, policy, history and admin integrity tests.""" import uuid from itertools import product from unittest.mock import patch from django.contrib import admin from django.contrib.admin.models import LogEntry 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 ProtectedError from django.test import Client, TestCase from django.urls import reverse from apps.access.authorization import authorized_queryset, is_authorized from apps.catalog import services as catalog_services from apps.catalog.identification import set_identification_policy from apps.catalog.models import ProductModel from apps.catalog.tests import make_catalog from . import queries, services from .identification import normalize_identifier from .models import Device, DeviceIdentifier def synthetic_imei(seed=1): """Fabricated test prefix and payload, not copied from a physical device.""" body = f"990000{seed:08d}" # Independent check-digit construction: double alternate digits from right. checksum = 0 for offset, char in enumerate(reversed(body)): digit = int(char) * (2 if offset % 2 == 0 else 1) checksum += digit - 9 if digit > 9 else digit return body + str((-checksum) % 10) def policy(model, imei1="OPTIONAL", imei2="OPTIONAL", serial="OPTIONAL"): return set_identification_policy(product_model=model, imei1_requirement=imei1, imei2_requirement=imei2, serial_requirement=serial) class DeviceTests(TestCase): def setUp(self): self.brand, self.category, self.model, self.variant = make_catalog("REGISTRY") policy(self.model) def register(self, **kwargs): return services.register_device(product_model=self.model, **kwargs) def test_uuid_defaults_timestamps_and_display_without_identifier(self): device = self.register() self.assertIsInstance(device.pk, uuid.UUID) self.assertTrue(device.is_active) self.assertIsNotNone(device.created_at) self.assertIsNotNone(device.updated_at) self.assertIsNone(device.product_variant_id) self.assertIn(self.model.code, str(device)) self.assertIn(str(device.pk)[:8], str(device)) self.assertNotIn("customer", {f.name for f in device._meta.fields}) def test_matching_variant_and_mismatched_variant_rejection(self): device = self.register(product_variant=self.variant) self.assertEqual(device.product_variant_id, self.variant.pk) _, _, _, other_variant = make_catalog("OTHER") with self.assertRaises(ValidationError): self.register(product_variant=other_variant) self.assertEqual(Device.objects.count(), 1) def test_registration_requires_persisted_model_and_variant(self): for model in (None, ProductModel()): with self.assertRaises(ValidationError): services.register_device(product_model=model) with self.assertRaises(ValidationError): self.register(product_variant=type(self.variant)()) def test_unconfigured_is_not_an_implicit_optional_policy(self): self.model.identification_policy.delete() with self.assertRaises(ValidationError): self.register() self.assertFalse(Device.objects.exists()) policy(self.model, "NOT_APPLICABLE", "NOT_APPLICABLE", "NOT_APPLICABLE") self.assertEqual(self.register().identifiers.count(), 0) def test_policy_requirements_across_all_valid_configurations(self): # Exercise requirement semantics without rewriting catalog's own enum tests. options = ("NOT_APPLICABLE", "OPTIONAL", "REQUIRED") counter = 100 for requirements in product(options, repeat=3): if requirements[0] == "NOT_APPLICABLE" and requirements[1] != "NOT_APPLICABLE": continue policy(self.model, *requirements) supplied = {} for index, (field, requirement) in enumerate(zip(("imei1", "imei2", "serial"), requirements)): if requirement == "REQUIRED": supplied[field] = f"SYNTHETIC-{counter}" if field == "serial" else synthetic_imei(counter) counter += 1 self.register(**supplied) for field, requirement in zip(("imei1", "imei2", "serial"), requirements): invalid = dict(supplied) if requirement == "REQUIRED": invalid.pop(field) elif requirement == "NOT_APPLICABLE": invalid[field] = "SYNTHETIC-FORBIDDEN" if field == "serial" else synthetic_imei(counter) else: continue with self.subTest(requirements=requirements, field=field), self.assertRaises(ValidationError): self.register(**invalid) def test_valid_imeis_normalize_spaces_and_hyphens_and_use_luhn(self): value = synthetic_imei() entered = f" {value[:3]}-{value[3:9]} {value[9:]} " device = self.register(imei1=entered, imei2=synthetic_imei(2)) record = device.identifiers.get(identifier_type="IMEI1") self.assertEqual(record.normalized_value, value) self.assertEqual(record.identity_key, value) self.assertEqual(record.value, entered.strip()) self.assertIsInstance(record.pk, uuid.UUID) self.assertIsNotNone(record.created_at) def test_invalid_imeis_and_supplied_blank_serial_roll_back_registration(self): valid = synthetic_imei() for value in (valid[:-1], valid + "0", "A" + valid[1:], valid[:-1] + str((int(valid[-1]) + 1) % 10), valid[:5] + "\t" + valid[5:], "９" + valid[1:], " ", None): with self.subTest(case=type(value).__name__), self.assertRaises(ValidationError): self.register(imei1=value) for value in (" ", "\t", "SYNTHETIC\x00SERIAL"): with self.assertRaises(ValidationError): self.register(serial=value) self.assertFalse(Device.objects.exists()) self.assertFalse(DeviceIdentifier.objects.exists()) def test_serial_normalization_preserves_internal_characters_and_is_case_insensitive(self): device = self.register(serial=" synthetic-Ab/ 12-x ") record = device.identifiers.get() self.assertEqual(record.value, "synthetic-Ab/ 12-x") self.assertEqual(record.normalized_value, "SYNTHETIC-AB/ 12-X") self.assertEqual(queries.find_device_by_identifier("SYNTHETIC-ab/ 12-X"), device) with self.assertRaises(ValidationError): self.register(serial="SYNTHETIC-AB/ 12-X") def test_global_imei_reservation_across_types_and_same_device_slots(self): value = synthetic_imei() self.register(imei1=value) for kwargs in ({"imei1": value}, {"imei2": value}, {"imei1": synthetic_imei(2), "imei2": synthetic_imei(2)}): with self.assertRaises(ValidationError): self.register(**kwargs) self.assertEqual(Device.objects.count(), 1) def test_serial_reservation_is_global_across_models(self): self.register(serial="SYNTHETIC-GLOBAL") _, _, other, _ = make_catalog("OTHER") policy(other) with self.assertRaises(ValidationError): services.register_device(product_model=other, serial="synthetic-global") def test_numeric_serial_imei_aliases_cannot_make_lookup_ambiguous(self): value = synthetic_imei() serial = value[:5] + "-" + value[5:] device = self.register(serial=serial) self.assertEqual(queries.find_device_by_identifier(value), device) self.assertEqual(queries.find_device_by_identifier(serial), device) self.assertEqual(device.identifiers.get().normalized_value, serial) with self.assertRaises(ValidationError): self.register(imei2=value) with self.assertRaises(ValidationError): self.register(serial=value) def test_all_catalog_ancestor_states_block_registration(self): for deactivate, reactivate in ( (lambda: catalog_services.deactivate_brand(brand=self.brand), lambda: catalog_services.reactivate_brand(brand=self.brand)), (lambda: catalog_services.deactivate_category(category=self.category), lambda: catalog_services.reactivate_category(category=self.category)), (lambda: catalog_services.deactivate_product_model(product_model=self.model), lambda: catalog_services.reactivate_product_model(product_model=self.model)), (lambda: catalog_services.deactivate_variant(variant=self.variant), lambda: catalog_services.reactivate_variant(variant=self.variant)), ): deactivate() with self.assertRaises(ValidationError): self.register(product_variant=self.variant) reactivate() catalog_services.reactivate_product_model(product_model=self.model) catalog_services.reactivate_variant(variant=self.variant) def test_direct_save_paths_cannot_bypass_policy_ownership_or_identifier_history(self): device = self.register(imei1=synthetic_imei()) _, _, other_model, other_variant = make_catalog("OTHER") for field, value in (("product_model", other_model), ("product_variant", other_variant)): device.refresh_from_db() setattr(device, field, value) with self.assertRaises(ValidationError): device.full_clean() with self.assertRaises(ValidationError): device.save() with self.assertRaises(ValidationError): Device.objects.create(product_model=self.model) record = DeviceIdentifier.objects.get() record.value = synthetic_imei(3) with self.assertRaises(ValidationError): record.save() with self.assertRaises(ValidationError): DeviceIdentifier.objects.create(device=device, identifier_type="SERIAL", value="SYNTHETIC") def test_correction_preserves_history_and_historical_lookup(self): old, new = synthetic_imei(), synthetic_imei(2) device = self.register(imei1=old) original = device.identifiers.get() services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=new) original.refresh_from_db() self.assertFalse(original.is_active) self.assertEqual(original.normalized_value, old) self.assertEqual(device.identifiers.count(), 2) for value in (old, new): self.assertEqual(queries.find_device_by_identifier(value), device) with self.assertRaises(ValidationError): self.register(imei2=old) def test_same_device_restore_reuses_original_row_and_same_value_is_noop(self): device = self.register(imei1=synthetic_imei()) original = device.identifiers.get() services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei()) self.assertEqual(device.identifiers.get().updated_at, original.updated_at) services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)) services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei()) self.assertEqual(device.identifiers.get(is_active=True).pk, original.pk) self.assertEqual(device.identifiers.count(), 2) with self.assertRaises(ValidationError): services.replace_device_identifier(device=device, identifier_type="IMEI2", new_value=synthetic_imei(2)) def test_correction_conflict_and_late_failure_preserve_current_configuration(self): device = self.register(imei1=synthetic_imei()) self.register(imei1=synthetic_imei(2)) with self.assertRaises(ValidationError): services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)) original_persist = DeviceIdentifier._persist def fail_new(record, **kwargs): if record._state.adding: raise ValidationError("Synthetic insert failure") return original_persist(record, **kwargs) with patch.object(DeviceIdentifier, "_persist", fail_new): with self.assertRaises(ValidationError): services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(3)) self.assertEqual(device.identifiers.count(), 1) self.assertEqual(device.identifiers.get(is_active=True).normalized_value, synthetic_imei()) def test_late_registration_failure_leaves_no_partial_device(self): original = DeviceIdentifier._persist def fail_serial(record, **kwargs): if record.identifier_type == "SERIAL": raise ValidationError("Synthetic final insert failure") return original(record, **kwargs) with patch.object(DeviceIdentifier, "_persist", fail_serial): with self.assertRaises(ValidationError): self.register(imei1=synthetic_imei(), serial="SYNTHETIC") self.assertFalse(Device.objects.exists()) self.assertFalse(DeviceIdentifier.objects.exists()) def test_current_policy_controls_correction_but_never_deletes_history(self): device = self.register(imei1=synthetic_imei()) policy(self.model, "REQUIRED", "OPTIONAL", "REQUIRED") with self.assertRaises(ValidationError): services.reactivate_device(device=device) services.replace_device_identifier(device=device, identifier_type="SERIAL", new_value="SYNTHETIC-REQUIRED") self.assertTrue(services.reactivate_device(device=device).is_active) with self.assertRaises(ValidationError): services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value="") policy(self.model, "NOT_APPLICABLE", "NOT_APPLICABLE", "OPTIONAL") with self.assertRaises(ValidationError): services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)) self.assertEqual(device.identifiers.count(), 2) def test_device_lifecycle_retains_identifier_reservations_and_is_idempotent(self): device = self.register(imei1=synthetic_imei()) identifier = device.identifiers.get() first = services.deactivate_device(device=device) self.assertEqual(services.deactivate_device(device=device).updated_at, first.updated_at) identifier.refresh_from_db() self.assertTrue(identifier.is_active) self.assertEqual(queries.find_device_by_identifier(synthetic_imei()), device) with self.assertRaises(ValidationError): self.register(imei2=synthetic_imei()) self.assertFalse(queries.active_devices_for_product_model(self.model).exists()) first = services.reactivate_device(device=device) self.assertTrue(first.is_active) self.assertEqual(services.reactivate_device(device=device).updated_at, first.updated_at) def test_correction_on_inactive_device_does_not_reactivate_it(self): device = self.register(imei1=synthetic_imei()) services.deactivate_device(device=device) result = services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)) self.assertFalse(result.is_active) self.assertEqual(result.identifiers.filter(is_active=True).count(), 1) def test_catalog_deactivation_preserves_devices_and_blocks_reactivation(self): device = self.register(product_variant=self.variant, imei1=synthetic_imei()) catalog_services.deactivate_product_model(product_model=self.model) device.refresh_from_db() self.assertTrue(device.is_active) self.assertTrue(device.identifiers.get().is_active) self.assertFalse(queries.active_devices_for_product_model(self.model).exists()) services.deactivate_device(device=device) with self.assertRaises(ValidationError): services.reactivate_device(device=device) def test_exact_lookup_and_query_counts_include_history_not_substrings(self): device = self.register(imei1=synthetic_imei(), imei2=synthetic_imei(2), serial="SYNTHETIC-LOOKUP") for value in (synthetic_imei(), synthetic_imei(2), " synthetic-lookup "): with self.assertNumQueries(1): self.assertEqual(queries.find_device_by_identifier(value), device) for value in ("SYNTHETIC", synthetic_imei()[:8], "UNKNOWN", "", None): self.assertIsNone(queries.find_device_by_identifier(value)) with self.assertNumQueries(0): identifiers = queries.active_identifiers_for_device(device) devices = queries.active_devices_for_product_model(self.model) with self.assertNumQueries(1): self.assertEqual(len(list(identifiers)), 3) with self.assertNumQueries(1): self.assertEqual(list(devices), [device]) self.assertFalse(queries.identifier_history_for_device(Device()).exists()) self.assertFalse(queries.active_devices_for_product_model(ProductModel()).exists()) def test_database_constraints_protect_types_cardinality_shape_and_reservations(self): device = self.register(imei1=synthetic_imei(), serial="SYNTHETIC-DB") first = device.identifiers.get(identifier_type="IMEI1") for changes in ({"identifier_type": "OTHER"}, {"normalized_value": ""}, {"identity_key": ""}, {"normalized_value": "123"}, {"value": ""}, {"identity_key": "MISMATCH"}): with self.subTest(changes=changes), self.assertRaises(IntegrityError), transaction.atomic(): DeviceIdentifier.objects.filter(pk=first.pk).update(**changes) with self.assertRaises(IntegrityError), transaction.atomic(): DeviceIdentifier.objects.bulk_create([DeviceIdentifier(device=device, identifier_type="IMEI1", value=synthetic_imei(2), normalized_value=synthetic_imei(2), identity_key=synthetic_imei(2))]) services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)) other = self.register() with self.assertRaises(IntegrityError), transaction.atomic(): DeviceIdentifier.objects.bulk_create([DeviceIdentifier(device=other, identifier_type="IMEI2", value=first.value, normalized_value=first.normalized_value, identity_key=first.identity_key)]) with self.assertRaises(IntegrityError), transaction.atomic(): DeviceIdentifier.objects.bulk_create([DeviceIdentifier(device=other, identifier_type="SERIAL", value="SYNTHETIC-DB", normalized_value="SYNTHETIC-DB", identity_key="SYNTHETIC-DB")]) def test_registry_deletion_disabled_and_catalog_references_protected(self): device = self.register(product_variant=self.variant, imei1=synthetic_imei()) for record in (device, device.identifiers.get()): with self.assertRaises(ValidationError): record.delete() with self.assertRaises(ValidationError): type(record).objects.filter(pk=record.pk).delete() with self.assertRaises(ProtectedError): self.variant.delete() with self.assertRaises(ProtectedError): self.model.delete() def test_device_and_identifier_do_not_extend_scope_authorization(self): device = self.register(imei1=synthetic_imei()) user = get_user_model().objects.create_superuser(username="registry-admin", password="test-only-password") for obj in (device, device.identifiers.get()): permission = f"devices.view_{obj._meta.model_name}" self.assertTrue(user.has_perm(permission)) self.assertFalse(is_authorized(user=user, permission=permission, target=obj)) self.assertFalse(authorized_queryset(user=user, permission=permission, queryset=type(obj).objects.all()).exists()) class DeviceAdminTests(TestCase): @classmethod def setUpTestData(cls): cls.brand, cls.category, cls.model, cls.variant = make_catalog("ADMINDEVICE") policy(cls.model, "REQUIRED") cls.user = get_user_model().objects.create_superuser(username="device-admin", password="test-only-password") def setUp(self): self.client.force_login(self.user) def payload(self, **kwargs): return {"product_model": str(self.model.pk), "product_variant": "", "imei1": synthetic_imei(), "imei2": "", "serial": "", "_save": "Save", **kwargs} def device(self): return services.register_device(product_model=self.model, imei1=synthetic_imei()) def test_admin_registration_uses_policy_and_generated_device_identity(self): response = self.client.post(reverse("admin:devices_device_add"), self.payload()) self.assertEqual(response.status_code, 302) device = Device.objects.get() self.assertEqual(device.identifiers.get().normalized_value, synthetic_imei()) self.assertTrue(device.is_active) self.assertNotIn(synthetic_imei(), LogEntry.objects.get().change_message) self.assertNotIn(synthetic_imei(), LogEntry.objects.get().object_repr) def test_admin_missing_required_and_unconfigured_policy_rejected(self): self.assertEqual(self.client.post(reverse("admin:devices_device_add"), self.payload(imei1="")).status_code, 200) self.model.identification_policy.delete() self.assertEqual(self.client.post(reverse("admin:devices_device_add"), self.payload()).status_code, 200) self.assertFalse(Device.objects.exists()) def test_admin_catalog_ownership_flags_and_posted_old_identifiers_are_ignored(self): device = self.device() _, _, other, variant = make_catalog("OTHER") response = self.client.post(reverse("admin:devices_device_change", args=[device.pk]), self.payload(product_model=str(other.pk), product_variant=str(variant.pk), imei1=synthetic_imei(2), is_active="")) self.assertEqual(response.status_code, 302) device.refresh_from_db() self.assertEqual(device.product_model_id, self.model.pk) self.assertIsNone(device.product_variant_id) self.assertTrue(device.is_active) self.assertEqual(device.identifiers.get().normalized_value, synthetic_imei()) def test_admin_correction_preserves_history_and_duplicate_failure_rolls_back(self): device = self.device() url = reverse("admin:devices_device_change", args=[device.pk]) revision = self.client.get(url).context["adminform"].form.initial["identifier_revision"] response = self.client.post(reverse("admin:devices_device_change", args=[device.pk]), {"identifier_revision": revision, "correction_type": "IMEI1", "correction_value": synthetic_imei(2), "_save": "Save"}) self.assertEqual(response.status_code, 302) self.assertEqual(device.identifiers.count(), 2) services.register_device(product_model=self.model, imei1=synthetic_imei(3)) revision = self.client.get(url).context["adminform"].form.initial["identifier_revision"] response = self.client.post(reverse("admin:devices_device_change", args=[device.pk]), {"identifier_revision": revision, "correction_type": "IMEI1", "correction_value": synthetic_imei(3), "_save": "Save"}) self.assertEqual(response.status_code, 302) self.assertEqual(device.identifiers.get(is_active=True).normalized_value, synthetic_imei(2)) def test_stale_admin_save_does_not_undo_deactivation_or_correction(self): device = self.device() registered = admin.site._registry[Device] original = registered.save_model def interleave(request, obj, form, change): services.deactivate_device(device=device) services.replace_device_identifier(device=device, identifier_type="IMEI1", new_value=synthetic_imei(2)) return original(request, obj, form, change) with patch.object(registered, "save_model", side_effect=interleave): response = self.client.post(reverse("admin:devices_device_change", args=[device.pk]), self.payload()) self.assertEqual(response.status_code, 302) device.refresh_from_db() self.assertFalse(device.is_active) self.assertEqual(device.identifiers.get(is_active=True).normalized_value, synthetic_imei(2)) def test_admin_lifecycle_actions_and_no_delete(self): device = self.device() for action, active in (("deactivate_selected", False), ("reactivate_selected", True)): response = self.client.post(reverse("admin:devices_device_changelist"), {"action": action, "_selected_action": str(device.pk)}) self.assertEqual(response.status_code, 302) device.refresh_from_db() self.assertEqual(device.is_active, active) self.assertEqual(self.client.post(reverse("admin:devices_device_delete", args=[device.pk]), {"post": "yes"}).status_code, 403) def test_identifier_admin_is_audit_only(self): record = self.device().identifiers.get() self.assertEqual(self.client.get(reverse("admin:devices_deviceidentifier_change", args=[record.pk])).status_code, 200) for action in ("change", "delete"): self.assertEqual(self.client.post(reverse(f"admin:devices_deviceidentifier_{action}", args=[record.pk]), {"value": synthetic_imei(2), "post": "yes"}).status_code, 403) self.assertEqual(self.client.post(reverse("admin:devices_deviceidentifier_add"), {}).status_code, 403) def test_view_only_staff_and_csrf_cannot_mutate_devices(self): device = self.device() reader = get_user_model().objects.create_user(username="device-reader", is_staff=True) reader.user_permissions.add(Permission.objects.get(content_type__app_label="devices", codename="view_device")) self.client.force_login(reader) self.assertEqual(self.client.post(reverse("admin:devices_device_change", args=[device.pk]), {"correction_type": "IMEI1", "correction_value": synthetic_imei(2), "_save": "Save"}).status_code, 403) client = Client(enforce_csrf_checks=True) client.force_login(self.user) self.assertEqual(client.post(reverse("admin:devices_device_add"), self.payload()).status_code, 403)