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

"""Tests for the collection relations views."""

import re
from typing import Any, ClassVar
from urllib.parse import urlencode

from django.urls import reverse
from rest_framework import status
from rest_framework.settings import api_settings

from debusine.artifacts.models import CollectionCategory
from debusine.client.models import (
    CollectionIdentifierRequest,
    CollectionIdentifierResponse,
    CollectionRelationSetResponse,
    CollectionRelationType,
)
from debusine.db.models import Collection, CollectionRelation, User, Workspace
from debusine.db.playground import scenarios
from debusine.server.views.tests.base import TestCase


class CollectionRelationListViewTests(TestCase):
    """Tests for CollectionRelationListView."""

    scenario = scenarios.DefaultContextAPI()

    sid: ClassVar[Collection]
    trixie: ClassVar[Collection]
    experimental: ClassVar[Collection]

    @classmethod
    def setUpTestData(cls) -> None:
        """Set up test data."""
        super().setUpTestData()
        cls.sid = cls.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        cls.trixie = cls.playground.create_collection(
            "trixie", CollectionCategory.SUITE
        )
        cls.experimental = cls.playground.create_collection(
            "experimental", CollectionCategory.SUITE
        )

    def list_url(self) -> str:
        return reverse(
            "api:collection-relation-list",
            kwargs={"workspace": self.scenario.workspace.name},
        )

    def assert_list(
        self, api_response: dict[str, Any], expected: list[dict[str, Any]]
    ) -> None:
        self.assertEqual(api_response["count"], len(expected))
        self.assertEqual(api_response["results"], expected)

    def test_list_relations(self) -> None:
        """The list API returns ordered relation records."""
        CollectionRelation.objects.create(
            source=self.sid,
            target=self.trixie,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=1,
            created_by_user=self.scenario.user,
        )
        CollectionRelation.objects.create(
            source=self.sid,
            target=self.experimental,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=2,
            created_by_user=self.scenario.user,
        )

        response = self.client.get(
            self.list_url(), headers={"Token": self.scenario.user_token.key}
        )
        data = self.assertAPIResponseOk(response)

        self.assert_list(
            data,
            [
                {
                    "source_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "sid",
                    },
                    "target_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "trixie",
                    },
                    "position": 1,
                    "type": CollectionRelation.Types.SUITE_REQUIRES,
                },
                {
                    "source_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "sid",
                    },
                    "target_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "experimental",
                    },
                    "position": 2,
                    "type": CollectionRelation.Types.SUITE_REQUIRES,
                },
            ],
        )

    def test_list_relations_target_filter(self) -> None:
        CollectionRelation.objects.create(
            source=self.sid,
            target=self.trixie,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=1,
            created_by_user=self.scenario.user,
        )
        CollectionRelation.objects.create(
            source=self.sid,
            target=self.experimental,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=2,
            created_by_user=self.scenario.user,
        )

        response = self.client.get(
            self.list_url() + "?target=trixie@debian:suite",
            headers={"Token": self.scenario.user_token.key},
        )
        data = self.assertAPIResponseOk(response)

        self.assert_list(
            data,
            [
                {
                    "source_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "sid",
                    },
                    "target_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "trixie",
                    },
                    "position": 1,
                    "type": CollectionRelation.Types.SUITE_REQUIRES,
                }
            ],
        )

    def test_list_relations_etag(self) -> None:
        response = self.client.get(
            self.list_url() + "?source=sid@debian:suite&type=suite-based-on",
            headers={"Token": self.scenario.user_token.key},
        )

        etag = response["ETag"]

        self.assertIsNotNone(etag)

        CollectionRelation.objects.create(
            source=self.sid,
            target=self.trixie,
            type=CollectionRelation.Types.SUITE_BASED_ON,
            created_by_user=self.scenario.user,
        )

        response = self.client.get(
            self.list_url() + "?source=sid@debian:suite&type=suite-based-on",
            headers={"Token": self.scenario.user_token.key},
        )

        self.assertNotEqual(response["ETag"], etag)

    def test_list_relations_source_type_no_pagination_yes_etag(self) -> None:
        page_size = api_settings.PAGE_SIZE
        assert page_size is not None

        for number in range(page_size + 1):
            suite = self.playground.create_collection(
                f"suite-{number}", CollectionCategory.SUITE
            )
            CollectionRelation.objects.create(
                source=self.trixie,
                target=suite,
                type=CollectionRelation.Types.SUITE_REQUIRES,
                position=number,
                created_by_user=self.scenario.user,
            )

        response = self.client.get(
            self.list_url() + "?source=trixie@debian:suite&type=suite-requires",
            headers={"Token": self.scenario.user_token.key},
        )

        self.assertIn("ETag", response.headers)

        data = self.assertAPIResponseOk(response)

        relations = page_size + 1

        self.assertEqual(len(data["results"]), relations)

    def test_list_relations_yes_pagination_no_etag(self) -> None:
        page_size = api_settings.PAGE_SIZE
        assert page_size is not None

        for number in range(page_size + 1):
            suite = self.playground.create_collection(
                f"suite-{number}", CollectionCategory.SUITE
            )
            CollectionRelation.objects.create(
                source=self.trixie,
                target=suite,
                type=CollectionRelation.Types.SUITE_REQUIRES,
                position=number,
                created_by_user=self.scenario.user,
            )

        response = self.client.get(
            self.list_url() + "?source=trixie@debian:suite",
            headers={"Token": self.scenario.user_token.key},
        )

        self.assertNotIn("ETag", response.headers)

        data = self.assertAPIResponseOk(response)

        self.assertEqual(len(data["results"]), page_size)
        self.assertEqual(data["count"], page_size + 1)

    def test_list_relations_source_filter(self) -> None:
        CollectionRelation.objects.create(
            source=self.sid,
            target=self.trixie,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=1,
            created_by_user=self.scenario.user,
        )
        CollectionRelation.objects.create(
            source=self.trixie,
            target=self.experimental,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=2,
            created_by_user=self.scenario.user,
        )

        response = self.client.get(
            self.list_url() + "?source=sid@debian:suite",
            headers={"Token": self.scenario.user_token.key},
        )
        data = self.assertAPIResponseOk(response)

        self.assert_list(
            data,
            [
                {
                    "source_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "sid",
                    },
                    "target_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "trixie",
                    },
                    "position": 1,
                    "type": CollectionRelation.Types.SUITE_REQUIRES,
                }
            ],
        )

    def test_list_relations_type_filter(self) -> None:
        CollectionRelation.objects.create(
            source=self.sid,
            target=self.trixie,
            type=CollectionRelation.Types.SUITE_BASED_ON,
            created_by_user=self.scenario.user,
        )
        CollectionRelation.objects.create(
            source=self.trixie,
            target=self.experimental,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=2,
            created_by_user=self.scenario.user,
        )

        response = self.client.get(
            self.list_url() + "?type=suite-based-on",
            headers={"Token": self.scenario.user_token.key},
        )
        data = self.assertAPIResponseOk(response)

        self.assert_list(
            data,
            [
                {
                    "source_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "sid",
                    },
                    "target_collection": {
                        "category": CollectionCategory.SUITE,
                        "name": "trixie",
                    },
                    "position": None,
                    "type": CollectionRelation.Types.SUITE_BASED_ON,
                }
            ],
        )

    def test_list_relations_invalid_type_filter(self) -> None:
        response = self.client.get(
            self.list_url() + "?type=something-invalid",
            headers={"Token": self.scenario.user_token.key},
        )
        self.assertResponseProblem(
            response,
            "Invalid relation type",
            detail_pattern="Unknown relation type: 'something-invalid'",
            status_code=status.HTTP_400_BAD_REQUEST,
        )

    def test_list_relation_invalid_collection_identifier(self) -> None:
        collection = "does-not/exist@debian:suite"

        response = self.client.get(
            self.list_url() + f"?source={collection}",
            headers={"Token": self.scenario.user_token.key},
        )

        self.assertResponseProblem(
            response,
            "Invalid collection identifier in query parameter",
            detail_pattern=(
                f"Collection identifier {collection!r} must not contain '/'"
            ),
            status_code=status.HTTP_400_BAD_REQUEST,
        )

    def test_list_relation_collection_filter_does_not_exist(self) -> None:
        collection = "does-not-exist@debian:suite"

        response = self.client.get(
            self.list_url() + f"?source={collection}",
            headers={"Token": self.scenario.user_token.key},
        )

        self.assertResponseProblem(
            response,
            "Collection not found",
            detail_pattern=(
                f"Collection '{collection}' does not "
                f"exist in workspace 'System'"
            ),
            status_code=status.HTTP_404_NOT_FOUND,
        )

    def test_list_relations_invalid_source_format_query_parameter(self) -> None:
        response = self.client.get(
            self.list_url() + "?source=sid",
            headers={"Token": self.scenario.user_token.key},
        )

        self.assertResponseProblem(
            response,
            "Invalid collection identifier in query parameter",
            detail_pattern="sid",
            status_code=status.HTTP_400_BAD_REQUEST,
        )

    def test_list_relations_invalid_source_category_query_parameter(
        self,
    ) -> None:
        response = self.client.get(
            self.list_url() + "?source=sid@invalid-category",
            headers={"Token": self.scenario.user_token.key},
        )

        self.assertResponseProblem(
            response,
            "Collection not found",
            detail_pattern=(
                "Collection 'sid@invalid-category' "
                "does not exist in workspace 'System'"
            ),
            status_code=status.HTTP_404_NOT_FOUND,
        )

    def test_unauthenticated(self) -> None:
        """Authentication is required."""
        response = self.client.get(self.list_url())

        self.assertResponseProblem(
            response,
            "Error",
            detail_pattern="Authentication credentials were not provided.",
            status_code=status.HTTP_403_FORBIDDEN,
        )


class CollectionRelationSetViewTests(TestCase):
    """Tests for CollectionRelationSetView."""

    scenario = scenarios.DefaultContextAPI()

    trixie: ClassVar[Collection]
    trixie_backports: ClassVar[Collection]
    personal: ClassVar[Collection]

    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.trixie = cls.playground.create_collection(
            "trixie", CollectionCategory.SUITE
        )
        cls.trixie_backports = cls.playground.create_collection(
            "trixie-backports", CollectionCategory.SUITE
        )
        cls.personal = cls.playground.create_collection(
            "personal", CollectionCategory.SUITE
        )

    def grant_user_can_create_relations(
        self, user: User, collection: Collection
    ) -> None:
        owners = self.playground.create_group_role(
            collection.workspace, Workspace.Roles.OWNER
        )
        self.playground.add_user(owners, user)

    def set_url(self, *, relation_type: CollectionRelation.Types) -> str:
        return reverse(
            "api:collection-relation-update",
            kwargs={
                "workspace": self.scenario.workspace.name,
                "collection": "personal@debian:suite",
                "relation_type": relation_type,
            },
        )

    def get_etag(self, *, relation_type: CollectionRelation.Types) -> str:
        url = reverse(
            "api:collection-relation-list",
            kwargs={"workspace": self.scenario.workspace.name},
        )

        url += f"?source=personal@debian:suite&type={relation_type}"

        response = self.client.get(
            url, headers={"Token": self.scenario.user_token.key}
        )

        return response.headers["ETag"]

    def test_set_relationships(self) -> None:
        user = self.scenario.user

        self.grant_user_can_create_relations(user, self.personal)

        targets = [
            CollectionIdentifierRequest(
                name="trixie", category=CollectionCategory.SUITE
            ),
            CollectionIdentifierRequest(
                name="trixie-backports", category=CollectionCategory.SUITE
            ),
        ]

        self.assertEqual(
            CollectionRelation.objects.filter(source=self.personal).count(), 0
        )

        etag = self.get_etag(
            relation_type=CollectionRelation.Types.SUITE_REQUIRES
        )

        response = self.get_debusine().collection_relation_set(
            self.scenario.workspace.name,
            source="personal@debian:suite",
            relation_type=CollectionRelationType.SUITE_REQUIRES,
            targets=targets,
            if_match=etag,
        )

        self.assertEqual(
            response,
            CollectionRelationSetResponse(
                targets=[
                    CollectionIdentifierResponse(
                        category=CollectionCategory.SUITE,
                        name="trixie",
                    ),
                    CollectionIdentifierResponse(
                        category=CollectionCategory.SUITE,
                        name="trixie-backports",
                    ),
                ]
            ),
        )

        trixie, trixie_backports = CollectionRelation.objects.filter(
            source=self.personal
        ).order_by("position")

        self.assertEqual(trixie.target, self.trixie)
        self.assertEqual(trixie_backports.target, self.trixie_backports)

    def test_set_relationships_failed_invalid_relation_targets(self) -> None:
        user = self.scenario.user

        self.grant_user_can_create_relations(user, self.personal)

        etag = self.get_etag(
            relation_type=CollectionRelation.Types.SUITE_DEFAULT_QA_RESULTS
        )

        response = self.client.put(
            self.set_url(
                relation_type=CollectionRelation.Types.SUITE_DEFAULT_QA_RESULTS
            ),
            headers={"Token": self.scenario.user_token.key, "If-Match": etag},
            data={
                "targets": [
                    {"name": "trixie", "category": CollectionCategory.SUITE},
                ]
            },
            content_type="application/json",
        )

        self.assertResponseProblem(
            response,
            title="Invalid relation targets",
            detail_pattern="db_collectionrelation_categories_correct",
        )

    def test_set_relationships_failed_no_if_match(self) -> None:
        user = self.scenario.user

        self.grant_user_can_create_relations(user, self.personal)

        response = self.client.put(
            self.set_url(relation_type=CollectionRelation.Types.SUITE_REQUIRES),
            headers={"Token": self.scenario.user_token.key},
            data={
                "targets": [
                    {"name": "trixie", "category": CollectionCategory.SUITE},
                ]
            },
            content_type="application/json",
        )

        self.assertResponseProblem(
            response,
            title='"If-Match" header is required',
            detail_pattern=(
                'The "If-Match" header is required for PUT. '
                'Fetch the current ETag from'
            ),
            status_code=status.HTTP_400_BAD_REQUEST,
        )

    def test_set_relationships_failed_invalid_if_match(self) -> None:
        user = self.scenario.user

        self.grant_user_can_create_relations(user, self.personal)

        etag = self.get_etag(
            relation_type=CollectionRelation.Types.SUITE_REQUIRES
        )

        CollectionRelation.objects.create(
            source=self.personal,
            target=self.trixie_backports,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=1,
            created_by_user=self.scenario.user,
        )

        response = self.client.put(
            self.set_url(relation_type=CollectionRelation.Types.SUITE_REQUIRES),
            headers={"Token": self.scenario.user_token.key, "If-Match": etag},
            data={
                "targets": [
                    {"name": "trixie", "category": CollectionCategory.SUITE},
                ]
            },
            content_type="application/json",
        )

        list_url = (
            reverse(
                "api:collection-relation-list",
                kwargs={"workspace": self.scenario.workspace.name},
            )
            + "?"
            + urlencode(
                {
                    "source": "personal@debian:suite",
                    "type": "suite-requires",
                }
            )
        )
        self.assertResponseProblem(
            response,
            "Collection relations changed",
            detail_pattern=re.escape(
                'The "If-Match" header does not match the '
                f'current ETag. Fetch the current ETag from '
                f'"{list_url}" and retry.'
            ),
            status_code=status.HTTP_412_PRECONDITION_FAILED,
        )
