import itertools import uuid 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 TestCase, TransactionTestCase from django.urls import reverse from apps.organization import test_assignment_concurrency as concurrency from . import services from .identification import get_identification_policy, set_identification_policy from .models import DeviceIdentificationPolicy as Policy, ProductModel from .tests import make_catalog R = Policy.Requirement def requirements(imei1=R.REQUIRED, imei2=R.OPTIONAL, serial=R.REQUIRED): return dict(imei1_requirement=imei1, imei2_requirement=imei2, serial_requirement=serial) class IdentificationPolicyTests(TestCase): @classmethod def setUpTestData(cls): cls.brand, cls.category, cls.model, cls.variant = make_catalog() cls.other_brand, cls.other_category, cls.other_model, cls.other_variant = make_catalog("B") def create(self, **values): return set_identification_policy(product_model=self.model, **(values or requirements())) def test_identity_timestamps_one_to_one_and_no_defaults(self): policy = self.create() self.assertIsInstance(policy.pk, uuid.UUID) self.assertEqual(policy.product_model, self.model) self.assertEqual(self.model.identification_policy, policy) self.assertIsNotNone(policy.created_at) self.assertIsNotNone(policy.updated_at) for field in requirements(): self.assertFalse(Policy._meta.get_field(field).has_default()) self.assertNotIn("is_active", [field.name for field in Policy._meta.fields]) with self.assertRaises(ValidationError): Policy.objects.create(product_model=self.other_model) def test_missing_distinct_from_explicit_none_and_category_not_inferred(self): self.assertIsNone(get_identification_policy(self.model)) policy = set_identification_policy(product_model=self.other_model, **requirements(R.NOT_APPLICABLE, R.NOT_APPLICABLE, R.NOT_APPLICABLE)) result = get_identification_policy(self.other_model) self.assertEqual(result.pk, policy.pk) self.assertEqual([getattr(result, field) for field in requirements()], [R.NOT_APPLICABLE] * 3) self.assertIsNone(get_identification_policy(self.model)) self.assertEqual(Policy.objects.count(), 1) def test_all_enum_combinations_model_and_database(self): for imei1, imei2, serial in itertools.product(R.values, repeat=3): values = requirements(imei1, imei2, serial) valid = imei1 != R.NOT_APPLICABLE or imei2 == R.NOT_APPLICABLE with self.subTest(**values): with transaction.atomic(): if valid: self.create(**values) else: with self.assertRaises(ValidationError): self.create(**values) transaction.set_rollback(True) with transaction.atomic(): if valid: Policy.objects.bulk_create([Policy(product_model=self.model, **values)]) else: with self.assertRaises(IntegrityError), transaction.atomic(): Policy.objects.bulk_create([Policy(product_model=self.model, **values)]) transaction.set_rollback(True) def test_invalid_enum_values_model_and_database(self): for field in requirements(): for value in ["UNKNOWN", "required", "", None]: values = requirements() values[field] = value with self.subTest(field=field, value=value): with self.assertRaises(ValidationError): self.create(**values) with self.assertRaises(IntegrityError), transaction.atomic(): Policy.objects.bulk_create([Policy(product_model=self.model, **values)]) def test_duplicate_model_and_database_rejection(self): original = self.create() with self.assertRaises(ValidationError): Policy.objects.create(product_model=self.model, **requirements()) with self.assertRaises(IntegrityError), transaction.atomic(): Policy.objects.bulk_create([Policy(product_model=self.model, **requirements())]) self.assertEqual(Policy.objects.get().pk, original.pk) def test_stable_ownership_in_full_and_partial_writes(self): policy = self.create() policy.product_model = self.other_model for fields in [None, ["product_model"], ["serial_requirement"]]: with self.assertRaises(ValidationError): policy.save(update_fields=fields) policy.refresh_from_db() self.assertEqual(policy.product_model, self.model) self.assertIsNone(get_identification_policy(self.other_model)) def test_update_preserves_identity_and_creation_timestamp(self): first = self.create() second = self.create(**requirements(R.OPTIONAL, R.REQUIRED, R.NOT_APPLICABLE)) self.assertEqual(second.pk, first.pk) self.assertEqual(second.created_at, first.created_at) self.assertGreater(second.updated_at, first.updated_at) self.assertEqual(second.imei2_requirement, R.REQUIRED) self.assertEqual(Policy.objects.count(), 1) def test_invalid_update_preserves_original(self): original = self.create() with self.assertRaises(ValidationError): self.create(**requirements(R.NOT_APPLICABLE, R.REQUIRED)) current = get_identification_policy(self.model) self.assertEqual(current.pk, original.pk) self.assertEqual(current.updated_at, original.updated_at) self.assertEqual(current.imei1_requirement, R.REQUIRED) def test_invalid_first_write_does_not_create_policy(self): with self.assertRaises(ValidationError): self.create(**requirements(R.NOT_APPLICABLE, R.OPTIONAL)) self.assertIsNone(get_identification_policy(self.model)) def test_partial_save_validates_effective_persisted_combination(self): policy = self.create(**requirements(R.NOT_APPLICABLE, R.NOT_APPLICABLE)) policy.imei1_requirement, policy.imei2_requirement = R.REQUIRED, R.REQUIRED with self.assertRaises(ValidationError): policy.save(update_fields=["imei2_requirement"]) policy.refresh_from_db() self.assertEqual(policy.imei2_requirement, R.NOT_APPLICABLE) def test_partial_edit_updates_timestamp(self): policy = self.create() previous = policy.updated_at policy.serial_requirement = R.OPTIONAL policy.save(update_fields=["serial_requirement"]) policy.refresh_from_db() self.assertGreater(policy.updated_at, previous) self.assertEqual(policy.serial_requirement, R.OPTIONAL) def test_outer_transaction_rollback_restores_policy(self): original = self.create() with self.assertRaises(RuntimeError): with transaction.atomic(): self.create(**requirements(R.REQUIRED, R.REQUIRED)) raise RuntimeError("simulated caller failure") self.assertEqual(get_identification_policy(self.model).imei2_requirement, original.imei2_requirement) def test_read_helper_ignores_cached_present_or_missing_relation(self): with self.assertRaises(Policy.DoesNotExist): _ = self.model.identification_policy original = self.create() self.assertEqual(get_identification_policy(self.model).pk, original.pk) fresh_model = ProductModel.objects.get(pk=self.model.pk) self.assertEqual(fresh_model.identification_policy.imei2_requirement, R.OPTIONAL) self.create(**requirements(R.REQUIRED, R.REQUIRED)) self.assertEqual(get_identification_policy(fresh_model).imei2_requirement, R.REQUIRED) def test_helpers_reject_unsaved_or_deleted_model(self): unsaved = ProductModel(code="UNSAVED", name="Unsaved", brand=self.brand, category=self.category) for model in [None, unsaved]: with self.assertRaises(ValueError): get_identification_policy(model) with self.assertRaises(ValueError): set_identification_policy(product_model=model, **requirements()) self.other_variant.delete() stale = ProductModel.objects.get(pk=self.other_model.pk) self.other_model.delete() with self.assertRaises(ProductModel.DoesNotExist): get_identification_policy(stale) def test_lifecycle_preserves_configuration(self): policy = self.create() for deactivate, reactivate, argument, target in [ (services.deactivate_brand, services.reactivate_brand, "brand", self.brand), (services.deactivate_category, services.reactivate_category, "category", self.category), (services.deactivate_product_model, services.reactivate_product_model, "product_model", self.model), ]: with self.subTest(operation=deactivate.__name__), transaction.atomic(): deactivate(**{argument: target}) self.model.refresh_from_db() self.variant.refresh_from_db() self.assertFalse(self.model.is_active) self.assertFalse(self.variant.is_active) result = get_identification_policy(self.model) self.assertEqual(result.pk, policy.pk) self.assertEqual(result.updated_at, policy.updated_at) reactivate(**{argument: target}) services.reactivate_product_model(product_model=self.model) self.variant.refresh_from_db() self.assertFalse(self.variant.is_active) self.assertEqual(get_identification_policy(self.model).imei2_requirement, R.OPTIONAL) transaction.set_rollback(True) def test_policy_can_be_created_and_edited_on_inactive_model(self): services.deactivate_brand(brand=self.brand) policy = self.create() self.assertEqual(self.create(**requirements(R.REQUIRED, R.REQUIRED)).pk, policy.pk) self.model.refresh_from_db() self.assertFalse(self.model.is_active) def test_policy_protects_model_and_deletion_means_unconfigured(self): policy = self.create() self.variant.delete() with self.assertRaises(ProtectedError): self.model.delete() policy.delete() self.assertIsNone(get_identification_policy(self.model)) def test_representation_uses_no_query_or_identifier_value(self): policy = self.create() with self.assertNumQueries(0): self.assertIn(str(self.model.pk), str(policy)) class IdentificationAdminTests(TestCase): @classmethod def setUpTestData(cls): cls.brand, cls.category, cls.model, cls.variant = make_catalog() cls.other_brand, cls.other_category, cls.other_model, cls.other_variant = make_catalog("B") cls.admin = get_user_model().objects.create_superuser(username="policy-admin", password="test-only-password") def setUp(self): self.client.force_login(self.admin) def add(self, **values): return self.client.post(reverse("admin:catalog_deviceidentificationpolicy_add"), { "product_model": str(self.model.pk), **(values or requirements()), "_save": "Save", }) def test_admin_create_display_and_edit(self): self.assertEqual(self.add().status_code, 302) policy = Policy.objects.get() response = self.client.get(reverse("admin:catalog_deviceidentificationpolicy_changelist")) self.assertContains(response, "Optional") response = self.client.post(reverse("admin:catalog_deviceidentificationpolicy_change", args=[policy.pk]), { **requirements(R.REQUIRED, R.REQUIRED), "_save": "Save", }) self.assertEqual(response.status_code, 302) policy.refresh_from_db() self.assertEqual(policy.imei2_requirement, R.REQUIRED) self.assertEqual(Permission.objects.filter(content_type__app_label="catalog", content_type__model="deviceidentificationpolicy").count(), 4) def test_invalid_dependency_and_enum_are_form_errors(self): self.assertContains(self.add(**requirements(R.NOT_APPLICABLE, R.OPTIONAL)), "IMEI2 requires IMEI1") self.assertContains(self.add(**requirements("INVALID")), "Select a valid choice") self.assertFalse(Policy.objects.exists()) def test_duplicate_add_cannot_overwrite_existing_policy(self): self.add() response = self.add(**requirements(R.REQUIRED, R.REQUIRED)) self.assertEqual(response.status_code, 200) self.assertEqual(Policy.objects.count(), 1) self.assertEqual(Policy.objects.get().imei2_requirement, R.OPTIONAL) def test_forged_parent_change_is_ignored(self): self.add() policy = Policy.objects.get() response = self.client.post(reverse("admin:catalog_deviceidentificationpolicy_change", args=[policy.pk]), { "product_model": str(self.other_model.pk), **requirements(), "_save": "Save", }) self.assertEqual(response.status_code, 302) policy.refresh_from_db() self.assertEqual(policy.product_model, self.model) def test_view_only_permission_cannot_edit_policy(self): self.add() reader = get_user_model().objects.create_user(username="policy-reader", is_staff=True) reader.user_permissions.add(Permission.objects.get(content_type__app_label="catalog", codename="view_deviceidentificationpolicy")) self.client.force_login(reader) policy = Policy.objects.get() response = self.client.post(reverse("admin:catalog_deviceidentificationpolicy_change", args=[policy.pk]), requirements()) self.assertEqual(response.status_code, 403) class IdentificationConcurrencyTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def setUp(self): self.brand, self.category, self.model, self.variant = make_catalog() def first(self): return set_identification_policy(product_model=self.model, **requirements()) def second(self): return set_identification_policy(product_model=self.model, **requirements(R.REQUIRED, R.REQUIRED)) def test_concurrent_first_service_creation_serializes(self): self.run_concurrent(self.first, self.second, expected="success") self.assertEqual(Policy.objects.count(), 1) self.assertEqual(Policy.objects.get().imei2_requirement, R.REQUIRED) def test_concurrent_service_updates_serialize(self): original = self.first() self.run_concurrent(self.first, self.second, expected="success") self.assertEqual(Policy.objects.get().pk, original.pk) self.assertEqual(Policy.objects.get().imei2_requirement, R.REQUIRED) def test_database_unique_constraint_protects_bypassing_creators(self): def raw_create(): Policy.objects.bulk_create([Policy(product_model=self.model, **requirements())]) self.run_concurrent(raw_create, raw_create, expected="integrity") self.assertEqual(Policy.objects.count(), 1) def test_policy_update_then_model_deactivation_preserves_configuration(self): self.run_concurrent(self.first, lambda: services.deactivate_product_model(product_model=self.model), expected="success") self.model.refresh_from_db() self.assertFalse(self.model.is_active) self.assertEqual(Policy.objects.get().imei2_requirement, R.OPTIONAL)