# Copyright 2022 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.

"""Tests for the work request views."""

import textwrap
from datetime import timedelta
from types import GeneratorType

from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.template.loader import get_template
from django.test import TestCase
from django.urls import reverse
from django.utils import timezone

from rest_framework import status

import yaml

from debusine.artifacts import LintianArtifact
from debusine.db.models import Artifact, WorkRequest, Workspace
from debusine.tasks import Lintian
from debusine.test import TestHelpersMixin
from debusine.web.tests.views.test_views import (
    _date_format,
    _html_work_request_result,
    _html_work_request_row,
    _html_work_request_status,
    _sort_table_handle,
)
from debusine.web.views.lintian import LintianView
from debusine.web.views.work_request import (
    WorkRequestDetailView,
    WorkRequestListView,
)


class WorkRequestDetailViewTests(TestHelpersMixin, TestCase):
    """Tests for WorkRequestDetailView class."""

    def setUp(self):
        """Set up common objects."""
        workspace = Workspace.objects.first()
        started_at = timezone.now()
        duration = 73
        task_data = {
            "build_components": ["any", "all"],
            "distribution": "stable",
        }
        completed_at = started_at + timedelta(seconds=duration)

        self.work_request = self.create_work_request(
            mark_running=True,
            workspace=workspace,
            started_at=timezone.now(),
            completed_at=completed_at,
            task_data=task_data,
            assign_new_worker=True,
            task_name="sbuild",
            result=WorkRequest.Results.SUCCESS,
        )

    def test_get_work_request(self):
        """View detail return work request information."""
        artifact, _ = self.create_artifact()
        artifact.created_by_work_request = self.work_request
        artifact.save()

        response = self.client.get(
            reverse("work-requests:detail", kwargs={"pk": self.work_request.id})
        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)

        # Check some fields
        self.assertContains(
            response, f"<li>Id: {self.work_request.id}</li>", html=True
        )
        self.assertContains(
            response,
            f"<li>Status: "
            f"{_html_work_request_status(self.work_request.status)}",
            html=True,
        )
        self.assertContains(
            response,
            "<li>Result: "
            f"{_html_work_request_result(self.work_request.result)}</li>",
            html=True,
        )
        self.assertContains(
            response,
            f"<li>Worker: {self.work_request.worker.name}</li>",
            html=True,
        )
        self.assertContains(
            response,
            f"<li>Workspace: {self.work_request.workspace.name}</li>",
            html=True,
        )
        self.assertContains(
            response,
            f"<li>Task name: {self.work_request.task_name}</li>",
            html=True,
        )

        for artifact in Artifact.objects.filter(
            created_by_work_request=self.work_request
        ):
            artifact_url = reverse(
                "artifacts:detail",
                kwargs={"artifact_id": artifact.id},
            )

            # Contains artifact's URL, download link and category
            artifact_download_url = artifact_url + "?archive=tar.gz"
            self.assertContains(
                response,
                f'<li>Id: <a href="{artifact_url}">{artifact.id}</a> '
                f'(<a href="{artifact_download_url}">download</a>) '
                f'Category: {artifact.category}</li>',
                html=True,
            )

        self.assertEqual(
            response.context["artifacts"].query.order_by, ("category",)
        )

        # Contains started_at, completed_at and duration
        created_at_fmt = _date_format(self.work_request.created_at)
        started_at_fmt = _date_format(self.work_request.started_at)
        self.assertContains(
            response, f"Created at: {created_at_fmt}", html=True
        )
        self.assertContains(
            response, f"Started at: {started_at_fmt}", html=True
        )
        self.assertContains(
            response,
            f"Duration: {round(self.work_request.duration)} seconds",
            html=True,
        )

        task_data_fmt = yaml.safe_dump(self.work_request.task_data)
        self.assertContains(
            response,
            f"<li>Task data:<pre><code>{task_data_fmt}</code></pre></li>",
            html=True,
        )

    def make_work_request_lintian(self, work_request: WorkRequest):
        """Change work_request to "lintian", create source artifact."""
        work_request.task_name = "lintian"
        source_artifact, _ = self.create_artifact(
            paths=["hello.dsc"], create_files=True
        )
        work_request.task_data = {
            "input": {
                "source_artifact_id": source_artifact.id,
                "binary_artifacts_ids": [],
            }
        }
        work_request.save()
        work_request.artifact_set.add(self.create_lintian_source_artifact())

    def test_template_work_request_detail(self):
        """Test that the template does not output "task_name view"."""
        rendered = get_template("web/work_request-detail.html").render(
            {"work_request": self.work_request}
        )
        self.assertNotIn(f"{self.work_request.task_name} view", rendered)

    def test_use_specific_plugin(self):
        """Test usage of a plugin instead of generic view."""
        self.make_work_request_lintian(self.work_request)

        response = self.client.get(
            reverse("work-requests:detail", kwargs={"pk": self.work_request.id})
        )

        # WorkRequest_get_template_names() return the correct template name
        self.assertIn(LintianView.template_name, response.template_name)

        # The LintianView().get_context_data() is in the response
        self.assertDictContainsAll(
            response.context_data,
            LintianView().get_context_data(self.work_request),
        )

    def create_lintian_source_artifact(self) -> Artifact:
        """
        Create a Lintian source artifact result.

        Contains lintian.txt file.
        """
        artifact, _ = self.create_artifact(
            paths=[Lintian.CAPTURE_OUTPUT_FILENAME],
            create_files=True,
        )
        artifact.category = LintianArtifact._category
        artifact.created_by_work_request = self.work_request
        artifact.data = {
            "summary": {"package_filename": {"hello": "hello.dsc"}}
        }
        artifact.save()

        return artifact

    def test_do_not_use_available_plugin_use_default(self):
        """Request to detail with view=generic: do not use plugin."""
        self.make_work_request_lintian(self.work_request)

        path = reverse(
            "work-requests:detail", kwargs={"pk": self.work_request.id}
        )

        response = self.client.get(path + "?view=generic")

        self.assertIn(
            WorkRequestDetailView.default_template_name, response.template_name
        )

        with self.assertRaises(AssertionError):
            # The context_data does not contain the specific Lintian
            # work request plugin data
            self.assertDictContainsAll(
                response.context_data,
                LintianView().get_context_data(self.work_request),
            )

        specialized_view_path = path
        self.assertContains(
            response,
            f'<p><a href="{specialized_view_path}">'
            f'{self.work_request.task_name} view</a></p>',
            html=True,
        )

    def test_do_not_use_available_plugin_invalid_task_data(self):
        """If task data is invalid, do not use plugin."""
        self.make_work_request_lintian(self.work_request)
        self.work_request.task_data["input"]["binary_artifacts_ids"] = 1
        self.work_request.save()

        with self.assertRaises(ValidationError) as cm:
            self.work_request.full_clean()
        validation_error = str(cm.exception)

        path = reverse(
            "work-requests:detail", kwargs={"pk": self.work_request.id}
        )

        response = self.client.get(path)

        self.assertEqual(response.status_code, status.HTTP_200_OK)

        self.assertIn(
            WorkRequestDetailView.default_template_name, response.template_name
        )

        # The context_data does not contain the specific Lintian work
        # request plugin data
        self.assertNotIn("lintian_txt_path", response.context_data)

        # The page does not link to the specialized view; instead, it shows
        # the validation error.
        specialized_view_path = path
        self.assertNotContains(
            response,
            f'<p><a href="{specialized_view_path}">'
            f'{self.work_request.task_name} view</a></p>',
            html=True,
        )
        self.assertContains(
            response,
            f"Validation error: <pre><code>{validation_error}</code></pre>",
            html=True,
        )


class WorkRequestListViewTests(TestHelpersMixin, TestCase):
    """Tests for WorkRequestListView class."""

    only_public_work_requests_message = (
        "Not authenticated. Only work requests "
        "in public workspaces are listed."
    )

    def setUp(self):
        """Create workspaces used in the tests."""
        self.private_workspace = self.create_workspace(name="Private")
        self.public_workspace = self.create_workspace(
            name="Public", public=True
        )

    def test_get_no_work_request(self):
        """No work requests in the server: 'No work requests' in response."""
        response = self.client.get(reverse("work-requests:list"))

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertContains(response, "No work requests.", html=True)

    def test_get_work_requests_all_authenticated_request(self):
        """Two work requests: all information appear in the response."""
        public_work_request = self.create_work_request(
            result=WorkRequest.Results.SUCCESS, workspace=self.public_workspace
        )
        private_work_request = self.create_work_request(
            workspace=self.private_workspace
        )

        user = get_user_model().objects.create_user(
            username="testuser", password="testpassword", email="testemail"
        )
        self.client.force_login(user)

        response = self.client.get(reverse("work-requests:list"))
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        # For all the workspaces does not display "Workspace: "
        self.assertNotContains(response, "Workspace: ")

        self.assertQuerysetEqual(
            response.context_data["object_list"],
            WorkRequest.objects.all().order_by("-created_at"),
        )

        self.assertContains(
            response, _html_work_request_row(public_work_request), html=True
        )
        self.assertContains(
            response,
            _html_work_request_row(private_work_request),
            html=True,
        )

        self.assertNotContains(response, self.only_public_work_requests_message)

        # the view have the handle to sort the table
        self.assertContains(response, _sort_table_handle, html=True)

    def test_get_work_requests_public_not_authenticated(self):
        """One work request: public one only."""
        public_work_request = self.create_work_request(
            result=WorkRequest.Results.SUCCESS, workspace=self.public_workspace
        )
        private_work_request = self.create_work_request(
            workspace=self.private_workspace
        )

        response = self.client.get(reverse("work-requests:list"))

        self.assertContains(
            response, _html_work_request_row(public_work_request), html=True
        )
        self.assertNotContains(
            response,
            _html_work_request_row(private_work_request),
            html=True,
        )

        self.assertContains(response, self.only_public_work_requests_message)

    def test_get_work_requests_filter_by_workspace(self):
        """Work request is created and filtered: not the requested workspace."""
        self.create_work_request()

        workspace_name = "Test"
        response = self.client.get(
            reverse("work-requests:list") + f"?workspace={workspace_name}"
        )
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        self.assertQuerysetEqual(
            response.context_data["object_list"],
            WorkRequest.objects.filter(workspace__name=workspace_name),
        )

        self.assertContains(
            response, f"<p>Workspace: {workspace_name}</p>", html=True
        )

    def test_pagination(self):
        """Pagination is set up and rendered by the template."""
        self.create_work_request(workspace=self.public_workspace)

        self.assertGreaterEqual(WorkRequestListView.paginate_by, 10)
        response = self.client.get(reverse("work-requests:list"))
        self.assertContains(response, '<nav aria-label="pagination">')
        self.assertIsInstance(
            response.context["elided_page_range"], GeneratorType
        )

    def test_sorting(self):
        """Test sorting."""
        for field in ["id", "created_at", "status", "result"]:
            for asc in ["0", "1"]:
                response = self.client.get(
                    reverse("work-requests:list") + f"?order={field}&asc={asc}"
                )

            self.assertEqual(response.context["order"], field)
            self.assertEqual(response.context["asc"], asc)

    def test_sorting_invalid_field(self):
        """Test sorting by a non-valid field: sorted by id."""
        response = self.client.get(
            reverse("work-requests:list") + "?order=something"
        )

        self.assertEqual(response.context["order"], "created_at")
        self.assertEqual(response.context["asc"], "0")

    def test_sorting_field_is_valid(self):
        """Test sorting with asc=0: valid order and asc."""
        response = self.client.get(
            reverse("work-requests:list") + "?order=id&asc=0"
        )
        self.assertEqual(response.context["order"], "id")
        self.assertEqual(response.context["asc"], "0")


class WorkRequestCreateViewTests(TestCase):
    """Tests for WorkRequestCreateView."""

    @classmethod
    def setUpTestData(cls):
        """Initialize class data."""
        cls.user = get_user_model().objects.create_user(
            username="testuser", password="testpass"
        )

    def test_create_work_request_permission_denied(self):
        """A non-authenticated request cannot get the form (or post)."""
        for method in [self.client.get, self.client.post]:
            with self.subTest(method):
                response = method(reverse("work_requests:create"))
                self.assertContains(
                    response,
                    "You need to be authenticated to create a Work Request",
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_create_work_request(self):
        """Post to "work_requests:create" to create a work request."""
        self.client.force_login(self.user)
        workspace = Workspace.objects.first()
        name = "sbuild"
        task_data_yaml = textwrap.dedent(
            """
        build_components:
        - any
        - all
        backend: schroot
        distribution: stable
        host_architecture: amd64
        input:
          source_artifact_id: 5
        """  # noqa: E501
        )

        self.assertEqual(WorkRequest.objects.count(), 0)

        response = self.client.post(
            reverse("work_requests:create"),
            {
                "workspace": workspace.id,
                "task_name": name,
                "task_data": task_data_yaml,
            },
        )

        # The work request got created
        work_request = WorkRequest.objects.first()
        self.assertIsNotNone(work_request.id)

        # and has the user assigned
        self.assertEqual(work_request.created_by, self.user)

        # the browser got redirected to the work_requests:detail
        self.assertRedirects(
            response,
            reverse("work_requests:detail", kwargs={"pk": work_request.id}),
        )
