"""Administration discovery contracts; existing configuration tests stay frozen."""
from datetime import time, timedelta
from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.db import connection
from django.test import TestCase, Client, override_settings
from django.test.utils import CaptureQueriesContext
from django.urls import reverse
from django.utils import timezone
from apps.frontdesk.tests import prepare
from apps.frontdesk import services as desk
from apps.operations.tests import grant
from apps.operations.test_ui import capture_page
from apps.sla import services as sla
from apps.communications import services as communications
from apps.configuration.readiness import inspect_installation
from apps.organization.models import ServiceCenter, Department
from apps.catalog.models import ProductModel
from apps.access.services import set_role_permissions


class AdministrationWorkspaceTests(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="administration-center")
        cls.role = grant(cls.manager, cls.company, ["frontdesk.manage_slots"], cls.center)
        cls.company_manager = get_user_model().objects.create_user(username="administration-company")
        grant(cls.company_manager, cls.company, ["sla.manage_slapolicy", "communications.manage_templates"])
        cls.policy = cls.make_policy(cls.company, "LOCAL")
        cls.foreign_policy = cls.make_policy(cls.other_company, "FOREIGN")
        cls.template = cls.make_template(cls.company, "LOCAL")
        cls.foreign_template = cls.make_template(cls.other_company, "FOREIGN")

    @classmethod
    def make_policy(cls, company, code, **extra):
        return sla.save_policy(actor=cls.admin, company=company, code=code, name=code,
            effective_from=cls.today, target_minutes=60, warning_minutes=15, **extra)

    @classmethod
    def make_template(cls, company, code, **extra):
        return communications.save_template(actor=cls.admin, company=company, code=code, name=code,
            channel="EMAIL", subject="Synthetic template", body="Synthetic body", **extra)

    def page(self, user=None, **params):
        self.client.force_login(user or self.manager)
        return self.client.get(reverse("configuration:index"), params)

    @staticmethod
    def counts(response):
        return {row["label"]: row["value"] for row in response.context["metrics"]}

    def test_anonymous_ordinary_staff_and_inactive_boundaries(self):
        self.assertEqual(self.client.get("/settings/").status_code, 302)
        self.assertEqual(self.page(self.user).status_code, 403)
        self.user.is_staff = True
        self.user.save()
        self.user.user_permissions.set(Permission.objects.filter(content_type__app_label="organization"))
        self.assertEqual(self.page(self.user).status_code, 403)
        self.manager.is_active = False
        self.manager.save()
        self.assertIn(self.page(self.manager).status_code, (302, 403))

    def test_scoped_sections_and_navigation_do_not_advertise_global_admin(self):
        response = self.page()
        self.assertContains(response, "Administration / Settings")
        self.assertContains(response, "Appointment slots and capacity")
        self.assertNotContains(response, 'href="/admin/')
        self.assertNotContains(response, "People & Access")
        self.assertNotContains(response, 'href="/settings/readiness/')
        self.assertEqual(set(self.counts(response)), {"Active dated appointment slots"})
        capture_page(response, "administration-scoped")

    def test_global_sections_reuse_registered_admin_and_real_taxonomy(self):
        response = self.page(self.admin)
        for route in ("accounts_user", "access_role", "access_userroleassignment",
                      "organization_userorganizationassignment", "organization_company",
                      "organization_region", "organization_servicecenter", "organization_department",
                      "catalog_productmodel", "catalog_productvariant", "catalog_deviceidentificationpolicy",
                      "service_catalog_complaintsymptom", "service_catalog_faultdiagnosis",
                      "service_catalog_rootcause", "service_catalog_repairaction", "parts_sparepart"):
            self.assertContains(response, reverse("admin:" + route + "_changelist"))
        self.assertContains(response, "ComplaintSymptom is the complaint dimension")
        self.assertContains(response, "reactivation does not reopen children")
        self.assertContains(response, "Existing Django Admin change history")
        capture_page(response, "administration-global")

    def test_active_global_metrics_have_explicit_state_definitions(self):
        inactive = get_user_model().objects.create_user(username="inactive-count", is_active=False)
        counts = self.counts(self.page(self.admin))
        self.assertEqual(counts["Active users"], get_user_model().objects.filter(is_active=True).count())
        for label, model in [("Active service centers", ServiceCenter), ("Active departments", Department),
                             ("Active product models", ProductModel)]:
            self.assertEqual(counts[label], model.objects.filter(is_active=True).count())
        self.assertFalse(inactive.is_active)

    def test_slot_count_is_center_scoped_and_includes_active_past_dates(self):
        with patch("apps.frontdesk.services.timezone.now", return_value=timezone.now()-timedelta(days=2)):
            for center in (self.center, self.center2, self.other_center):
                desk.create_slot(actor=self.admin, service_center=center, date=self.today-timedelta(days=1),
                                 start_time=time(9), end_time=time(10), capacity=5)
        inactive = desk.create_slot(actor=self.admin, service_center=self.center, date=self.today+timedelta(days=1),
                         start_time=time(9), end_time=time(10), capacity=5)
        desk.configure_slot(actor=self.admin, slot=inactive, capacity=5, is_active=False)
        self.assertEqual(self.counts(self.page())["Active dated appointment slots"], 2)

    def test_company_metrics_and_destinations_preserve_isolation(self):
        response = self.page(self.company_manager)
        self.assertEqual(self.counts(response), {"Active SLA policies": 1, "Active notification templates": 1})
        self.assertContains(response, reverse("sla:policies"))
        self.assertContains(response, reverse("communications:templates"))
        self.assertNotContains(response, self.foreign_policy.name)
        self.assertNotContains(response, 'href="/admin/')
        self.assertEqual(self.client.get(reverse("configuration:readiness")).status_code, 403)
        capture_page(response, "administration-company")

    def test_sla_and_template_destinations_enforce_scope_and_csrf(self):
        self.client.force_login(self.company_manager)
        for route, local, foreign, name in [
            ("sla:policy_edit", self.policy, self.foreign_policy, "sla-form"),
            ("communications:template-edit", self.template, self.foreign_template, "template-form"),
        ]:
            local_url = reverse(route, args=[local.pk])
            response = self.client.get(local_url)
            self.assertEqual(response.status_code, 200)
            capture_page(response, "administration-" + name)
            foreign_url = reverse(route, args=[foreign.pk])
            self.assertEqual(self.client.get(foreign_url).status_code, 404)
            self.assertEqual(self.client.post(foreign_url, {}).status_code, 404)
            csrf_client = Client(enforce_csrf_checks=True)
            csrf_client.force_login(self.company_manager)
            self.assertEqual(csrf_client.post(local_url, {}).status_code, 403)
        for route, name in [("sla:policies", "sla-policies"), ("communications:templates", "templates")]:
            response = self.client.get(reverse(route))
            self.assertNotContains(response, "FOREIGN")
            capture_page(response, "administration-" + name)

    def test_center_permissions_do_not_grant_company_management(self):
        set_role_permissions(role=self.role, permissions=Permission.objects.filter(
            codename__in=["manage_slapolicy", "manage_templates"]))
        self.assertEqual(self.page().status_code, 403)

    def test_nonstaff_superuser_has_no_advanced_admin_links_or_global_user_metrics(self):
        response = self.page(self.second)
        self.assertNotContains(response, 'href="/admin/')
        self.assertNotIn("Active users", self.counts(response))
        self.assertEqual(self.client.get(reverse("configuration:readiness")).status_code, 403)

    def test_scoped_inventory_admin_link_needs_both_business_and_native_authority(self):
        self.manager.is_staff = True
        self.manager.save(update_fields=["is_staff"])
        set_role_permissions(role=self.role, permissions=Permission.objects.filter(
            content_type__app_label="inventory", codename__in=["manage_inventory", "view_stock"]))
        self.assertEqual(self.page().status_code, 403)
        self.manager.user_permissions.add(Permission.objects.get(
            content_type__app_label="inventory", codename="change_inventorylocation"))
        response = self.page()
        self.assertContains(response, reverse("admin:inventory_inventorylocation_changelist"))
        self.assertNotContains(response, reverse("admin:access_role_changelist"))
        self.assertEqual(self.counts(response), {})
        capture_page(response, "administration-inventory-manager")

    def test_department_only_assignment_does_not_grant_center_configuration(self):
        from apps.organization.assignment_services import create_assignment
        from apps.access.services import create_role_assignment
        department = Department.objects.create(company=self.company, code="ONLY", name="Only department")
        assignment = create_assignment(user=self.user, company=self.company, department=department)
        create_role_assignment(user=self.user, role=self.role, organization_assignment=assignment)
        self.assertEqual(self.page(self.user).status_code, 403)
        self.assertEqual(self.client.get(reverse("configuration:slot_new", args=[self.center.pk])).status_code, 404)

    def test_search_only_filters_authorized_labels_and_escapes_input(self):
        response = self.page(self.admin, q="Root causes")
        links = [label for section in response.context["sections"] for label, _ in section["links"]]
        self.assertEqual(links, ["Root causes and applicability (optional)"])
        capture_page(response, "administration-search")
        response = self.page(q="Roles")
        self.assertEqual(response.context["sections"], [])
        capture_page(response, "administration-empty")
        response = self.page(q='<script>alert(1)</script>')
        self.assertNotContains(response, '<script>alert(1)</script>')
        self.assertContains(response, '&lt;script&gt;')

    def test_search_does_not_search_template_content(self):
        response = self.page(self.company_manager, q="Synthetic body")
        self.assertEqual(response.context["sections"], [])
        self.assertNotContains(response, "Synthetic template")

    def test_readiness_reuses_existing_inspector_and_does_not_send_or_disclose_secrets(self):
        self.client.force_login(self.admin)
        secret = "Synthetic-credential-never-render-6E"
        with override_settings(EMAIL_HOST_PASSWORD=secret, COMMUNICATION_PROVIDERS={"EMAIL": secret}), \
                patch("apps.communications.providers.get_provider") as provider, \
                patch("apps.configuration.readiness.inspect_installation", wraps=inspect_installation) as inspector:
            response = self.client.get(reverse("configuration:readiness"))
        inspector.assert_called_once_with()
        provider.assert_not_called()
        self.assertNotContains(response, secret)
        for level in ("REQUIRED", "RECOMMENDED", "OPTIONAL"):
            self.assertContains(response, level)
        self.assertContains(response, "not production certification")
        capture_page(response, "administration-readiness")

    def test_overview_is_read_only_and_cannot_escalate_privileges(self):
        self.client.force_login(self.manager)
        with CaptureQueriesContext(connection) as queries:
            response = self.client.get("/settings/")
        self.assertEqual(response.status_code, 200)
        self.assertFalse(any(q["sql"].lstrip().upper().startswith(("INSERT", "UPDATE", "DELETE")) for q in queries))
        self.assertEqual(self.client.post("/settings/", {"is_superuser": True}).status_code, 405)
        self.manager.refresh_from_db()
        self.assertFalse(self.manager.is_superuser)
        self.assertEqual(self.client.get(reverse("admin:accounts_user_changelist")).status_code, 302)

    def test_slot_forms_and_error_states_retain_service_validation_and_csrf(self):
        self.client.force_login(self.manager)
        for route, args, name in [("configuration:slots", [], "centers"),
            ("configuration:center_slots", [self.center.pk], "slots"),
            ("configuration:slot_new", [self.center.pk], "new-slot"),
            ("configuration:slot_edit", [self.center.pk, self.slot.pk], "edit-slot")]:
            response = self.client.get(reverse(route, args=args))
            self.assertContains(response, "Administration / Settings")
            capture_page(response, "administration-"+name)
        url = reverse("configuration:slot_edit", args=[self.center.pk, self.slot.pk])
        response = self.client.post(url, {"capacity": 0, "is_active": "on"})
        self.assertEqual(response.status_code, 400)
        capture_page(response, "administration-slot-error")
        csrf_client = Client(enforce_csrf_checks=True)
        csrf_client.force_login(self.manager)
        self.assertEqual(csrf_client.post(url, {"capacity": 3}).status_code, 403)
        self.slot.refresh_from_db()
        self.assertEqual(self.slot.capacity, 2)

    def test_sibling_and_foreign_slot_destinations_reject_get_and_post(self):
        self.client.force_login(self.manager)
        for center in (self.center2, self.other_center):
            url = reverse("configuration:slot_new", args=[center.pk])
            self.assertEqual(self.client.get(url).status_code, 404)
            self.assertEqual(self.client.post(url, {"capacity": 3}).status_code, 404)

    def test_overview_and_slot_list_query_growth_is_constant(self):
        self.client.force_login(self.manager)
        urls = ["/settings/", reverse("configuration:center_slots", args=[self.center.pk])]
        def measure():
            result = []
            for url in urls:
                with CaptureQueriesContext(connection) as captured:
                    response = self.client.get(url)
                self.assertEqual(response.status_code, 200)
                self.assertLessEqual(len(captured), 10)
                result.append(len(captured))
            return result
        small = measure()
        for number in range(1, 9):
            desk.create_slot(actor=self.admin, service_center=self.center, date=self.today+timedelta(days=number),
                             start_time=time(9), end_time=time(10), capacity=5)
        large = measure()
        self.assertEqual(small, large)
        print("Administration scoped overview/slot queries 1 -> 9:", small, large)

    def test_global_and_company_overview_queries_do_not_grow_with_configuration(self):
        def measure():
            counts = []
            for user in (self.admin, self.company_manager):
                self.client.force_login(user)
                with CaptureQueriesContext(connection) as captured:
                    response = self.client.get("/settings/")
                self.assertEqual(response.status_code, 200)
                self.assertEqual(len(captured), 13 if user == self.admin else 8)
                counts.append(len(captured))
            return counts
        small = measure()
        for number in range(1, 9):
            self.make_template(self.company, "GROW"+str(number))
            # Non-overlapping future policies are active records, not current coverage.
            sla.save_policy(actor=self.admin, company=self.company, code="GROW"+str(number), name="Future",
                effective_from=self.today+timedelta(days=number), effective_until=self.today+timedelta(days=number),
                service_center=self.center, target_minutes=60, warning_minutes=15)
            get_user_model().objects.create_user(username="growth"+str(number))
        large = measure()
        self.assertEqual(small, large)
        self.assertEqual(self.counts(self.page(self.company_manager)), {"Active SLA policies": 9, "Active notification templates": 9})
        print("Administration global/company overview queries 1 -> 9:", small, large)

    def test_review_all_secret_sources_stay_out_of_overview_and_readiness(self):
        import os
        secrets = {name: "Synthetic-6E1-" + name for name in
                   ("SMTP", "API_TOKEN", "APP_SECRET", "SECRET_KEY", "ENCRYPTION_KEY", "DATABASE")}
        with override_settings(SECRET_KEY=secrets["SECRET_KEY"], EMAIL_HOST_PASSWORD=secrets["SMTP"],
                COMMUNICATION_PROVIDERS={"EMAIL": secrets["API_TOKEN"]}), \
                patch.dict(os.environ, {key: secrets[key] for key in ("API_TOKEN", "APP_SECRET", "ENCRYPTION_KEY")}), \
                patch("apps.communications.providers.get_provider") as provider, \
                patch("apps.communications.services.get_provider") as service_provider:
            self.client.force_login(self.admin)
            # This TestCase already owns an open test-database transaction.
            # Substitute only its in-memory password setting, never reconnect.
            with patch.dict(connection.settings_dict, {"PASSWORD": secrets["DATABASE"]}), \
                    CaptureQueriesContext(connection) as captured:
                for route in ("configuration:index", "configuration:readiness"):
                    response = self.client.get(reverse(route))
                    self.assertEqual(response.status_code, 200)
                    for secret in secrets.values():
                        self.assertNotContains(response, secret)
            self.assertFalse(any(q["sql"].lstrip().upper().startswith(("INSERT", "UPDATE", "DELETE")) for q in captured))
        provider.assert_not_called()
        service_provider.assert_not_called()

    def test_review_ordinary_user_configuration_posts_are_denied_without_changes(self):
        self.client.force_login(self.user)
        for route, args in [("configuration:slot_edit", [self.center.pk, self.slot.pk]),
                            ("sla:policy_edit", [self.policy.pk]),
                            ("communications:template-edit", [self.template.pk])]:
            self.assertEqual(self.client.post(reverse(route, args=args),
                {"capacity": 99, "is_active": "on", "is_superuser": "on"}).status_code, 404)
        self.slot.refresh_from_db()
        self.policy.refresh_from_db()
        self.template.refresh_from_db()
        self.user.refresh_from_db()
        self.assertEqual(self.slot.capacity, 2)
        self.assertEqual(self.policy.target_minutes, 60)
        self.assertEqual(self.template.body, "Synthetic body")
        self.assertFalse(self.user.is_superuser)

    def test_review_inactive_policies_and_templates_do_not_inflate_active_metrics(self):
        self.make_policy(self.company, "INACTIVE", is_active=False)
        self.make_template(self.company, "INACTIVE", is_active=False)
        self.assertEqual(self.counts(self.page(self.company_manager)),
            {"Active SLA policies": 1, "Active notification templates": 1})