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.access.authorization import is_authorized, authorized_queryset from . import services from .models import Brand, ProductCategory, ProductModel, ProductVariant def make_catalog(prefix="A"): brand = Brand.objects.create(code=prefix, name=prefix) category = ProductCategory.objects.create(code=prefix, name=prefix) model = ProductModel.objects.create(brand=brand, category=category, code="M1", name="Model") variant = ProductVariant.objects.create(product_model=model, code="8-256", name="8GB + 256GB") return brand, category, model, variant class CatalogFixture(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 objects(self): return [self.brand, self.category, self.model, self.variant] def assert_active(self, objects, active): for obj in objects: obj.refresh_from_db() self.assertEqual(obj.is_active, active) class CatalogModelTests(CatalogFixture): def test_uuid_timestamps_defaults_and_relationships(self): for obj in self.objects(): 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.assertEqual(self.model.brand, self.brand) self.assertEqual(self.model.category, self.category) self.assertEqual(self.variant.product_model, self.model) def test_normalization_code_correction_and_text_preservation(self): for obj in self.objects(): obj.code, obj.name, obj.description = " fixed-1 ", "Mixed Name", "Mixed Description" previous = obj.updated_at obj.save(update_fields=["code", "name", "description"]) obj.refresh_from_db() self.assertEqual(obj.code, "FIXED-1") self.assertEqual(obj.name, "Mixed Name") self.assertEqual(obj.description, "Mixed Description") self.assertGreater(obj.updated_at, previous) def test_required_fields_and_invalid_codes(self): for obj in self.objects(): for field, value in [("code", ""), ("code", "bad code"), ("code", "_BAD"), ("name", "")]: with self.subTest(model=type(obj).__name__, field=field, value=value): original = getattr(obj, field) setattr(obj, field, value) with self.assertRaises(ValidationError): obj.save() setattr(obj, field, original) def test_canonical_codes_enforced_by_database(self): for obj in self.objects(): for code in ["lower", " BAD", "BAD CODE", ""]: with self.subTest(model=type(obj).__name__, code=code): with self.assertRaises(IntegrityError), transaction.atomic(): type(obj).objects.filter(pk=obj.pk).update(code=code) def test_unique_and_case_equivalent_codes(self): cases = [(Brand, {"code": " a "}), (ProductCategory, {"code": " a "}), (ProductModel, {"brand": self.brand, "category": self.category, "code": " m1 "}), (ProductVariant, {"product_model": self.model, "code": " 8-256 "})] for model, values in cases: with self.subTest(model=model.__name__), self.assertRaises(ValidationError): model.objects.create(name="Duplicate", **values) values["code"] = values["code"].strip().upper() with self.assertRaises(IntegrityError), transaction.atomic(): model.objects.bulk_create([model(name="Duplicate", **values)]) self.assertEqual(self.model.code, self.other_model.code) self.assertEqual(self.variant.code, self.other_variant.code) def test_model_requires_active_brand_and_category(self): for parent in [self.brand, self.category]: with transaction.atomic(): services._deactivate(parent) with self.assertRaises(ValidationError): ProductModel.objects.create(brand=self.brand, category=self.category, code="NEW", name="New") transaction.set_rollback(True) def test_variant_requires_active_model_and_ancestors(self): for parent in [self.brand, self.category, self.model]: with transaction.atomic(): # Deliberately bypass lifecycle to test indirect parent validation. type(parent).objects.filter(pk=parent.pk).update(is_active=False) with self.assertRaises(ValidationError): ProductVariant.objects.create(product_model=self.model, code="NEW", name="New") transaction.set_rollback(True) def test_inactive_history_under_inactive_parents(self): services.deactivate_brand(brand=self.brand) model = ProductModel.objects.create(brand=self.brand, category=self.category, code="HISTORY", name="History", is_active=False) variant = ProductVariant.objects.create(product_model=model, code="HISTORY", name="History", is_active=False) self.assertFalse(variant.is_active) def test_ownership_stability(self): self.model.brand = self.other_brand self.variant.product_model = self.other_model for obj in [self.model, self.variant]: with self.assertRaises(ValidationError): obj.save() def test_category_correction_with_existing_variants(self): self.model.category = self.other_category self.model.save(update_fields=["category"]) services.deactivate_category(category=self.category) self.assert_active([self.model, self.variant], True) services.deactivate_category(category=self.other_category) self.assert_active([self.model, self.variant], False) def test_active_category_correction_rejects_inactive_destination(self): services.deactivate_category(category=self.other_category) self.model.category = self.other_category with self.assertRaises(ValidationError): self.model.save(update_fields=["category"]) def test_partial_save_checks_effective_persisted_state(self): services.deactivate_product_model(product_model=self.model) services.deactivate_category(category=self.category) self.model.refresh_from_db() self.model.category = self.other_category self.model.is_active = True with self.assertRaises(ValidationError): self.model.save(update_fields=["is_active"]) self.assert_active([self.model], False) def test_direct_deactivation_rejected_without_cascade(self): for obj in [self.brand, self.category, self.model]: obj.is_active = False with self.assertRaises(ValidationError): obj.save() self.assert_active(self.objects(), True) def test_protected_parents_and_unreferenced_deletion(self): services.deactivate_brand(brand=self.brand) for obj in [self.brand, self.category, self.model]: with self.assertRaises(ProtectedError): obj.delete() self.variant.delete() self.model.delete() self.brand.delete() self.category.delete() def test_active_helpers_and_no_query_representation(self): for obj in self.objects(): self.assertIn(obj, type(obj).objects.active()) with self.assertNumQueries(0): self.assertIn(obj.code, str(obj)) services.deactivate_brand(brand=self.brand) self.assertNotIn(self.variant, ProductVariant.objects.active()) class CatalogLifecycleTests(CatalogFixture): def test_brand_cascade_scoped_and_atomic(self): services.deactivate_brand(brand=self.brand) self.assert_active([self.brand, self.model, self.variant], False) self.assert_active([self.category, self.other_brand, self.other_model, self.other_variant], True) def test_category_cascade_across_brands_only_matching_category(self): linked = ProductModel.objects.create(brand=self.other_brand, category=self.category, code="LINK", name="Linked") child = ProductVariant.objects.create(product_model=linked, code="X", name="X") services.deactivate_category(category=self.category) self.assert_active([self.category, self.model, self.variant, linked, child], False) self.assert_active([self.brand, self.other_brand, self.other_category, self.other_model, self.other_variant], True) def test_model_and_variant_deactivation_are_scoped(self): services.deactivate_variant(variant=self.variant) self.assert_active([self.model, self.brand, self.category], True) services.reactivate_variant(variant=self.variant) services.deactivate_product_model(product_model=self.model) self.assert_active([self.model, self.variant], False) self.assert_active([self.brand, self.category, self.other_variant], True) def test_top_down_reactivation_does_not_restore_descendants(self): services.deactivate_brand(brand=self.brand) with self.assertRaises(ValidationError): services.reactivate_product_model(product_model=self.model) with self.assertRaises(ValidationError): services.reactivate_variant(variant=self.variant) services.reactivate_brand(brand=self.brand) self.assert_active([self.model, self.variant], False) services.reactivate_product_model(product_model=self.model) self.assert_active([self.variant], False) services.reactivate_variant(variant=self.variant) self.assert_active(self.objects(), True) services.deactivate_category(category=self.category) services.reactivate_category(category=self.category) self.assert_active([self.model, self.variant], False) def test_cascade_failure_rolls_back_all_changes(self): from .models import CatalogQuerySet original = CatalogQuerySet.update for obj in [self.brand, self.category, self.model]: def fail_final(queryset, **kwargs): if queryset.model is type(obj): raise RuntimeError("simulated final update failure") return original(queryset, **kwargs) with patch.object(CatalogQuerySet, "update", fail_final): with self.assertRaises(RuntimeError): services._deactivate(obj) self.assert_active(self.objects(), True) def test_history_timestamp_retention_and_fresh_service_inputs(self): services.deactivate_variant(variant=self.variant) self.variant.refresh_from_db() timestamp = self.variant.updated_at self.brand.name = "Unsaved" services.deactivate_brand(brand=self.brand) self.variant.refresh_from_db() self.brand.refresh_from_db() self.assertEqual(self.variant.updated_at, timestamp) self.assertEqual(self.brand.name, "A") self.assertEqual(ProductVariant.objects.count(), 2) class CatalogAdminTests(CatalogFixture): def setUp(self): self.admin = get_user_model().objects.create_superuser(username="catalog-admin", password="test-only-password") self.client.force_login(self.admin) def test_lists_and_permission_generation(self): for obj in self.objects(): model = obj._meta.model_name self.assertEqual(self.client.get(reverse(f"admin:catalog_{model}_changelist")).status_code, 200) self.assertEqual(Permission.objects.filter(content_type__app_label="catalog", content_type__model=model).count(), 4) def test_admin_actions_delegate_and_reactivation_is_deliberate(self): url = reverse("admin:catalog_brand_changelist") with patch.object(services, "deactivate_brand", wraps=services.deactivate_brand) as operation: self.client.post(url, {"action": "deactivate_selected", "_selected_action": [str(self.brand.pk)]}) operation.assert_called_once() self.assert_active([self.brand, self.model, self.variant], False) self.client.post(url, {"action": "reactivate_selected", "_selected_action": [str(self.brand.pk)]}) self.assert_active([self.brand], True) self.assert_active([self.model, self.variant], False) def test_existing_activation_and_ownership_cannot_be_forged(self): response = self.client.post(reverse("admin:catalog_productmodel_change", args=[self.model.pk]), { "code": self.model.code, "name": self.model.name, "category": str(self.category.pk), "brand": str(self.other_brand.pk), "_save": "Save", }) self.assertEqual(response.status_code, 302) self.model.refresh_from_db() self.assertTrue(self.model.is_active) self.assertEqual(self.model.brand, self.brand) def test_admin_invalid_creation_and_reactivation_show_errors(self): services.deactivate_brand(brand=self.brand) response = self.client.post(reverse("admin:catalog_productmodel_add"), { "code": "NEW", "name": "New", "brand": str(self.brand.pk), "category": str(self.category.pk), "is_active": "on", "_save": "Save", }) self.assertContains(response, "requires active Brand and Category") response = self.client.post(reverse("admin:catalog_productmodel_changelist"), { "action": "reactivate_selected", "_selected_action": [str(self.model.pk)], }, follow=True) self.assertContains(response, "requires active Brand and Category") self.assert_active([self.model], False) def test_view_only_staff_cannot_run_lifecycle(self): user = get_user_model().objects.create_user(username="catalog-reader", is_staff=True) user.user_permissions.add(Permission.objects.get(content_type__app_label="catalog", codename="view_brand")) self.client.force_login(user) with patch.object(services, "deactivate_brand") as operation: self.client.post(reverse("admin:catalog_brand_changelist"), { "action": "deactivate_selected", "_selected_action": [str(self.brand.pk)], }) operation.assert_not_called() self.assert_active([self.brand], True) def test_catalog_not_added_to_organizational_authorization(self): for obj in self.objects(): permission = f"catalog.view_{obj._meta.model_name}" self.assertTrue(self.admin.has_perm(permission)) self.assertFalse(is_authorized(user=self.admin, permission=permission, target=obj)) self.assertFalse(authorized_queryset(user=self.admin, permission=permission, queryset=type(obj).objects.all()).exists())