# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Test tables."""

from typing import Any, ClassVar, override

import lxml
from django.test import RequestFactory

from debusine.artifacts.models import TaskTypes
from debusine.db.context import context
from debusine.db.models import (
    Group,
    WorkRequest,
    Worker,
    WorkerPool,
    WorkflowTemplate,
)
from debusine.db.models.auth import GroupMembership
from debusine.db.playground import scenarios
from debusine.server.workflows.models import WorkRequestWorkflowData
from debusine.test.django import TestCase
from debusine.web.views.tables import (
    FilterFailedWorkRequests,
    GroupMembershipAdminTable,
    GroupMembershipTable,
    WorkRequestTable,
    WorkerPoolTable,
    WorkerTable,
    WorkflowPaginator,
    WorkflowTable,
)


class GroupMembershipTableTests(TestCase):
    """Tests for :py:class:`GroupMembershipTable`."""

    scenario = scenarios.DefaultContext(set_current=True)

    @override
    def setUp(self) -> None:
        super().setUp()
        self.request = RequestFactory().get("/")

    def assertUserTableRow(
        self, table: GroupMembershipTable
    ) -> lxml.objectify.ObjectifiedElement:
        tree = self.assertHTMLValid(
            self.render_string(
                "{% load debusine %}{% widget test %}",
                test=table.get_paginator(per_page=10),
                request=self.request,
                user=context.user,
            )
        )
        row = self.assertHasElement(tree, "*/table/tbody/tr")

        self.assertTextContentEqual(
            row.td[0], f"{self.scenario.user.username} (you)"
        )

        self.assertTextContentEqual(
            row.td[1], self.scenario.user.get_full_name()
        )

        return row

    def test_group_without_workspace(self) -> None:
        group = self.playground.create_group("Admins", [self.scenario.user])
        table = GroupMembershipTable(
            self.request, GroupMembership.objects.filter(group=group)
        )
        row = self.assertUserTableRow(table)
        role = row.td[2]
        self.assertFalse(role.xpath("a"))
        self.assertTextContentEqual(role, Group.Roles.MEMBER)
        self.assertEqual(len(row.td), 3)

    def test_admin_group_without_workspace(self) -> None:
        group = self.playground.create_group("Admins", [self.scenario.user])
        gm = group.set_user_role(self.scenario.user, Group.Roles.ADMIN)
        table = GroupMembershipAdminTable(
            self.request, GroupMembership.objects.filter(group=group)
        )
        row = self.assertUserTableRow(table)

        role = row.td[2]
        a = self.assertHasElement(role, "a")
        self.assertEqual(a.get("href"), gm.get_absolute_url_update())
        self.assertTextContentEqual(a, Group.Roles.ADMIN)

        actions = row.td[3]
        a = self.assertHasElement(actions, "a")
        self.assertEqual(a.get("href"), gm.get_absolute_url_remove())
        self.assertTextContentEqual(a, "Remove")

    def test_group_with_workspace(self) -> None:
        group = self.playground.create_group(
            "Admins", [self.scenario.user], workspace=self.scenario.workspace
        )
        table = GroupMembershipTable(
            self.request, GroupMembership.objects.filter(group=group)
        )
        row = self.assertUserTableRow(table)
        role = row.td[2]
        self.assertFalse(role.xpath("a"))
        self.assertTextContentEqual(role, Group.Roles.MEMBER)
        self.assertEqual(len(row.td), 3)

    def test_admin_group_with_workspace(self) -> None:
        group = self.playground.create_group(
            "Admins", [self.scenario.user], workspace=self.scenario.workspace
        )
        gm = group.set_user_role(self.scenario.user, Group.Roles.ADMIN)
        table = GroupMembershipAdminTable(
            self.request, GroupMembership.objects.filter(group=group)
        )
        row = self.assertUserTableRow(table)

        role = row.td[2]
        a = self.assertHasElement(role, "a")
        self.assertEqual(a.get("href"), gm.get_absolute_url_update())
        self.assertTextContentEqual(a, Group.Roles.ADMIN)

        actions = row.td[3]
        a = self.assertHasElement(actions, "a")
        self.assertEqual(a.get("href"), gm.get_absolute_url_remove())
        self.assertTextContentEqual(a, "Remove")


class WorkerTableTests(TestCase):
    """Tests for :py:class:`WorkerTable`."""

    def test_defaults(self) -> None:
        """Test default arguments for custom Pagination object."""
        request = RequestFactory().get("/")
        table = WorkerTable(request, Worker.objects.all())
        self.assertEqual(table.default_order, "name")
        self.assertEqual(
            [c.name for c in table.columns],
            ["type", "name", "pool", "last_seen", "status"],
        )
        self.assertEqual(table.template_name, "web/_worker-list.html")

    def test_worker_pool_options(self) -> None:
        """Test population of worker pool options."""
        pool = self.playground.create_worker_pool(name="testpool")
        self.playground.create_worker(worker_pool=pool)
        request = RequestFactory().get("/")
        table = WorkerTable(request, Worker.objects.all())
        f = table.filters["pool"]
        self.assertEqual(
            list(f.options.keys()),  # type: ignore[attr-defined]
            ["_none", str(pool.pk)],
        )


class WorkerPoolTableTests(TestCase):
    """Tests for :py:class:`WorkerPoolTable`."""

    def test_defaults(self) -> None:
        """Test default arguments for custom Pagination object."""
        request = RequestFactory().get("/")
        table = WorkerPoolTable(request, WorkerPool.objects.all())
        self.assertEqual(table.default_order, "name")
        self.assertEqual(
            [c.name for c in table.columns],
            ["name", "enabled", "instance_wide", "ephemeral", "workers"],
        )


class WorkRequestTableTests(TestCase):
    """Tests for :py:class:`WorkRequestTable`."""

    def test_defaults(self) -> None:
        """Test default arguments for custom Pagination object."""
        request = RequestFactory().get("/")
        table = WorkRequestTable(request, WorkRequest.objects.all())
        self.assertEqual(table.default_order, "-created_at")
        self.assertEqual(
            [c.name for c in table.columns],
            ["id", "created_at", "task_type", "task_name", "status", "result"],
        )


class FilterFailedWorkRequestsTests(TestCase):
    """Tests for :py:class:`FilterFailedWorkRequests`."""

    def test_filter_when_inactive(self) -> None:
        """Test calling filter_queryset with an inactive filter."""
        f = FilterFailedWorkRequests("Test")
        queryset = WorkRequest.objects.all()
        self.assertIs(f.filter_queryset(queryset, None), queryset)


class WorkflowTableTests(TestCase):
    """Tests for :py:class:`WorkflowTable`."""

    template: ClassVar[WorkflowTemplate]
    workflow_1: ClassVar[WorkRequest]
    workflow_2: ClassVar[WorkRequest]

    scenario = scenarios.DefaultContext(set_current=True)

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()

        cls.template = cls.playground.create_workflow_template(
            "name-2", "noop", workspace=cls.scenario.workspace
        )

        cls.workflow_1 = cls.playground.create_workflow(
            cls.template,
            task_data={},
        )
        cls.workflow_2 = cls.playground.create_workflow(
            cls.template,
            task_data={},
        )

    def _table(
        self, query_params: dict[str, Any] | None = None
    ) -> WorkflowTable:
        """Instantiate a table with the given query string parameters."""
        request = RequestFactory().get("/", data=query_params)
        return WorkflowTable(
            request,
            WorkRequest.objects.filter(
                task_type=TaskTypes.WORKFLOW, parent__isnull=True
            ),
        )

    def test_defaults(self) -> None:
        """Test default arguments for custom Pagination object."""
        table = self._table()
        self.assertEqual(table.default_order, "id.desc")
        self.assertEqual(
            [c.name for c in table.columns],
            [
                "id",
                "workflow_template",
                "status",
                "result",
                "started_at",
                "completed_at",
                "wr_count",
                "last_activity",
                "started_by",
            ],
        )

    def test_filter_form_with_issues(self) -> None:
        """Client submit filter form which is not valid."""
        table = self._table(
            {
                "filter-statuses": "does-not-exist",
                "filter-with_failed_work_requests": "1",
            }
        )
        f = table.filters["statuses"]
        self.assertIsNone(f.value)
        f = table.filters["with_failed_work_requests"]
        self.assertEqual(f.value, "1")

    def assert_filtered(self, workflow: WorkRequest, **kwargs: Any) -> None:
        """Make a request with the form and assert one workflow is listed."""
        table = self._table({f"filter-{k}": v for k, v in kwargs.items()})
        self.assertQuerySetEqual(table.rows, [workflow])

    def test_list_filtering_started_by(self) -> None:
        """View detail filters by started_by user."""
        user = self.playground.create_user(username="relevant")
        self.workflow_1.created_by = user
        self.workflow_1.save()
        self.assert_filtered(self.workflow_1, started_by=user.username)

    def test_list_filtering_status(self) -> None:
        """View detail filters by status."""
        self.workflow_1.status = WorkRequest.Statuses.COMPLETED
        self.workflow_1.save()
        self.assert_filtered(
            self.workflow_1, statuses=[WorkRequest.Statuses.COMPLETED]
        )

    def test_list_filtering_runtime_status(self) -> None:
        """View detail filters by runtime status."""
        self.workflow_1.status = WorkRequest.Statuses.RUNNING
        self.workflow_1.save()
        self.workflow_2.status = WorkRequest.Statuses.RUNNING
        self.workflow_2.save()
        self.workflow_1.create_child_worker("noop")
        self.assert_filtered(self.workflow_1, statuses=["running__blocked"])

    def test_list_filtering_runtime_any(self) -> None:
        """
        Test Workflow with running status is displayed.

        No need for any specific WorkRequest.workflow_running_status.
        """
        self.workflow_1.status = WorkRequest.Statuses.RUNNING
        self.workflow_1.save()
        self.workflow_1.create_child_worker("noop")
        self.assert_filtered(self.workflow_1, statuses=["running__any"])

    def test_list_filtering_result(self) -> None:
        """View detail filters by result."""
        self.workflow_1.result = WorkRequest.Results.SUCCESS
        self.workflow_1.save()
        self.assert_filtered(
            self.workflow_1, results=[WorkRequest.Results.SUCCESS]
        )

    def test_list_filtering_with_failed_work_requests(self) -> None:
        """View detail filters by with failed work requests."""
        table = self._table({"filter-with_failed_work_requests": "1"})
        self.assertQuerySetEqual(table.rows, [])

        wr = self.workflow_1.create_child_worker("noop")
        wr.result = WorkRequest.Results.FAILURE
        wr.save()
        self.assert_filtered(self.workflow_1, with_failed_work_requests="1")

        for _ in range(2):
            wr = self.workflow_2.create_child_worker("noop")
            wr.result = WorkRequest.Results.FAILURE
            wr.save()
        table = self._table({"filter-with_failed_work_requests": "1"})
        self.assertQuerySetEqual(
            table.rows, [self.workflow_1, self.workflow_2], ordered=False
        )

    def test_list_filtering_with_failed_work_requests_allow_failure(
        self,
    ) -> None:
        """Workflows with only allow_failure failures are excluded."""
        # workflow_1 has a failure with allow_failure=True: should be excluded
        wr_allow = self.workflow_1.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                allow_failure=True, display_name="test", step="test"
            ),
        )
        wr_allow.result = WorkRequest.Results.FAILURE
        wr_allow.save()

        # workflow_2 has a failure without allow_failure: should be included
        wr_no_allow = self.workflow_2.create_child_worker("noop")
        wr_no_allow.result = WorkRequest.Results.FAILURE
        wr_no_allow.save()

        self.assert_filtered(self.workflow_2, with_failed_work_requests="1")

    def test_paginator_allow_failure_skips_result_count(self) -> None:
        """Failed work requests with allow_failure are not counted."""
        # Child with allow_failure=True: should be excluded from result counts
        wr_allow = self.workflow_1.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                allow_failure=True, display_name="test", step="allow"
            ),
        )
        wr_allow.result = WorkRequest.Results.FAILURE
        wr_allow.save()

        # Child without allow_failure: should be counted as a failure
        wr_no_allow = self.workflow_1.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                display_name="test", step="no-allow"
            ),
        )
        wr_no_allow.result = WorkRequest.Results.FAILURE
        wr_no_allow.save()

        table = self._table()
        paginator = table.get_paginator(per_page=10)
        ctx = paginator.get_context_data()

        self.assertIsInstance(paginator, WorkflowPaginator)
        # Full equality test: only the non-allow_failure child is counted
        self.assertEqual(
            ctx["work_request_result_counts"],
            {self.workflow_1.pk: {WorkRequest.Results.FAILURE: 1}},
        )

    def test_list_filtering_workflow_templates(self) -> None:
        """View detail filters by workflow_templates."""
        other_template = self.playground.create_workflow_template(
            "other_template", "noop"
        )
        self.workflow_2.workflow_template = other_template
        self.workflow_2.save()
        self.assert_filtered(
            self.workflow_1, workflow_templates=self.template.name
        )

    def test_sorting_id(self) -> None:
        """Test output of workflows for "id" and "-id" sorting."""
        table = self._table({"order": "id.desc"})
        self.assertQuerySetEqual(table.rows, [self.workflow_2, self.workflow_1])
        table = self._table({"order": "id.asc"})
        self.assertQuerySetEqual(table.rows, [self.workflow_1, self.workflow_2])
