import uuid from unittest.mock import patch 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 from django.urls import reverse from apps.catalog.models import ProductCategory from apps.catalog import services as catalog_services from apps.access.authorization import is_authorized from .models import ServiceCategory, ComplaintSymptom, ComplaintSymptomProductCategory as Mapping from . import services from .queries import applicable_complaints_for_category, complaint_applies_to_category class TaxonomyFixture(TestCase): @classmethod def setUpTestData(cls): cls.category = ProductCategory.objects.create(code="A", name="A") cls.other_category = ProductCategory.objects.create(code="B", name="B") cls.service_category = ServiceCategory.objects.create(code="EXAMPLE", name="Example") cls.complaint = ComplaintSymptom.objects.create(code="EXAMPLE", name="Example") def configure(self, categories=(), global_mode=False): return services.set_complaint_applicability(complaint=self.complaint, applies_to_all_product_categories=global_mode, product_categories=categories) def applies(self, category=None): return complaint_applies_to_category(complaint=self.complaint, product_category=self.category if category is None else category) class TaxonomyModelTests(TaxonomyFixture): def test_uuid_timestamps_active_defaults_and_safe_applicability(self): for obj in [self.service_category, self.complaint]: self.assertIsInstance(obj.pk, uuid.UUID) self.assertIsNotNone(obj.created_at) self.assertIsNotNone(obj.updated_at) self.assertTrue(obj.is_active) self.assertEqual(obj.description, "") self.assertFalse(self.complaint.applies_to_all_product_categories) self.assertFalse(self.applies()) def test_normalization_code_correction_and_text_preservation(self): for obj in [self.service_category, self.complaint]: before = obj.updated_at obj.code, obj.name, obj.description = " corrected-1 ", "Mixed Name", "Mixed Description" obj.save(update_fields=["code", "name", "description"]) obj.refresh_from_db() self.assertEqual(obj.code, "CORRECTED-1") self.assertEqual(obj.name, "Mixed Name") self.assertEqual(obj.description, "Mixed Description") self.assertGreater(obj.updated_at, before) def test_required_and_invalid_fields(self): for model in [ServiceCategory, ComplaintSymptom]: for values in [dict(code="", name="X"), dict(code="NEW", name=""), dict(code="INVALID CODE", name="X"), dict(code="_BAD", name="X")]: with self.subTest(model=model.__name__, values=values), self.assertRaises(ValidationError): model.objects.create(**values) def test_uniqueness_case_equivalence_and_database_codes(self): for model in [ServiceCategory, ComplaintSymptom]: with self.assertRaises(ValidationError): model.objects.create(code=" example ", name="Duplicate") for code in ["EXAMPLE", "lower", " BAD", "", "BAD CODE"]: with self.subTest(model=model.__name__, code=code): with self.assertRaises(IntegrityError), transaction.atomic(): model.objects.bulk_create([model(code=code, name="Bad")]) def test_mapping_identity_timestamps_duplicates_and_protection(self): self.configure([self.category]) mapping = Mapping.objects.get() self.assertIsInstance(mapping.pk, uuid.UUID) self.assertIsNotNone(mapping.created_at) self.assertIsNotNone(mapping.updated_at) with self.assertRaises(ValidationError): Mapping.objects.create(complaint_symptom=self.complaint, product_category=self.category) with self.assertRaises(IntegrityError), transaction.atomic(): Mapping.objects.bulk_create([Mapping(complaint_symptom=self.complaint, product_category=self.category)]) for parent in [self.category, self.complaint]: with self.assertRaises(ProtectedError): parent.delete() self.service_category.delete() self.assertTrue(ComplaintSymptom.objects.filter(pk=self.complaint.pk).exists()) def test_mapping_endpoints_cannot_be_reparented(self): self.configure([self.category]) mapping = Mapping.objects.get() mapping.product_category = self.other_category with self.assertRaises(ValidationError): mapping.save(update_fields=["product_category"]) mapping.refresh_from_db() self.assertEqual(mapping.product_category, self.category) def test_direct_global_and_mapping_writes_reject_contradiction(self): self.configure([self.category]) self.complaint.applies_to_all_product_categories = True with self.assertRaises(ValidationError): self.complaint.save() self.configure(global_mode=True) with self.assertRaises(ValidationError): Mapping.objects.create(complaint_symptom=self.complaint, product_category=self.other_category) def test_active_helpers_and_representations(self): for obj in [self.service_category, self.complaint]: self.assertIn(obj, type(obj).objects.active()) with self.assertNumQueries(0): self.assertIn(obj.code, str(obj)) class ApplicabilityTests(TaxonomyFixture): def test_global_restricted_and_nowhere_transitions(self): self.assertFalse(self.applies()) self.configure(global_mode=True) self.assertTrue(self.applies()) self.assertTrue(self.applies(self.other_category)) self.configure([self.category]) self.assertTrue(self.applies()) self.assertFalse(self.applies(self.other_category)) self.configure() self.assertFalse(self.applies()) self.assertFalse(Mapping.objects.exists()) def test_replacement_deduplicates_and_retains_unchanged_mapping(self): self.configure([self.category]) original = Mapping.objects.get() self.configure([self.category, self.category, self.other_category]) retained = Mapping.objects.get(pk=original.pk) self.assertEqual(retained.created_at, original.created_at) self.assertEqual(retained.updated_at, original.updated_at) self.assertEqual(Mapping.objects.count(), 2) self.configure([self.other_category]) self.assertEqual(list(Mapping.objects.values_list("product_category_id", flat=True)), [self.other_category.pk]) def test_invalid_configuration_leaves_previous_state(self): self.configure([self.category]) bad = ProductCategory(code="UNSAVED", name="Unsaved") for categories, global_mode in [([self.other_category], True), ([bad], False), ([object()], False), ([], "False")]: with self.assertRaises(ValidationError): self.configure(categories, global_mode) self.assertTrue(self.applies()) self.assertEqual(Mapping.objects.count(), 1) self.other_category.delete() with self.assertRaises(ValidationError): self.configure([self.other_category]) self.assertTrue(self.applies()) def test_failure_after_removal_rolls_back_mappings_and_flag(self): self.configure([self.category]) original = Mapping.objects.get() with patch.object(Mapping, "save", side_effect=RuntimeError("simulated insert failure")): with self.assertRaises(RuntimeError): self.configure([self.other_category]) self.assertEqual(Mapping.objects.get().pk, original.pk) self.assertTrue(self.applies()) self.configure(global_mode=True) with patch.object(Mapping, "save", side_effect=RuntimeError("simulated insert failure")): with self.assertRaises(RuntimeError): self.configure([self.category]) self.complaint.refresh_from_db() self.assertTrue(self.complaint.applies_to_all_product_categories) self.assertFalse(Mapping.objects.exists()) def test_failure_during_global_switch_restores_removed_mappings(self): self.configure([self.category]) with patch.object(ComplaintSymptom, "save", side_effect=RuntimeError("simulated save failure")): with self.assertRaises(RuntimeError): self.configure(global_mode=True) self.assertTrue(self.applies()) self.assertEqual(Mapping.objects.count(), 1) def test_lifecycle_retains_configuration_and_reactivation_restores_effect(self): self.configure([self.category]) original = Mapping.objects.get() services.deactivate_complaint(complaint=self.complaint) self.assertFalse(self.applies()) self.assertEqual(Mapping.objects.get().pk, original.pk) services.reactivate_complaint(complaint=self.complaint) self.assertTrue(self.applies()) catalog_services.deactivate_category(category=self.category) self.assertFalse(self.applies()) catalog_services.reactivate_category(category=self.category) self.assertTrue(self.applies()) self.assertEqual(Mapping.objects.get().updated_at, original.updated_at) def test_service_category_lifecycle_independent(self): self.configure([self.category]) services.deactivate_service_category(service_category=self.service_category) self.service_category.refresh_from_db() self.assertFalse(self.service_category.is_active) self.assertTrue(self.applies()) services.reactivate_service_category(service_category=self.service_category) self.service_category.refresh_from_db() self.assertTrue(self.service_category.is_active) def test_inactive_categories_can_be_configured_but_not_selected(self): catalog_services.deactivate_category(category=self.category) self.configure([self.category]) self.assertEqual(Mapping.objects.count(), 1) self.assertFalse(self.applies()) self.configure(global_mode=True) self.assertFalse(self.applies()) def test_query_sql_order_filters_and_no_duplicates(self): global_complaint = ComplaintSymptom.objects.create(code="A_GLOBAL", name="Global", applies_to_all_product_categories=True) unmapped = ComplaintSymptom.objects.create(code="UNMAPPED", name="Unmapped") inactive = ComplaintSymptom.objects.create(code="INACTIVE", name="Inactive", is_active=False, applies_to_all_product_categories=True) self.configure([self.category, self.other_category]) with self.assertNumQueries(0): queryset = applicable_complaints_for_category(self.category) with self.assertNumQueries(1): self.assertEqual(list(queryset), [global_complaint, self.complaint]) self.assertNotIn(unmapped, queryset) self.assertNotIn(inactive, queryset) self.assertEqual(list(applicable_complaints_for_category(self.category).filter(pk=self.complaint.pk)), [self.complaint]) with self.assertNumQueries(1): self.assertTrue(self.applies()) def test_queries_use_current_state_and_reject_invalid_inputs(self): self.configure(global_mode=True) self.category.is_active = False self.category.save(update_fields=["is_active"]) stale = ProductCategory.objects.get(pk=self.category.pk) stale.is_active = True self.assertFalse(self.applies(stale)) for category in [None, ProductCategory(code="UNSAVED", name="Unsaved")]: self.assertFalse(applicable_complaints_for_category(category).exists()) self.assertFalse(complaint_applies_to_category(complaint=ComplaintSymptom(code="UNSAVED", name="Unsaved"), product_category=self.other_category)) self.assertFalse(complaint_applies_to_category(complaint=None, product_category=self.other_category)) def test_corrupt_global_plus_mapping_fails_closed(self): self.configure([self.category]) # Deliberate unsupported corruption: query should not expose it globally. ComplaintSymptom.objects.filter(pk=self.complaint.pk).update(applies_to_all_product_categories=True) self.assertFalse(self.applies()) self.assertFalse(self.applies(self.other_category)) class TaxonomyAdminTests(TaxonomyFixture): @classmethod def setUpTestData(cls): super().setUpTestData() cls.admin = get_user_model().objects.create_superuser(username="taxonomy-admin", password="test-only-password") def setUp(self): self.client.force_login(self.admin) def edit(self, categories=(), global_mode=False, name="Changed"): data = {"code": self.complaint.code, "name": name, "description": "", "_save": "Save", "product_categories": [str(category.pk) for category in categories]} if global_mode: data["global_applicability"] = "on" return self.client.post(reverse("admin:service_catalog_complaintsymptom_change", args=[self.complaint.pk]), data) def test_registration_native_permissions_and_authorization_boundary(self): for obj in [self.service_category, self.complaint]: self.assertEqual(self.client.get(reverse(f"admin:service_catalog_{obj._meta.model_name}_changelist")).status_code, 200) self.assertEqual(Permission.objects.filter(content_type__app_label="service_catalog", content_type__model=obj._meta.model_name).count(), 4) self.assertFalse(is_authorized(user=self.admin, permission=f"service_catalog.view_{obj._meta.model_name}", target=obj)) def test_admin_transitions_use_applicability_service(self): with patch.object(services, "set_complaint_applicability", wraps=services.set_complaint_applicability) as operation: self.assertEqual(self.edit([self.category]).status_code, 302) operation.assert_called_once() self.assertTrue(self.applies()) self.assertEqual(self.edit(global_mode=True).status_code, 302) self.assertTrue(self.applies(self.other_category)) self.assertFalse(Mapping.objects.exists()) self.assertEqual(self.edit([self.other_category]).status_code, 302) self.assertFalse(self.applies()) self.assertTrue(self.applies(self.other_category)) def test_admin_add_uses_service(self): with patch.object(services, "set_complaint_applicability", wraps=services.set_complaint_applicability) as operation: response = self.client.post(reverse("admin:service_catalog_complaintsymptom_add"), { "code": "NEW", "name": "New", "is_active": "on", "product_categories": [str(self.category.pk)], }) self.assertEqual(response.status_code, 302) operation.assert_called_once() self.assertTrue(Mapping.objects.filter(complaint_symptom__code="NEW", product_category=self.category).exists()) def test_admin_invalid_contradiction_preserves_configuration(self): self.configure([self.category]) response = self.edit([self.other_category], global_mode=True) self.assertContains(response, "Global complaints cannot have explicit category mappings") self.assertTrue(self.applies()) self.assertFalse(self.applies(self.other_category)) def test_admin_service_failure_rolls_back_record_and_mappings(self): self.configure([self.category]) with patch.object(Mapping, "save", side_effect=RuntimeError("simulated insert failure")): with self.assertRaises(RuntimeError): self.edit([self.other_category], name="Must roll back") self.complaint.refresh_from_db() self.assertEqual(self.complaint.name, "Example") self.assertTrue(self.applies()) def test_lifecycle_action_retains_mapping_and_calls_service(self): self.configure([self.category]) with patch.object(services, "deactivate_complaint", wraps=services.deactivate_complaint) as operation: response = self.client.post(reverse("admin:service_catalog_complaintsymptom_changelist"), { "action": "deactivate_selected", "_selected_action": [str(self.complaint.pk)], }) self.assertEqual(response.status_code, 302) operation.assert_called_once() self.assertFalse(self.applies()) self.assertEqual(Mapping.objects.count(), 1) def test_view_only_staff_cannot_change_applicability(self): reader = get_user_model().objects.create_user(username="taxonomy-reader", is_staff=True) reader.user_permissions.add(Permission.objects.get(content_type__app_label="service_catalog", codename="view_complaintsymptom")) self.client.force_login(reader) with patch.object(services, "set_complaint_applicability") as operation: self.assertEqual(self.edit(global_mode=True).status_code, 403) operation.assert_not_called()