"""Phase 2A freeze audit: cross-domain integration and stale/admin attacks.""" from unittest.mock import patch from django.contrib import admin from django.contrib.auth import get_user_model from django.test import Client, TestCase, TransactionTestCase from django.urls import reverse from apps.catalog import services as catalog_services from django.core.exceptions import ValidationError from django.db import IntegrityError, transaction from django.db.models.deletion import ProtectedError from apps.catalog import models as catalog_models from apps.catalog.models import Brand, ProductCategory, ProductModel, ProductVariant, DeviceIdentificationPolicy from apps.catalog.identification import get_identification_policy, set_identification_policy from apps.catalog.tests import make_catalog from apps.service_catalog import models as taxonomy_models, queries from apps.service_catalog import test_technical as technical_fixtures from apps.organization import test_assignment_concurrency as concurrency from apps.access.authorization import authorized_queryset, is_authorized from apps.service_catalog import services as taxonomy_services from apps.service_catalog.models import ComplaintSymptom class StaleAdminAuditTests(TestCase): @classmethod def setUpTestData(cls): cls.admin_user = get_user_model().objects.create_superuser(username="freeze-audit", password="test-only-password") def setUp(self): self.client.force_login(self.admin_user) def test_catalog_metadata_edit_does_not_reverse_completed_deactivation(self): obj = Brand.objects.create(code="AUDIT", name="Before") registered = admin.site._registry[Brand] original = registered.save_model def interleave(request, instance, form, change): # The form already holds an active instance. A lifecycle operation # completes before this request saves its metadata, as in a stale write. catalog_services.deactivate_brand(brand=obj) return original(request, instance, form, change) with patch.object(registered, "save_model", side_effect=interleave): response = self.client.post(reverse("admin:catalog_brand_change", args=[obj.pk]), { "code": obj.code, "name": "After", "description": "", "_save": "Save", }) self.assertEqual(response.status_code, 302) obj.refresh_from_db() self.assertEqual(obj.name, "After") self.assertFalse(obj.is_active) def test_taxonomy_metadata_edit_does_not_reverse_completed_deactivation(self): obj = ComplaintSymptom.objects.create(code="AUDIT", name="Before") registered = admin.site._registry[ComplaintSymptom] original = registered.save_model def interleave(request, instance, form, change): taxonomy_services.deactivate_complaint(complaint=obj) return original(request, instance, form, change) with patch.object(registered, "save_model", side_effect=interleave): response = self.client.post(reverse("admin:service_catalog_complaintsymptom_change", args=[obj.pk]), { "code": obj.code, "name": "After", "description": "", "_save": "Save", }) self.assertEqual(response.status_code, 302) obj.refresh_from_db() self.assertEqual(obj.name, "After") self.assertFalse(obj.is_active) TAXONOMIES = ( (taxonomy_models.ComplaintSymptom, taxonomy_models.ComplaintSymptomProductCategory, "complaint_symptom", "complaint", "set_complaint_applicability", "applicable_complaints_for_category", "complaint_applies_to_category", "deactivate_complaint", "reactivate_complaint"), *technical_fixtures.CASES, ) class MasterDataIntegrationAuditTests(TestCase): @classmethod def setUpTestData(cls): cls.brand, cls.category, cls.model, cls.variant = make_catalog("AUDIT") cls.other_brand, cls.other_category, cls.other_model, cls.other_variant = make_catalog("OTHER") cls.service_category = taxonomy_models.ServiceCategory.objects.create(code="SERVICE", name="Service") cls.records = {} for case in TAXONOMIES: master = case[0].objects.create(code="VOCAB", name="Vocabulary") getattr(taxonomy_services, case[4])(**{case[3]: master}, applies_to_all_product_categories=False, product_categories=[cls.category, cls.other_category]) cls.records[case[0]] = master R = DeviceIdentificationPolicy.Requirement cls.policy = set_identification_policy(product_model=cls.model, imei1_requirement=R.REQUIRED, imei2_requirement=R.OPTIONAL, serial_requirement=R.REQUIRED) def coded(self): return [self.brand, self.category, self.model, self.variant, self.service_category, *self.records.values()] def test_canonical_whitespace_duplicates_and_raw_control_characters_all_masters(self): for obj in self.coded(): original = obj.code for value in [original.lower(), " " + original.lower(), original.lower() + " ", "\t" + original.lower() + "\r\n", "\u00a0" + original.lower() + "\u00a0"]: with self.subTest(model=type(obj).__name__, normalized_input=repr(value)): obj.code = value obj.save(update_fields=["code"]) obj.refresh_from_db() self.assertEqual(obj.code, original) fields = {field.name: getattr(obj, field.name) for field in obj._meta.concrete_fields if field.name not in ("id", "created_at", "updated_at", "code")} with self.assertRaises(ValidationError): type(obj).objects.create(code=value, **fields) for invalid in ["ABC\n", "ABC\r\n", "AB\tC", "AB\u200bC", "\uff21BC", " "]: with self.subTest(model=type(obj).__name__, raw=repr(invalid)): with self.assertRaises(IntegrityError), transaction.atomic(): type(obj).objects.filter(pk=obj.pk).update(code=invalid) def test_complete_category_cascade_preserves_policy_and_all_taxonomy_configurations(self): before = {case[0]: list(case[1].objects.values_list("pk", "created_at", "updated_at")) for case in TAXONOMIES} catalog_services.deactivate_category(category=self.category) for obj in [self.category, self.model, self.variant]: obj.refresh_from_db() self.assertFalse(obj.is_active) for obj in [self.brand, self.other_brand, self.other_category, self.other_model, self.other_variant]: obj.refresh_from_db() self.assertTrue(obj.is_active) self.assertEqual(get_identification_policy(self.model).pk, self.policy.pk) for case in TAXONOMIES: self.assertEqual(list(case[1].objects.values_list("pk", "created_at", "updated_at")), before[case[0]]) self.assertFalse(getattr(queries, case[5])(self.category).exists()) self.assertEqual(list(getattr(queries, case[5])(self.other_category)), [self.records[case[0]]]) catalog_services.reactivate_category(category=self.category) for case in TAXONOMIES: self.assertEqual(list(getattr(queries, case[5])(self.category)), [self.records[case[0]]]) self.model.refresh_from_db() self.variant.refresh_from_db() self.assertFalse(self.model.is_active) self.assertFalse(self.variant.is_active) def test_category_correction_preserves_policy_and_does_not_rewrite_taxonomy(self): before = {case[0]: set(case[1].objects.values_list("pk", flat=True)) for case in TAXONOMIES} self.model.category = self.other_category self.model.save(update_fields=["category"]) catalog_services.deactivate_category(category=self.category) self.model.refresh_from_db() self.variant.refresh_from_db() self.assertTrue(self.model.is_active) self.assertTrue(self.variant.is_active) self.assertEqual(get_identification_policy(self.model).pk, self.policy.pk) for case in TAXONOMIES: self.assertEqual(set(case[1].objects.values_list("pk", flat=True)), before[case[0]]) def test_late_catalog_cascade_failure_preserves_cross_domain_state(self): original = catalog_models.CatalogQuerySet.update def fail_category(queryset, **kwargs): if queryset.model is ProductCategory: raise RuntimeError("simulated final cascade failure") return original(queryset, **kwargs) with patch.object(catalog_models.CatalogQuerySet, "update", fail_category): with self.assertRaises(RuntimeError): catalog_services.deactivate_category(category=self.category) for obj in [self.category, self.model, self.variant]: obj.refresh_from_db() self.assertTrue(obj.is_active) self.assertEqual(get_identification_policy(self.model).updated_at, self.policy.updated_at) for case in TAXONOMIES: self.assertEqual(getattr(queries, case[5])(self.category).count(), 1) self.assertEqual(case[1].objects.count(), 2) def test_all_public_applicability_replacements_restore_complete_old_set_after_failure(self): for case in TAXONOMIES: original = list(case[1].objects.values_list("pk", "product_category_id", "updated_at")) with patch.object(case[0], "save", side_effect=RuntimeError("simulated mode update failure")): with self.assertRaises(RuntimeError): getattr(taxonomy_services, case[4])(**{case[3]: self.records[case[0]]}, applies_to_all_product_categories=True, product_categories=[]) self.assertEqual(list(case[1].objects.values_list("pk", "product_category_id", "updated_at")), original) def test_lookup_query_counts_remain_bounded_with_many_results(self): for case in TAXONOMIES: for number in range(12): case[0].objects.create(code=f"GLOBAL_{number:02}", name="Global", applies_to_all_product_categories=True) with self.assertNumQueries(0): result = getattr(queries, case[5])(self.category) with self.assertNumQueries(1): records = list(result) self.assertEqual(len(records), 13) self.assertEqual([obj.code for obj in records], sorted(obj.code for obj in records)) with self.assertNumQueries(1): self.assertTrue(getattr(queries, case[6])(**{case[3]: self.records[case[0]]}, product_category=self.category)) with self.assertNumQueries(2): self.assertEqual(get_identification_policy(self.model).pk, self.policy.pk) with self.assertNumQueries(2): self.assertIsNone(get_identification_policy(self.other_model)) def test_querysets_evaluate_fresh_state_but_materialized_results_are_snapshots(self): for case in TAXONOMIES: cached = list(getattr(queries, case[5])(self.category)) pending = getattr(queries, case[5])(self.category) getattr(taxonomy_services, case[7])(**{case[3]: self.records[case[0]]}) self.assertEqual(list(pending), []) self.assertEqual(cached, [self.records[case[0]]]) getattr(taxonomy_services, case[8])(**{case[3]: self.records[case[0]]}) self.assertEqual(len(list(getattr(queries, case[5])(self.category))), 1) def test_all_phase_two_models_remain_unsupported_authorization_targets(self): user = get_user_model().objects.create_superuser(username="boundary-audit", password="test-only-password") targets = [*self.coded(), self.policy, *[case[1].objects.first() for case in TAXONOMIES]] for obj in targets: permission = f"{obj._meta.app_label}.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()) def test_policy_and_mapping_deletion_protect_configuration_parents(self): self.variant.delete() with self.assertRaises(ProtectedError): self.model.delete() # Remove catalog children in the isolated test so category PROTECT is # proven independently for each taxonomy mapping type. self.policy.delete() self.model.delete() for case in TAXONOMIES: with self.assertRaises(ProtectedError): self.records[case[0]].delete() with self.assertRaises(ProtectedError): self.category.delete() def test_admin_cannot_reparent_variant_or_post_readonly_activation(self): user = get_user_model().objects.create_superuser(username="variant-audit", password="test-only-password") self.client.force_login(user) response = self.client.post(reverse("admin:catalog_productvariant_change", args=[self.variant.pk]), { "code": self.variant.code, "name": "Changed", "description": "", "product_model": str(self.other_model.pk), "_save": "Save", }) self.assertEqual(response.status_code, 302) self.variant.refresh_from_db() self.assertEqual(self.variant.product_model_id, self.model.pk) self.assertTrue(self.variant.is_active) def test_csrf_and_view_only_admin_cannot_mutate_new_master_types(self): from django.contrib.auth.models import Permission reader = get_user_model().objects.create_user(username="master-reader", is_staff=True) for case in TAXONOMIES: reader.user_permissions.add(Permission.objects.get(content_type__app_label="service_catalog", codename=f"view_{case[0]._meta.model_name}")) self.client.force_login(reader) for case in TAXONOMIES: obj = self.records[case[0]] url = reverse(f"admin:service_catalog_{obj._meta.model_name}_change", args=[obj.pk]) self.assertEqual(self.client.post(url, {"code": obj.code, "name": "Forbidden", "global_applicability": "on"}).status_code, 403) protected = Client(enforce_csrf_checks=True) protected.force_login(get_user_model().objects.create_superuser(username="csrf-audit", password="test-only-password")) self.assertEqual(protected.post(reverse("admin:catalog_brand_add"), {"code": "CSRF", "name": "CSRF"}).status_code, 403) self.assertEqual(protected.post(reverse("admin:service_catalog_faultdiagnosis_add"), {"code": "CSRF", "name": "CSRF"}).status_code, 403) class AdminConcurrencyAuditTests(TransactionTestCase): run_concurrent = concurrency.AssignmentConcurrencyTests.run_concurrent def test_admin_edit_waits_for_brand_deactivation_without_reactivation(self): brand = Brand.objects.create(code="RACE", name="Before") user = get_user_model().objects.create_superuser(username="race-admin", password="test-only-password") def edit(): client = Client() client.force_login(user) response = client.post(reverse("admin:catalog_brand_change", args=[brand.pk]), { "code": brand.code, "name": "After", "description": "", "_save": "Save", }) self.assertEqual(response.status_code, 302) self.run_concurrent(lambda: catalog_services.deactivate_brand(brand=brand), edit, expected="success") brand.refresh_from_db() self.assertEqual(brand.name, "After") self.assertFalse(brand.is_active)