# 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 auth views."""
import re
from datetime import timedelta
from typing import ClassVar

from django.conf import settings
from django.contrib.auth import get_user_model
from django.template.response import SimpleTemplateResponse
from django.test import override_settings
from django.urls import reverse
from rest_framework import status

from debusine.db.models import Token, User
from debusine.db.models.auth import Identity
from debusine.db.playground import scenarios
from debusine.server.signon.providers import Provider
from debusine.test.django import TestCase
from debusine.web.views.tests.utils import (
    ViewTestMixin,
    date_format,
    html_check_icon,
)


class LoginViewTests(TestCase):
    """Tests for the LoginView class."""

    def test_get(self) -> None:
        """Test plain get, no next= argument."""
        response = self.client.get(reverse("login"))
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.client.session["sso_callback_next_url"], "/")

    def test_get_valid_next(self) -> None:
        """Test plain get, valid next= argument."""
        response = self.client.get(reverse("login") + "?next=/foo")
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.client.session["sso_callback_next_url"], "/foo")

    def test_get_invalid_next(self) -> None:
        """Test plain get, invalid next= argument."""
        response = self.client.get(
            reverse("login") + "?next=https://example.com"
        )
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.client.session["sso_callback_next_url"], "/")

    def test_successful_login(self) -> None:
        """Successful login redirect to the homepage with authenticated user."""
        username = "testuser"
        password = "testpassword"

        get_user_model().objects.create_user(
            username=username, password=password
        )

        response = self.client.post(
            reverse("login"),
            data={"username": username, "password": password},
        )

        self.assertRedirects(response, reverse("homepage:homepage"))
        self.assertTrue(response.wsgi_request.user.is_authenticated)

    def test_nexturl_in_session(self) -> None:
        """Login leaves the next url in the session."""
        response = self.client.get(
            reverse("login") + "?next=foo",
        )
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(self.client.session["sso_callback_next_url"], "foo")


class LogoutViewTests(TestCase):
    """Tests for the logout view."""

    def test_successful_render(self) -> None:
        """
        Logout template is rendered.

        The view is implemented (and tested) as part of Django. The template
        is implemented by Debusine, and it's rendered here to assert that
        no missing template variables and contains the expected strings.
        """
        response = self.client.post(reverse("logout"))

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


class UserDetailViewTests(ViewTestMixin, TestCase):
    """Test for UserDetailView class."""

    scenario = scenarios.DefaultScopeUser()

    def test_anonymous(self) -> None:
        """Test accessing as anonymous user."""
        response = self.client.get(self.scenario.user.get_absolute_url())
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertEqual(
            response.context["exception"], "No User matches the given query."
        )

    def test_detail(self) -> None:
        """Test viewing the default user."""
        self.client.force_login(self.scenario.user)
        response = self.client.get(self.scenario.user.get_absolute_url())
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        tree = self.assertHTMLValid(response)

        dl = self.assertHasElement(tree, "//dl[@id='person-info']")
        self.assertTextContentEqual(
            dl.dd[0], self.scenario.user.get_full_name()
        )
        self.assertTextContentEqual(dl.dd[1], self.scenario.user.username)
        self.assertTextContentEqual(
            dl.dd[2], str(self.scenario.user.token_set.count())
        )

        self.assertFalse(tree.xpath("//table[@id='person-groups']"))
        self.assertFalse(tree.xpath("//table[@id='person-identities']"))

    def test_detail_groups(self) -> None:
        """Test viewing a user with groups."""
        self.client.force_login(self.scenario.user)
        debusine_groups = ("foo", "bar", "baz")
        another_groups = ("wobble", "wibble", "wabble")

        for name in debusine_groups:
            self.playground.create_group(
                name=name, scope=self.scenario.scope, users=[self.scenario.user]
            )

        scope = self.playground.get_or_create_scope("another")
        for name in another_groups:
            self.playground.create_group(
                name=name, scope=scope, users=[self.scenario.user]
            )

        response = self.client.get(self.scenario.user.get_absolute_url())
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        tree = self.assertHTMLValid(response)

        table = self.assertHasElement(tree, "//table[@id='person-groups']")
        tbody = self.assertHasElement(table, "tbody")
        self.assertTextContentEqual(tbody.tr[0].td[0], "another")
        for li, name in zip(tbody.tr[0].td[1].ul.li, sorted(another_groups)):
            self.assertTextContentEqual(li, name)
        self.assertTextContentEqual(tbody.tr[1].td[0], "debusine")
        for li, name in zip(tbody.tr[1].td[1].ul.li, sorted(debusine_groups)):
            self.assertTextContentEqual(li, name)

    @override_settings(SIGNON_PROVIDERS=[Provider(name="salsa", label="Salsa")])
    def test_detail_identities(self) -> None:
        """Test viewing a user with SSO identities."""
        self.client.force_login(self.scenario.user)

        identity = Identity.objects.create(
            user=self.scenario.user,
            issuer="salsa",
            subject="playground@debian.org",
        )

        response = self.client.get(self.scenario.user.get_absolute_url())
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        tree = self.assertHTMLValid(response)

        table = self.assertHasElement(tree, "//table[@id='person-identities']")
        tr = self.assertHasElement(table, "tbody/tr")
        self.assertTextContentEqual(tr.td[0], "Salsa")
        self.assertTextContentEqual(tr.td[1], "playground@debian.org")
        self.assertTextContentEqual(
            tr.td[2], identity.last_used.strftime("%Y-%m-%d %H:%M")
        )

    @override_settings(SIGNON_PROVIDERS=[Provider(name="salsa", label="Salsa")])
    def test_detail_identities_cannot_manage(self) -> None:
        """Test viewing another user with SSO identities."""
        self.client.force_login(self.scenario.user)

        user = self.playground.create_user("test")
        Identity.objects.create(
            user=user,
            issuer="salsa",
            subject="playground@debian.org",
        )

        response = self.client.get(user.get_absolute_url())
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        tree = self.assertHTMLValid(response)
        self.assertFalse(tree.xpath("//table[@id='person-identities']"))

    def test_detail_identities_without_provider(self) -> None:
        """Test showing a user with identities without a SSO provider."""
        self.client.force_login(self.scenario.user)

        Identity.objects.create(
            user=self.scenario.user,
            issuer="salsa",
            subject="playground@debian.org",
        )
        response = self.client.get(self.scenario.user.get_absolute_url())
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        tree = self.assertHTMLValid(response)
        self.assertFalse(tree.xpath("//table[@id='person-identities']"))

    @override_settings(SIGNON_PROVIDERS=[Provider(name="salsa", label="Salsa")])
    def test_detail_identities_with_wrong_provider(self) -> None:
        """Test showing a user with identities of nonexisting SSO provider."""
        self.client.force_login(self.scenario.user)

        Identity.objects.create(
            user=self.scenario.user,
            issuer="nachos",
            subject="playground@debian.org",
        )
        response = self.client.get(self.scenario.user.get_absolute_url())
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        tree = self.assertHTMLValid(response)
        self.assertFalse(tree.xpath("//table[@id='person-identities']"))


class UserTokenListViewTests(ViewTestMixin, TestCase):
    """Tests for UserTokenListView class."""

    scenario = scenarios.DefaultScopeUser()

    def test_only_one_scope(self) -> None:
        """Test hiding the scopes list if there is only one."""
        self.client.force_login(self.scenario.user)
        response = self.client.get(
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            )
        )
        tree = self.assertHTMLValid(response)
        sample_cfg = self.assertHasElement(tree, "//pre[@id='example-config']")
        fqdn = settings.DEBUSINE_FQDN
        self.assertTextContentEqual(
            sample_cfg,
            f"""
    [General]
    default-server = {fqdn}

    [server:{fqdn}]
    api-url = https://{fqdn}/api
    scope = {self.scenario.scope}
    token = generated-token-goes-here
    """,
        )
        self.assertFalse(tree.xpath("ul[@id='scope-list']"))

    def test_two_scopes(self) -> None:
        """Test when there is more than one scope."""
        self.playground.get_or_create_scope("test")
        self.client.force_login(self.scenario.user)
        response = self.client.get(
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            )
        )
        tree = self.assertHTMLValid(response)
        sample_cfg = self.assertHasElement(tree, "//pre[@id='example-config']")
        fqdn = settings.DEBUSINE_FQDN
        self.assertTextContentEqual(
            sample_cfg,
            f"""
    [General]
    default-server = {fqdn}

    [server:{fqdn}]
    api-url = https://{fqdn}/api
    scope = one-of-the-scopes-listed-below
    token = generated-token-goes-here
    """,
        )
        ul = self.assertHasElement(tree, "//ul[@id='scope-list']")
        self.assertTextContentEqual(ul.li[0], "debusine")
        self.assertTextContentEqual(ul.li[1], "test")

    def test_logged_no_tokens(self) -> None:
        """Logged in user with no tokens."""
        another_user = get_user_model().objects.create_user(
            username="notme", password="testpassword", email="test2@example.com"
        )
        another_token = Token.objects.create(user=another_user)

        self.client.force_login(self.scenario.user)
        response = self.client.get(
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            )
        )
        tree = self.assertHTMLValid(response)
        self.assertNavHasUser(tree, self.scenario.user)

        self.assertContains(
            response,
            "You do not currently have any tokens",
        )

        create_token_url = reverse(
            "user:token-create",
            kwargs={"username": self.scenario.user.username},
        )
        self.assertContains(
            response,
            f'<a href="{create_token_url}">create a new token</a>',
            html=True,
        )

        # Ensure that a token for another user is not displayed
        self.assertNotContains(response, another_token.key)

    def row_for_token(self, token: Token) -> str:
        """Return HTML for the row of the table for token."""
        edit_url = reverse(
            "user:token-edit",
            kwargs={"pk": token.pk, "username": self.scenario.user.username},
        )
        delete_url = reverse(
            "user:token-delete",
            kwargs={"pk": token.pk, "username": self.scenario.user.username},
        )

        return (
            "<tr>"
            f"<td>{html_check_icon(token.enabled)}</td>"
            f"<td>{date_format(token.created_at)}</td>"
            f"<td>{token.comment}</td>"
            f'<td><a href="{delete_url}">Delete</a>&nbsp;|&nbsp;'
            f'<a href="{edit_url}">Edit</a></td>'
            "</tr>"
        )

    def test_logged_two_tokens(self) -> None:
        """
        Logged user with two tokens: both are displayed.

        Assert tokens are ordered by created_at.
        """
        token1 = Token.objects.create(
            user=self.scenario.user, comment="Token 1"
        )
        token2 = Token.objects.create(
            user=self.scenario.user, comment="Token 2"
        )

        self.client.force_login(self.scenario.user)
        response = self.client.get(
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            )
        )

        self.assertContains(response, self.row_for_token(token1), html=True)
        self.assertContains(response, self.row_for_token(token2), html=True)

        # Check tokens are ordered_by created_

        # token1 is displayed earlier than token2 (was created before)
        self.assertLess(
            response.content.index(token1.comment.encode("utf-8")),
            response.content.index(token2.comment.encode("utf-8")),
        )

        # Make token1 to be created_at after token2 created_at
        token1.created_at = token2.created_at + timedelta(minutes=1)
        token1.save()

        response = self.client.get(
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            )
        )

        # Tokens appear in the opposite order than before
        self.assertGreater(
            response.content.index(token1.comment.encode("utf-8")),
            response.content.index(token2.comment.encode("utf-8")),
        )

    def test_no_logged(self) -> None:
        """Request with a non-logged user: does not return any token."""
        response = self.client.get(
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            )
        )
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertEqual(
            response.context["exception"], "No User matches the given query."
        )


class UserTokenCreateViewTests(TestCase):
    """Tests for UserTokenCreateView."""

    user: ClassVar[User]

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

    def test_create_token(self) -> None:
        """Post to "user:token-create" to create a token."""
        self.client.force_login(self.user)
        comment = "Test 1"
        response = self.client.post(
            reverse(
                "user:token-create",
                kwargs={"username": self.user.username},
            ),
            {"comment": comment},
        )

        # The token got created...
        token = Token.objects.get(comment=comment)

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

        # check the returned token matches one in the db
        token_key_re = re.compile(
            r".*<code>(?P<token>[a-z0-9]*).*</code>.*",
            re.DOTALL,
        )
        match = token_key_re.match(response.content.decode())
        assert match is not None
        resp_key = match.group("token")
        self.assertEqual(token, Token.objects.get_token_or_none(resp_key))

        self.assertContains(
            response,
            _html_collapse_token(resp_key),
            status_code=status.HTTP_200_OK,
            html=True,
        )

    def test_no_logged(self) -> None:
        """Request with a non-logged user: cannot create a token."""
        response = self.client.get(
            reverse(
                "user:token-create",
                kwargs={"username": self.user.username},
            )
        )
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertEqual(
            response.context["exception"], "No User matches the given query."
        )

    def test_initial_form(self) -> None:
        """Get request to ensure the form is displayed."""
        self.client.force_login(self.user)

        response = self.client.get(
            reverse(
                "user:token-create",
                kwargs={"username": self.user.username},
            )
        )

        self.assertContains(
            response, "<title>Debusine - Create token</title>", html=True
        )
        self.assertContains(response, "<h1>Create token</h1>", html=True)
        self.assertContains(
            response,
            '<input class="btn btn-primary btn-sm" '
            'type="submit" value="Create">',
            html=True,
        )


class UserTokenUpdateViewTests(TestCase):
    """Tests for UserTokenCreateView when editing a token."""

    scenario = scenarios.DefaultScopeUserAPI()

    def test_edit_token(self) -> None:
        """Get "user:token-edit" to edit a token."""
        token = self.scenario.user_token
        self.client.force_login(self.scenario.user)

        enabled = True
        comment = "This is a token"

        response = self.client.post(
            reverse(
                "user:token-edit",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            ),
            {"comment": comment, "enabled": enabled},
        )

        self.assertRedirects(
            response,
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            ),
        )

        token.refresh_from_db()

        self.assertEqual(token.comment, comment)
        self.assertEqual(token.enabled, enabled)

    def test_get_404_not_found(self) -> None:
        """Get of token/PK/ that exist but belongs to another user: 404."""
        token = self.scenario.user_token
        user2 = get_user_model().objects.create(
            username="testuser2",
            password="testpass2",
            email="user2@example.com",
        )

        self.client.force_login(user2)
        response = self.client.get(
            reverse(
                "user:token-edit",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
        self.assertEqual(
            response.context["exception"],
            "testuser2 cannot manage user playground",
        )
        self.assertNotContains(
            response, token.hash, status_code=status.HTTP_403_FORBIDDEN
        )

    def test_put_404_not_found(self) -> None:
        """Put of token/PK/ that exist but belongs to another user: 404."""
        token = self.scenario.user_token
        user2 = get_user_model().objects.create(
            username="testuser2",
            password="testpass2",
            email="user2@example.com",
        )

        self.client.force_login(user2)
        comment = "This is the new comment"
        enabled = False
        response = self.client.put(
            reverse(
                "user:token-edit",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            ),
            {"comment": comment, "enabled": enabled},
        )
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
        self.assertEqual(
            response.context["exception"],
            "testuser2 cannot manage user playground",
        )
        self.assertNotContains(
            response, token.hash, status_code=status.HTTP_403_FORBIDDEN
        )

    def test_initial_form(self) -> None:
        """Get request to ensure the form is displayed."""
        token = self.scenario.user_token
        self.client.force_login(self.scenario.user)

        response = self.client.get(
            reverse(
                "user:token-edit",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )

        self.assertContains(
            response, "<title>Debusine - Edit token</title>", html=True
        )
        self.assertContains(response, "<h1>Edit token</h1>", html=True)
        self.assertContains(
            response,
            '<input class="btn btn-primary btn-sm" type="submit" value="Edit">',
            html=True,
        )
        assert isinstance(response, SimpleTemplateResponse)
        self.assertEqual(
            response.context_data["form"]["comment"].initial,
            token.comment,
        )


class UserTokenDeleteViewTests(TestCase):
    """Tests for UserTokenDeleteView."""

    scenario = scenarios.DefaultScopeUserAPI()

    def test_get_delete_token_authenticated_user(self) -> None:
        """Get request to delete a token. Token information is returned."""
        token = self.scenario.user_token
        self.client.force_login(self.scenario.user)

        response = self.client.get(
            reverse(
                "user:token-delete",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )

        expected_contents = (
            "<ul>"
            f"<li>Comment: {token.comment}</li>"
            f"<li>Enabled: {html_check_icon(token.enabled)}</li>"
            f"<li>Created at: {date_format(token.created_at)}</li>"
            "</ul>"
        )

        self.assertContains(response, expected_contents, html=True)

    def test_post_delete_token_authenticated_user(self) -> None:
        """Post request to delete a token. The token is deleted."""
        token = self.scenario.user_token
        self.client.force_login(self.scenario.user)

        response = self.client.post(
            reverse(
                "user:token-delete",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )

        self.assertRedirects(
            response,
            reverse(
                "user:token-list",
                kwargs={"username": self.scenario.user.username},
            ),
        )

        # The token was deleted
        self.assertEqual(Token.objects.filter(pk=token.pk).count(), 0)

    def test_get_delete_token_non_authenticated(self) -> None:
        """
        Non authenticated Get and Post request. Return 403.

        The token is not displayed.
        """
        token = self.scenario.user_token
        response = self.client.get(
            reverse(
                "user:token-delete",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertEqual(
            response.context["exception"], "No User matches the given query."
        )
        self.assertNotContains(
            response, token.hash, status_code=status.HTTP_404_NOT_FOUND
        )

        response = self.client.post(
            reverse(
                "user:token-delete",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertEqual(
            response.context["exception"], "No User matches the given query."
        )
        self.assertNotContains(
            response, token.hash, status_code=status.HTTP_404_NOT_FOUND
        )

    def test_get_post_delete_token_another_user(self) -> None:
        """
        Get and Post request delete token from another user. Return 404.

        The token is not deleted neither displayed.
        """
        token = self.scenario.user_token
        user2 = get_user_model().objects.create_user(
            username="testuser2",
            password="testpass2",
            email="user2@example.com",
        )

        self.client.force_login(user2)

        response = self.client.get(
            reverse(
                "user:token-delete",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
        self.assertEqual(
            response.context["exception"],
            "testuser2 cannot manage user playground",
        )
        self.assertNotContains(
            response, token.hash, status_code=status.HTTP_403_FORBIDDEN
        )

        response = self.client.post(
            reverse(
                "user:token-delete",
                kwargs={
                    "pk": token.pk,
                    "username": self.scenario.user.username,
                },
            )
        )
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
        self.assertEqual(
            response.context["exception"],
            "testuser2 cannot manage user playground",
        )
        self.assertNotContains(
            response, token.hash, status_code=status.HTTP_403_FORBIDDEN
        )

        self.assertEqual(Token.objects.filter(pk=token.pk).count(), 1)


def _html_collapse_token(key: str) -> str:
    """Return HTML for a show/hide button for token key."""
    return f'''<button class="btn btn-primary btn-sm" type="button"
        data-bs-toggle="collapse"
            data-bs-target="#{key}Div">
        Show / Hide Token
        </button>
        <div class="collapse" id="{key}Div">
            <code>{key}</code>
        </div>
    '''
