import ast
from html.parser import HTMLParser
import re
import time
from pathlib import Path

from django.contrib.auth import get_user_model
from django.core.management import call_command
from django.test import TestCase

from requests_portal.models import ActivityLog, SupportRequest


PROJECT_ROOT = Path(__file__).resolve().parents[1]


class StatusCountParser(HTMLParser):
    def __init__(self, target):
        super().__init__()
        self.target = target
        self.depth = 0
        self.parts = []

    def handle_starttag(self, tag, attrs):
        if self.depth:
            self.depth += 1
        elif dict(attrs).get("data-status-count") == self.target:
            self.depth = 1

    def handle_endtag(self, tag):
        if self.depth:
            self.depth -= 1

    def handle_data(self, data):
        if self.depth:
            self.parts.append(data)


class PortalAcceptanceTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        User = get_user_model()
        cls.alice = User.objects.create_user(username="alice-test", password="test-pass-123")
        cls.bob = User.objects.create_user(username="bob-test", password="test-pass-123")
        cls.staff = User.objects.create_user(
            username="staff-test", password="test-pass-123", is_staff=True
        )
        cls.alice_request = SupportRequest.objects.create(
            title="Alice database error",
            description="The customer database returns an error.",
            created_by=cls.alice,
            priority="high",
        )
        cls.bob_request = SupportRequest.objects.create(
            title="Bob export request",
            description="Please add a CSV export.",
            created_by=cls.bob,
            status="closed",
        )
        ActivityLog.objects.create(
            support_request=cls.alice_request,
            actor=cls.alice,
            action="Created for acceptance test",
        )

    def login(self, user):
        self.client.force_login(user)

    def test_required_model_defaults(self):
        request = SupportRequest.objects.create(
            title="Defaults", description="Check defaults", created_by=self.alice
        )
        self.assertEqual(request.status, "open", "New requests must default to open")
        self.assertEqual(request.priority, "normal", "New requests must default to normal priority")
        self.assertIsNone(request.assignee, "New requests must be unassigned")

    def test_required_model_field_contract(self):
        fields = {field.name: field for field in SupportRequest._meta.get_fields()}
        self.assertEqual(fields["title"].max_length, 200, "Title maximum length must be 200")
        self.assertFalse(fields["title"].blank, "Title must be required")
        self.assertFalse(fields["description"].blank, "Description must be required")
        self.assertFalse(fields["created_by"].null, "Creator must be required")
        self.assertTrue(fields["assignee"].null, "Assignee must be optional")
        self.assertEqual(
            {value for value, _label in fields["status"].choices},
            {"open", "in_progress", "closed"},
            "Status choices must match the contract",
        )
        self.assertEqual(
            {value for value, _label in fields["priority"].choices},
            {"low", "normal", "high"},
            "Priority choices must match the contract",
        )
        request = SupportRequest.objects.create(
            title="Timestamp check", description="Timestamp behavior", created_by=self.alice
        )
        self.assertIsNotNone(request.created_at, "Creation must populate created_at")
        self.assertIsNotNone(request.updated_at, "Creation must populate updated_at")
        original_created = request.created_at
        original_updated = request.updated_at
        time.sleep(0.002)
        request.title = "Timestamp changed"
        request.save()
        request.refresh_from_db()
        self.assertEqual(request.created_at, original_created, "created_at must remain stable")
        self.assertGreater(request.updated_at, original_updated, "Saving a change must advance updated_at")

    def test_login_and_logout_flow(self):
        self.assertEqual(self.client.get("/login/").status_code, 200, "Login page must render")
        response = self.client.post(
            "/login/", {"username": "alice-test", "password": "test-pass-123"}
        )
        self.assertEqual(response.status_code, 302, "Valid login must redirect")
        self.assertEqual(self.client.get("/requests/").status_code, 200, "Login must create a session")
        response = self.client.post("/logout/")
        self.assertIn(response.status_code, (200, 302), "POST logout must succeed")
        self.assertEqual(self.client.get("/requests/").status_code, 302, "Logout must end the session")

    def test_every_portal_page_protects_anonymous_users(self):
        paths = (
            "/",
            "/requests/",
            "/requests/new/",
            f"/requests/{self.alice_request.pk}/",
            f"/requests/{self.alice_request.pk}/edit/",
            f"/requests/{self.alice_request.pk}/manage/",
        )
        for path in paths:
            response = self.client.get(path)
            self.assertEqual(response.status_code, 302, f"Anonymous request to {path} must redirect")
            self.assertIn("/login/", response.url, f"Anonymous request to {path} must go to login")

    def test_normal_user_list_excludes_other_users_requests(self):
        self.login(self.alice)
        response = self.client.get("/requests/")
        self.assertContains(response, self.alice_request.title)
        self.assertNotContains(response, self.bob_request.title)

    def test_staff_list_includes_all_requests(self):
        self.login(self.staff)
        response = self.client.get("/requests/")
        self.assertContains(response, self.alice_request.title)
        self.assertContains(response, self.bob_request.title)

    def test_normal_user_cannot_view_or_edit_another_users_request(self):
        self.login(self.alice)
        detail = self.client.get(f"/requests/{self.bob_request.pk}/")
        edit_get = self.client.get(f"/requests/{self.bob_request.pk}/edit/")
        edit_post = self.client.post(
            f"/requests/{self.bob_request.pk}/edit/",
            {"title": "Stolen", "description": "Stolen", "priority": "high"},
        )
        for response in (detail, edit_get, edit_post):
            self.assertIn(response.status_code, (403, 404), "Other users' records must not be exposed")
        self.bob_request.refresh_from_db()
        self.assertNotEqual(self.bob_request.title, "Stolen", "Forbidden edit must not persist")

    def test_normal_user_manage_get_and_post_return_403(self):
        self.login(self.alice)
        path = f"/requests/{self.alice_request.pk}/manage/"
        self.assertEqual(self.client.get(path).status_code, 403, "Manage GET must be forbidden")
        response = self.client.post(path, {"assignee": self.staff.pk, "status": "closed"})
        self.assertEqual(response.status_code, 403, "Manage POST must be forbidden")
        self.alice_request.refresh_from_db()
        self.assertEqual(self.alice_request.status, "open", "Forbidden management must not persist")

    def test_create_ignores_privileged_fields_and_logs_activity(self):
        self.login(self.alice)
        response = self.client.post(
            "/requests/new/",
            {
                "title": "Printer offline",
                "description": "Third floor printer",
                "priority": "low",
                "status": "closed",
                "assignee": self.staff.pk,
                "created_by": self.bob.pk,
            },
        )
        self.assertEqual(response.status_code, 302, "Valid creation should redirect")
        created = SupportRequest.objects.get(title="Printer offline")
        self.assertEqual(created.created_by, self.alice, "Creator must come from the session")
        self.assertEqual(created.status, "open", "Normal users must not set status during creation")
        self.assertIsNone(created.assignee, "Normal users must not assign during creation")
        self.assertTrue(
            ActivityLog.objects.filter(support_request=created, actor=self.alice).exists(),
            "Creation must write an activity record",
        )

    def test_edit_ignores_privileged_fields_and_logs_activity(self):
        self.login(self.alice)
        before = ActivityLog.objects.filter(support_request=self.alice_request).count()
        response = self.client.post(
            f"/requests/{self.alice_request.pk}/edit/",
            {
                "title": "Updated title",
                "description": "Updated description",
                "priority": "normal",
                "status": "closed",
                "assignee": self.staff.pk,
                "created_by": self.bob.pk,
            },
        )
        self.assertEqual(response.status_code, 302, "Valid edit should redirect")
        self.alice_request.refresh_from_db()
        self.assertEqual(self.alice_request.title, "Updated title", "Allowed edit must persist")
        self.assertEqual(self.alice_request.status, "open", "Normal edit must not change status")
        self.assertIsNone(self.alice_request.assignee, "Normal edit must not assign")
        self.assertEqual(self.alice_request.created_by, self.alice, "Normal edit must not change owner")
        self.assertGreater(
            ActivityLog.objects.filter(support_request=self.alice_request).count(),
            before,
            "Edit must create activity",
        )

    def test_staff_can_assign_and_close_request_with_activity(self):
        self.login(self.staff)
        before = ActivityLog.objects.filter(support_request=self.alice_request).count()
        response = self.client.post(
            f"/requests/{self.alice_request.pk}/manage/",
            {"assignee": self.staff.pk, "status": "closed"},
        )
        self.assertEqual(response.status_code, 302, "Staff management should redirect after success")
        self.alice_request.refresh_from_db()
        self.assertEqual(self.alice_request.assignee, self.staff, "Staff must be able to assign")
        self.assertEqual(self.alice_request.status, "closed", "Staff must be able to close")
        self.assertGreater(
            ActivityLog.objects.filter(support_request=self.alice_request).count(),
            before,
            "Management changes must create activity",
        )

    def test_assignment_only_and_status_only_are_audited_to_staff_actor(self):
        self.login(self.staff)
        path = f"/requests/{self.alice_request.pk}/manage/"
        before = ActivityLog.objects.filter(support_request=self.alice_request).count()
        response = self.client.post(path, {"assignee": self.staff.pk, "status": "open"})
        self.assertEqual(response.status_code, 302, "Assignment-only management must succeed")
        assignment_logs = ActivityLog.objects.filter(support_request=self.alice_request).order_by("pk")
        self.assertGreater(assignment_logs.count(), before, "Assignment-only change must be audited")
        self.assertEqual(assignment_logs.last().actor, self.staff, "Staff actor must be recorded")

        before = assignment_logs.count()
        response = self.client.post(path, {"assignee": self.staff.pk, "status": "in_progress"})
        self.assertEqual(response.status_code, 302, "Status-only management must succeed")
        status_logs = ActivityLog.objects.filter(support_request=self.alice_request).order_by("pk")
        self.assertGreater(status_logs.count(), before, "Status-only change must be audited")
        self.assertEqual(status_logs.last().actor, self.staff, "Staff actor must be recorded")

    def test_detail_renders_activity_history_and_labels(self):
        self.login(self.alice)
        response = self.client.get(f"/requests/{self.alice_request.pk}/")
        self.assertContains(response, "Created for acceptance test")
        content = response.content.decode().lower()
        self.assertIn("open", content, "Detail must show the request status")
        self.assertIn("high", content, "Detail must show the request priority")

    def test_invalid_creation_shows_errors_without_writing(self):
        self.login(self.alice)
        before = SupportRequest.objects.count()
        response = self.client.post(
            "/requests/new/", {"title": "", "description": "", "priority": "normal"}
        )
        self.assertEqual(response.status_code, 200, "Invalid form must be redisplayed")
        self.assertEqual(SupportRequest.objects.count(), before, "Invalid form must not write")
        self.assertContains(response, "required", html=False)

    def test_search_matches_title_and_description(self):
        self.login(self.staff)
        title_response = self.client.get("/requests/", {"q": "database"})
        self.assertContains(title_response, self.alice_request.title)
        self.assertNotContains(title_response, self.bob_request.title)
        description_response = self.client.get("/requests/", {"q": "CSV"})
        self.assertContains(description_response, self.bob_request.title)

    def test_status_filter(self):
        self.login(self.staff)
        response = self.client.get("/requests/", {"status": "closed"})
        self.assertContains(response, self.bob_request.title)
        self.assertNotContains(response, self.alice_request.title)

    def test_sort_oldest_and_newest(self):
        self.login(self.staff)
        oldest = self.client.get("/requests/", {"sort": "oldest"}).content.decode()
        newest = self.client.get("/requests/", {"sort": "newest"}).content.decode()
        self.assertLess(
            oldest.index(self.alice_request.title),
            oldest.index(self.bob_request.title),
            "Oldest sort must show older record first",
        )
        self.assertLess(
            newest.index(self.bob_request.title),
            newest.index(self.alice_request.title),
            "Newest sort must show newer record first",
        )

    def test_list_paginates_twenty_and_preserves_query(self):
        extras = []
        for index in range(25):
            extras.append(
                SupportRequest.objects.create(
                    title=f"Database extra {index:02d}",
                    description="Pagination",
                    created_by=self.staff,
                )
            )
        self.login(self.staff)
        response = self.client.get(
            "/requests/", {"q": "database", "status": "open", "sort": "oldest"}
        )
        html = response.content.decode()
        visible = sum(item.title in html for item in [self.alice_request, *extras])
        self.assertEqual(visible, 20, "Exactly 20 matching request titles must appear on page one")
        self.assertIn("page=2", html, "Pagination must link to page two")
        self.assertIn("q=database", html, "Pagination must preserve search")
        self.assertIn("status=open", html, "Pagination must preserve status")
        self.assertIn("sort=oldest", html, "Pagination must preserve sort")

    def test_empty_state_is_useful(self):
        User = get_user_model()
        empty_user = User.objects.create_user(username="empty-test", password="test-pass-123")
        self.login(empty_user)
        response = self.client.get("/requests/")
        text = response.content.decode().lower()
        self.assertTrue(
            "no requests" in text or "no support requests" in text,
            "Empty list must explain that there are no requests",
        )

    def test_dashboard_counts_respect_visibility(self):
        self.login(self.alice)
        alice_html = self.client.get("/").content.decode().lower()
        self.assertEqual(self._dashboard_count(alice_html, "open"), 1, "Alice must see one open request")
        self.assertEqual(self._dashboard_count(alice_html, "closed"), 0, "Alice must see no closed requests")
        self.assertNotIn(self.bob_request.title.lower(), alice_html, "Alice dashboard must not leak Bob")
        self.login(self.staff)
        staff_html = self.client.get("/").content.decode().lower()
        self.assertEqual(self._dashboard_count(staff_html, "open"), 1, "Staff must see one open request")
        self.assertEqual(self._dashboard_count(staff_html, "closed"), 1, "Staff must see one closed request")

    def _dashboard_count(self, html, status):
        parser = StatusCountParser(status)
        parser.feed(html)
        text = " ".join(parser.parts).strip()
        match = re.search(r"\b(\d+)\b", text)
        self.assertIsNotNone(match, f"Dashboard must expose semantic {status} count")
        return int(match.group(1))

    def test_seed_command_is_fully_idempotent(self):
        before_requests = SupportRequest.objects.count()
        before_activities = ActivityLog.objects.count()
        call_command("seed_portal", verbosity=0)
        first_requests = SupportRequest.objects.count()
        first_activities = ActivityLog.objects.count()
        self.assertEqual(first_requests, before_requests + 100, "First seed must add 100 requests")
        self.assertGreaterEqual(
            first_activities,
            before_activities + 100,
            "Every seeded request must have activity",
        )
        call_command("seed_portal", verbosity=0)
        self.assertEqual(SupportRequest.objects.count(), first_requests, "Second seed must add no requests")
        self.assertEqual(ActivityLog.objects.count(), first_activities, "Second seed must add no activity")
        User = get_user_model()
        expected = {
            "admin": ("portal-admin-2026", True),
            "alice": ("portal-alice-2026", False),
            "bob": ("portal-bob-2026", False),
        }
        for username, (password, is_staff) in expected.items():
            user = User.objects.get(username=username)
            self.assertTrue(user.check_password(password), f"{username} password must match spec")
            self.assertEqual(user.is_staff, is_staff, f"{username} staff flag must match spec")
        seeded = SupportRequest.objects.exclude(pk__in=(self.alice_request.pk, self.bob_request.pk))
        self.assertEqual(seeded.values("status").distinct().count(), 3, "Seed must cover all statuses")
        self.assertEqual(seeded.values("priority").distinct().count(), 3, "Seed must cover all priorities")
        self.assertEqual(
            set(seeded.values_list("created_by__username", flat=True)),
            {"admin", "alice", "bob"},
            "Seed requests must be distributed across all fixed users",
        )
        without_activity = [
            request.pk
            for request in seeded
            if not ActivityLog.objects.filter(support_request=request).exists()
        ]
        self.assertFalse(without_activity, "Every seeded request must have related activity")

    def test_project_artifacts_and_test_count(self):
        requirements = (PROJECT_ROOT / "requirements.txt").read_text()
        requirement_lines = [
            line.strip()
            for line in requirements.splitlines()
            if line.strip() and not line.lstrip().startswith(("#", "-"))
        ]
        self.assertTrue(requirement_lines, "requirements.txt must contain runtime dependencies")
        self.assertTrue(all("==" in line for line in requirement_lines), "Runtime requirements must be pinned")
        pinned = {line.split("==", 1)[0].lower(): line.split("==", 1)[1] for line in requirement_lines}
        self.assertIn("django", pinned, "Django must be pinned")
        self.assertTrue(pinned["django"].startswith("5."), "Django must use major version 5")
        self.assertIn("gunicorn", pinned, "Gunicorn must be pinned")
        procfile = (PROJECT_ROOT / "Procfile").read_text().lower()
        self.assertIn("gunicorn", procfile, "Procfile must use Gunicorn")
        readme = (PROJECT_ROOT / "README.md").read_text().lower()
        for command in ("migrate", "seed_portal", "test", "runserver"):
            self.assertIn(command, readme, f"README must document {command}")
        for credential in (
            "admin",
            "portal-admin-2026",
            "alice",
            "portal-alice-2026",
            "bob",
            "portal-bob-2026",
        ):
            self.assertIn(credential, readme, f"README must document benchmark credential {credential}")
        migration_files = list((PROJECT_ROOT / "requests_portal" / "migrations").glob("[0-9]*.py"))
        self.assertTrue(migration_files, "Application migrations must be committed")

        test_count = 0
        for path in (PROJECT_ROOT / "requests_portal").rglob("*.py"):
            if "migrations" in path.parts:
                continue
            tree = ast.parse(path.read_text())
            test_count += sum(
                isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
                and node.name.startswith("test_")
                for node in ast.walk(tree)
            )
        self.assertGreaterEqual(test_count, 25, "Project must include at least 25 authored tests")

    def test_required_source_has_no_todo_placeholders(self):
        for root_name in ("portal", "requests_portal", "templates"):
            root = PROJECT_ROOT / root_name
            if not root.exists():
                continue
            for path in root.rglob("*"):
                if path.is_file() and path.suffix in (".py", ".html"):
                    self.assertNotIn(
                        "todo",
                        path.read_text(errors="ignore").lower(),
                        f"Required source still contains TODO: {path}",
                    )
