"""Scoped Admin QC orchestration, CSRF, stale forms and readonly evidence.""" from unittest.mock import patch from django.contrib.auth.models import Permission from django.test import Client, TestCase from django.urls import reverse from apps.access import services as access from apps.organization import assignment_services as postings from . import quality_control_services as services, quality_control_queries as queries from .quality_control_admin import QualityControlForm from .models import ServiceQualityControl from .test_quality_control import setup_qc, prepared, fill, check, pass_qc, fail_qc, abandon, invariants class QualityControlAdminTests(TestCase): def setUp(self): setup_qc(self) self.inspector.is_staff = True self.inspector.save() self.inspector.user_permissions.add(Permission.objects.get(content_type__app_label="service", codename="change_servicecase")) self.client.force_login(self.inspector) self.url = reverse("admin:service_servicecase_quality_control", args=[self.case.pk]) def tearDown(self): invariants(self) def token(self): return self.client.get(self.url).context["form"].initial["revision"] def post(self, operation, **kwargs): return self.client.post(self.url, dict(operation=operation, revision=self.token(), **kwargs)) def test_admin_submit_begin_checks_summary_and_pass(self): self.assertEqual(self.post("submit").status_code, 302) with patch.object(services, "begin_quality_control", wraps=services.begin_quality_control) as spy: self.assertEqual(self.post("begin").status_code, 302) self.assertEqual(spy.call_args.kwargs["inspector"], self.inspector) row = ServiceQualityControl.objects.get() technical = queries.quality_control_checks(row).get(check_code="BOOT") complaint = queries.quality_control_complaint_checks(row).get() self.assertEqual(self.post("check", check=technical.pk, result="PASS", note="Synthetic verified").status_code, 302) self.assertEqual(self.post("complaint", complaint_check=complaint.pk, complaint_result="RESOLVED").status_code, 302) self.assertEqual(self.post("summary", summary="Synthetic measurements").status_code, 302) fill(self, row) self.assertEqual(self.post("pass", summary="Synthetic passed").status_code, 302) self.assertEqual(ServiceQualityControl.objects.get().outcome, "PASSED") def test_choices_display_check_names_and_original_complaints(self): row = prepared(self) form = self.client.get(self.url).context["form"] technical = queries.quality_control_checks(row).get(check_code="BOOT") complaint = queries.quality_control_complaint_checks(row).get() self.assertEqual(form.fields["check"].label_from_instance(technical), "1. Power on / boot (Unanswered)") self.assertIn("Synthetic reported fault", form.fields["complaint_check"].label_from_instance(complaint)) def test_fail_and_abandon_services(self): row = prepared(self) self.assertEqual(self.post("abandon", reason="Synthetic interruption").status_code, 302) self.assertEqual(self.post("begin").status_code, 302) row = queries.current_quality_control(self.case) technical = queries.quality_control_checks(row).get(check_code="BOOT") self.assertEqual(self.post("check", check=technical.pk, result="FAIL").status_code, 302) self.assertEqual(self.post("fail", reason="Synthetic boot failure").status_code, 302) self.assertEqual(queries.latest_completed_quality_control(self.case).outcome, "FAILED") def test_csrf_and_native_permission_required(self): client = Client(enforce_csrf_checks=True) client.force_login(self.inspector) self.assertEqual(client.post(self.url, {}).status_code, 403) self.inspector2.is_staff = True self.inspector2.save() self.client.force_login(self.inspector2) self.assertEqual(self.client.get(self.url).status_code, 403) def test_superuser_without_qc_posting_cannot_access(self): self.user.is_superuser = self.user.is_staff = True self.user.save() self.client.force_login(self.user) self.assertEqual(self.client.get(self.url).status_code, 403) self.assertEqual(self.client.post(self.url, {}).status_code, 403) def test_cross_company_posting_cannot_access_case_or_history(self): row = prepared(self) self.user.is_staff = self.user.is_superuser = True self.user.save() path = postings.create_assignment(user=self.user, company=self.other_company) access.create_role_assignment(user=self.user, role=self.qc_role, organization_assignment=path) self.client.force_login(self.user) self.assertEqual(self.client.get(self.url).status_code, 403) response = self.client.get(reverse("admin:service_servicequalitycontrol_changelist")) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["cl"].queryset.count(), 0) response = self.client.get(reverse("admin:service_servicequalitycontrol_change", args=[row.pk])) self.assertEqual(response.status_code, 302) def test_tampered_signed_revision_rejected(self): response = self.client.post(self.url, dict(operation="submit", revision=self.token() + "tampered")) self.assertContains(response, "Reload the QC form") self.case.refresh_from_db() self.assertEqual(self.case.status, "REPAIRED") def test_stale_check_after_another_check(self): row = prepared(self) token = self.token() target = queries.quality_control_checks(row).get(check_code="BOOT") check(self, row, result="FAIL") response = self.client.post(self.url, dict(operation="check", revision=token, check=target.pk, result="PASS")) self.assertContains(response, "changed; reload") target.refresh_from_db() self.assertEqual(target.result, "FAIL") def test_fresh_choice_cannot_bypass_old_aggregate_revision(self): row = prepared(self) token = self.token() target = queries.quality_control_checks(row).get(check_code="BOOT") original = QualityControlForm.clean def intervening(form): data = original(form) data["check"] = check(self, row, result="FAIL") return data with patch.object(QualityControlForm, "clean", intervening): response = self.client.post(self.url, dict(operation="check", revision=token, check=target.pk, result="PASS")) self.assertContains(response, "changed; reload") target.refresh_from_db() self.assertEqual(target.result, "FAIL") def test_stale_check_after_pass_denied(self): row = prepared(self, filled=True) token = self.token() target = queries.quality_control_checks(row).first() pass_qc(self, row) response = self.client.post(self.url, dict(operation="check", revision=token, check=target.pk, result="FAIL")) self.assertEqual(response.status_code, 403) def test_stale_form_after_abandonment_and_new_attempt(self): row = prepared(self) token = self.token() target = queries.quality_control_checks(row).first() abandon(self, row) self.assertEqual(self.post("begin").status_code, 302) response = self.client.post(self.url, dict(operation="check", revision=token, check=target.pk, result="PASS")) self.assertContains(response, "changed; reload") target.refresh_from_db() self.assertIsNone(target.result) def test_revoked_scope_blocks_stale_form(self): row = prepared(self) token = self.token() access.deactivate_role_assignment(assignment=self.qc_role_assignment) self.assertEqual(self.client.post(self.url, dict(operation="pass", revision=token)).status_code, 403) row.refresh_from_db() self.assertEqual(row.status, "IN_PROGRESS") def test_history_readonly_and_deletion_disabled(self): row = prepared(self, filled=True) pass_qc(self, row) self.inspector.is_superuser = True # Native permissions; real QC posting retained. self.inspector.save() for model, obj in (("servicequalitycontrol", row), ("servicequalitycontrolcheck", queries.quality_control_checks(row).first()), ("servicequalitycontrolcomplaintcheck", queries.quality_control_complaint_checks(row).first())): url = reverse(f"admin:service_{model}_change", args=[obj.pk]) self.assertEqual(self.client.get(url).status_code, 200) self.assertEqual(self.client.post(url, dict(outcome="FAILED")).status_code, 403) self.assertEqual(self.client.get(reverse(f"admin:service_{model}_add")).status_code, 403) self.assertEqual(self.client.post(reverse(f"admin:service_{model}_delete", args=[obj.pk])).status_code, 403)