"""Focused configuration, installation and development bootstrap contracts.""" from datetime import time, timedelta from io import StringIO import json 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.core.management import call_command, CommandError from django.db import connection, OperationalError from django.test import TestCase, TransactionTestCase, Client, override_settings from django.test.utils import CaptureQueriesContext from django.urls import reverse from apps.frontdesk.tests import prepare from apps.frontdesk import services as desk from apps.frontdesk.models import AppointmentSlot from apps.operations.tests import grant from apps.organization.models import Company from apps.organization import services as organization from apps.organization import test_assignment_concurrency as concurrency from apps.service.services import cancel_service_case from apps.service.models import ServiceCase from apps.devices.models import Device from apps.customers.models import Customer from apps.access.models import Role from apps.access.services import set_role_permissions from apps.catalog.models import ProductModel, DeviceIdentificationPolicy from apps.service_catalog.models import RootCause from apps.communications.models import Notification from apps.inventory.models import StockLedgerEntry from . import demo from .readiness import inspect_installation,is_ready class ConfigurationTests(TestCase): @classmethod def setUpTestData(cls): prepare(cls) cls.admin.is_staff=True cls.admin.save(update_fields=["is_staff"]) cls.manager=get_user_model().objects.create_user(username="slot-configuration") cls.role=grant(cls.manager,cls.company,["frontdesk.manage_slots"],cls.center) def test_anonymous_and_unprivileged_denied(self): self.assertEqual(self.client.get("/settings/").status_code,302) self.client.force_login(self.user) self.assertEqual(self.client.get("/settings/").status_code,403) self.assertEqual(self.client.get("/settings/readiness/").status_code,403) def test_staff_and_native_org_permissions_are_not_global_configuration_access(self): self.user.is_staff=True self.user.save() self.user.user_permissions.set(Permission.objects.filter(content_type__app_label="organization")) self.client.force_login(self.user) self.assertEqual(self.client.get("/settings/").status_code,403) self.assertNotContains(self.client.get("/"),"Configuration center") def test_slot_manager_navigation_and_scoped_center_list(self): self.client.force_login(self.manager) response=self.client.get("/settings/") self.assertContains(response,"Appointment slots and capacity") self.assertNotContains(response,"Roles and permission membership") self.assertNotContains(response,"Readiness checks") self.assertContains(self.client.get("/"),"Configuration center") response=self.client.get("/settings/slots/") self.assertContains(response,reverse("configuration:center_slots",args=[self.center.pk])) self.assertNotContains(response,str(self.other_center.pk)) self.assertNotContains(response,str(self.center2.pk)) def test_superuser_inventory_links_use_existing_admin(self): self.client.force_login(self.admin) response=self.client.get("/settings/") self.assertContains(response,reverse("admin:organization_company_changelist")) self.assertContains(response,reverse("admin:catalog_deviceidentificationpolicy_changelist")) self.assertContains(response,reverse("admin:access_role_changelist")) self.assertContains(response,reverse("configuration:readiness")) self.assertTemplateUsed(response,"operations/base.html") def test_company_configuration_permissions_do_not_follow_center_scope(self): set_role_permissions(role=self.role,permissions=Permission.objects.filter(content_type__app_label="communications",codename="manage_templates")) self.client.force_login(self.manager) self.assertEqual(self.client.get("/settings/").status_code,403) grant(self.user,self.company,["communications.manage_templates","sla.manage_slapolicy"]) self.client.force_login(self.user) response=self.client.get("/settings/") self.assertContains(response,reverse("communications:templates")) self.assertContains(response,reverse("sla:policies")) self.assertNotContains(response,reverse("admin:organization_company_changelist")) def test_slot_only_manager_can_create_and_return_to_configuration(self): self.client.force_login(self.manager) url=reverse("configuration:slot_new",args=[self.center.pk]) response=self.client.post(url,dict(date=(self.today+timedelta(days=1)).isoformat(),start_time="09:00",end_time="10:00",capacity=3)) self.assertRedirects(response,reverse("configuration:center_slots",args=[self.center.pk])) self.assertEqual(AppointmentSlot.objects.filter(service_center=self.center,date=self.today+timedelta(days=1)).count(),1) def test_foreign_center_and_slot_idor_get_and_post(self): foreign=desk.create_slot(actor=self.admin,service_center=self.other_center,date=self.today+timedelta(days=1),start_time=time(8),end_time=time(9),capacity=2) self.client.force_login(self.manager) for url in [reverse("configuration:center_slots",args=[self.other_center.pk]), reverse("configuration:slot_new",args=[self.other_center.pk]), reverse("configuration:slot_edit",args=[self.center.pk,foreign.pk])]: self.assertEqual(self.client.get(url).status_code,404) self.assertEqual(self.client.post(reverse("configuration:slot_edit",args=[self.center.pk,foreign.pk]),dict(capacity=99,is_active="on")).status_code,404) foreign.refresh_from_db() self.assertEqual(foreign.capacity,2) def test_capacity_integrity_is_delegated_to_frozen_service(self): desk.book_appointment(actor=self.admin,slot=self.slot,customer=self.customer) desk.book_appointment(actor=self.admin,slot=self.slot,customer=self.customer) self.client.force_login(self.manager) response=self.client.post(reverse("configuration:slot_edit",args=[self.center.pk,self.slot.pk]),dict(capacity=1,is_active="on")) self.assertEqual(response.status_code,400) self.slot.refresh_from_db() self.assertEqual(self.slot.capacity,2) def test_post_csrf_and_readonly_landing(self): client=Client(enforce_csrf_checks=True) client.force_login(self.manager) self.assertEqual(client.post(reverse("configuration:slot_new",args=[self.center.pk]),{}).status_code,403) self.client.force_login(self.admin) self.assertEqual(self.client.post("/settings/",{}).status_code,405) self.assertEqual(self.client.post("/settings/readiness/",{}).status_code,405) def test_readiness_ui_is_global_admin_only_and_sanitized(self): self.client.force_login(self.manager) self.assertEqual(self.client.get("/settings/readiness/").status_code,403) self.client.force_login(self.admin) self.assertContains(self.client.get("/settings/readiness/"),"Installation readiness") with patch("apps.configuration.readiness.inspect_installation",side_effect=OperationalError("SECRET-PASSWORD")): response=self.client.get("/settings/readiness/") self.assertEqual(response.status_code,503) self.assertNotContains(response,"SECRET-PASSWORD",status_code=503) @override_settings(DEBUG=True,SETTINGS_MODULE="config.settings.development") class ReadinessAndSeedTests(TestCase): def checks(self): rows=inspect_installation() return {r["code"]:r for r in rows} def admin(self): return get_user_model().objects.create_superuser(username="installation-admin",password="Synthetic-installation-test-only-42!") def test_fresh_database_is_not_ready(self): checks=self.checks() self.assertTrue(checks["migrations"]["ready"]) for code in ("company","centers","administrator","catalog"): self.assertFalse(checks[code]["ready"]) out=StringIO() with self.assertRaises(CommandError): call_command("check_installation_readiness",stdout=out) self.assertIn("REQUIRED SETUP INCOMPLETE",out.getvalue()) self.assertEqual(Company.objects.count(),0) def test_partial_setup_not_mistaken_for_ready(self): self.admin() Company.objects.create(code="PARTIAL",name="Partial setup") checks=self.checks() self.assertTrue(checks["company"]["ready"]) self.assertTrue(checks["administrator"]["ready"]) self.assertFalse(checks["centers"]["ready"]) self.assertFalse(checks["catalog"]["ready"]) def test_minimum_ready_with_optional_features_missing(self): demo.seed_demo() self.admin() rows=inspect_installation() self.assertTrue(is_ready(rows)) self.assertFalse(self.checks()["root_causes"]["ready"]) self.assertFalse(self.checks()["slots"]["ready"]) out=StringIO() call_command("check_installation_readiness",stdout=out,json=True) self.assertTrue(json.loads(out.getvalue())["ready"]) def test_each_active_company_needs_a_usable_center(self): demo.seed_demo() self.admin() Company.objects.create(code="UNCONFIGURED",name="Unconfigured") self.assertFalse(self.checks()["centers"]["ready"]) def test_inactive_catalog_parent_is_not_registerable(self): rows=demo.seed_demo() from apps.catalog.services import deactivate_brand deactivate_brand(brand=rows["device"].product_model.brand) self.assertFalse(self.checks()["catalog"]["ready"]) def test_readiness_runs_only_select_queries_and_never_provider_calls(self): demo.seed_demo() with patch("apps.communications.providers.get_provider") as provider, CaptureQueriesContext(connection) as captured: inspect_installation() provider.assert_not_called() writes=[r["sql"] for r in captured if r["sql"].lstrip().upper().startswith(("INSERT","UPDATE","DELETE","CREATE","ALTER"))] self.assertEqual(writes,[]) def test_database_error_redaction(self): with patch("apps.configuration.management.commands.check_installation_readiness.inspect_installation",side_effect=OperationalError("SECRET")): with self.assertRaises(CommandError) as caught: call_command("check_installation_readiness",stdout=StringIO()) self.assertNotIn("SECRET",str(caught.exception)) def test_unapplied_migrations_stop_before_domain_queries(self): with patch("apps.configuration.readiness.MigrationExecutor") as executor: executor.return_value.loader.detect_conflicts.return_value={} executor.return_value.migration_plan.return_value=[object()] rows=inspect_installation() self.assertEqual(len(rows),1) self.assertFalse(is_ready(rows)) def test_explicit_confirmation_required(self): with self.assertRaises(CommandError): call_command("seed_demo_data",stdout=StringIO()) self.assertFalse(Company.objects.exists()) def test_seed_coherent_nonstaff_chain_without_outbound_or_financial_data(self): with patch("apps.communications.services.request_notification") as queue: rows=demo.seed_demo() queue.assert_not_called() self.assertFalse(rows["user"].is_superuser) self.assertFalse(rows["user"].is_staff) self.assertFalse(rows["user"].has_usable_password()) self.assertEqual(rows["case"].status,"RECEIVED") self.assertEqual(rows["case"].company,rows["customer"].company) self.assertEqual(rows["case"].device,rows["device"]) self.assertFalse(Notification.objects.exists()) self.assertFalse(StockLedgerEntry.objects.exists()) self.assertFalse(RootCause.objects.exists()) from apps.commercial.models import ServiceInvoice,ServicePayment self.assertFalse(ServiceInvoice.objects.exists()) self.assertFalse(ServicePayment.objects.exists()) self.client.force_login(rows["user"]) self.assertContains(self.client.get("/"),"Open service cases") def test_seed_repeated_command_is_idempotent(self): first=demo.seed_demo() before={model._meta.label:model.objects.count() for model in (Company,Customer,Device,ServiceCase,Role,ProductModel,DeviceIdentificationPolicy)} out=StringIO() call_command("seed_demo_data",confirm_development=True,stdout=out) call_command("seed_demo_data",confirm_development=True,stdout=out) after={model._meta.label:model.objects.count() for model in (Company,Customer,Device,ServiceCase,Role,ProductModel,DeviceIdentificationPolicy)} self.assertEqual(before,after) self.assertEqual(demo.seed_demo()["case"].pk,first["case"].pk) self.assertIn("history preserved",out.getvalue()) def test_completed_demo_activity_is_not_reset(self): rows=demo.seed_demo() cancel_service_case(service_case=rows["case"],reason="Demo exercise",cancelled_by=rows["user"]) repeated=demo.seed_demo() self.assertEqual(repeated["case"].status,"CANCELLED") self.assertEqual(ServiceCase.objects.count(),1) def test_production_refused_even_if_debug_true(self): with override_settings(SETTINGS_MODULE="config.settings.production",DEBUG=True),self.assertRaises(CommandError): call_command("seed_demo_data",confirm_development=True,stdout=StringIO()) self.assertFalse(Company.objects.exists()) def test_debug_off_and_remote_database_refused(self): with override_settings(DEBUG=False),self.assertRaises(ValidationError): demo.seed_demo() with patch.dict(connection.settings_dict,{"HOST":"remote.example"}),self.assertRaises(ValidationError): demo.seed_demo() self.assertFalse(Company.objects.exists()) def test_reserved_company_collision_not_adopted(self): existing=Company.objects.create(code=demo.CODE,name="Existing unrelated company") with self.assertRaises(ValidationError): demo.seed_demo() existing.refresh_from_db() self.assertEqual(existing.name,"Existing unrelated company") self.assertFalse(get_user_model().objects.filter(username=demo.USERNAME).exists()) def test_changed_role_is_not_silently_reprivileged(self): demo.seed_demo() role=Role.objects.get(code=demo.CODE) set_role_permissions(role=role,permissions=[]) with self.assertRaises(ValidationError): demo.seed_demo() self.assertEqual(role.permissions.count(),0) def test_failure_rolls_back_whole_seed(self): with patch("apps.configuration.demo.create_service_case",side_effect=ValidationError("Synthetic failure")),self.assertRaises(ValidationError): demo.seed_demo() self.assertFalse(Company.objects.exists()) self.assertFalse(Customer.objects.exists()) self.assertFalse(ProductModel.objects.exists()) self.assertFalse(get_user_model().objects.filter(username=demo.USERNAME).exists()) def test_deactivated_configuration_is_not_reactivated(self): rows=demo.seed_demo() organization.deactivate_company(company=rows["company"]) with self.assertRaises(ValidationError): demo.seed_demo() rows["company"].refresh_from_db() self.assertFalse(rows["company"].is_active) def test_password_only_by_explicit_prompt_never_echoed(self): password="Synthetic-demo-test-only-strong-42!" out=StringIO() with patch("apps.configuration.management.commands.seed_demo_data.getpass",side_effect=[password,password]): call_command("seed_demo_data",confirm_development=True,set_password=True,stdout=out) user=get_user_model().objects.get(username=demo.USERNAME) self.assertTrue(user.check_password(password)) self.assertNotIn(password,out.getvalue()) before=user.password demo.seed_demo() user.refresh_from_db() self.assertEqual(user.password,before) @override_settings(DEBUG=True,SETTINGS_MODULE="config.settings.development") class SeedConcurrencyTests(TransactionTestCase): run_concurrent=concurrency.AssignmentConcurrencyTests.run_concurrent def test_concurrent_seed_has_one_chain(self): self.run_concurrent(demo.seed_demo,demo.seed_demo,expected="success") self.assertEqual(Company.objects.filter(code=demo.CODE).count(),1) self.assertEqual(ServiceCase.objects.count(),1) self.assertEqual(Customer.objects.count(),1)