gestioCOF/kfet/auth/tests.py

from unittest import mock

from django.contrib.auth.models import AnonymousUser, Group, Permission, User
from django.core import signing
from django.test import RequestFactory, TestCase
from django.urls import reverse

from kfet.forms import UserGroupForm
from kfet.models import Account

from . import KFET_GENERIC_TRIGRAMME, KFET_GENERIC_USERNAME
from .backends import AccountBackend, GenericBackend
from .models import GenericTeamToken, KFetGroup
from .utils import get_kfet_generic_user
from .views import GenericLoginView

##
# Forms
##


class UserGroupFormTests(TestCase):
    """Test suite for UserGroupForm."""

    def setUp(self):
        # create user
        self.user = User.objects.create(username="foo", password="foo")

        # create some K-Fêt groups
        names = ["Group 1", "Group 2", "Group 3"]
        self.kfet_groups = [KFetGroup.objects.create(name=name) for name in names]

        # create a non-K-Fêt group
        self.other_group = Group.objects.create(name="Other group")

    def test_choices(self):
        """Only K-Fêt groups are selectable."""
        form = UserGroupForm(instance=self.user)
        groups_field = form.fields["groups"]
        self.assertQuerysetEqual(
            groups_field.queryset,
            [repr(g.group_ptr) for g in self.kfet_groups],
            ordered=False,
        )

    def test_keep_others(self):
        """User stays in its non-K-Fêt groups."""
        user = self.user

        # add user to a non-K-Fêt group
        user.groups.add(self.other_group)

        # add user to some K-Fêt groups through UserGroupForm
        data = {"groups": [group.pk for group in self.kfet_groups]}
        form = UserGroupForm(data, instance=user)

        form.is_valid()
        form.save()
        self.assertQuerysetEqual(
            user.groups.all(),
            [self.other_group.pk] + [group.pk for group in self.kfet_groups],
            transform=lambda group: group.pk,
            ordered=False,
        )


class KFetGenericUserTests(TestCase):
    def test_exists(self):
        """
        The account is set up when app is ready, so it should exist.
        """
        generic = Account.objects.get_generic()
        self.assertEqual(generic.trigramme, KFET_GENERIC_TRIGRAMME)
        self.assertEqual(generic.user.username, KFET_GENERIC_USERNAME)
        self.assertEqual(get_kfet_generic_user(), generic.user)


##
# Backends
##


class AccountBackendTests(TestCase):
    def setUp(self):
        self.request = RequestFactory().get("/")

    def test_valid(self):
        acc = Account(trigramme="000")
        acc.change_pwd("valid")
        acc.save({"username": "user"})

        auth = AccountBackend().authenticate(self.request, kfet_password="valid")

        self.assertEqual(auth, acc.user)

    def test_invalid(self):
        auth = AccountBackend().authenticate(self.request, kfet_password="invalid")
        self.assertIsNone(auth)


class GenericBackendTests(TestCase):
    def setUp(self):
        self.request = RequestFactory().get("/")

    def test_valid(self):
        token = GenericTeamToken.objects.create_token()

        auth = GenericBackend().authenticate(self.request, kfet_token=token.token)

        self.assertEqual(auth, get_kfet_generic_user())
        self.assertEqual(GenericTeamToken.objects.all().count(), 0)

    def test_invalid(self):
        auth = GenericBackend().authenticate(self.request, kfet_token="invalid")
        self.assertIsNone(auth)


##
# Views
##


class GenericLoginViewTests(TestCase):
    def setUp(self):
        patcher_messages = mock.patch("gestioncof.signals.messages")
        patcher_messages.start()
        self.addCleanup(patcher_messages.stop)

        user_acc = Account(trigramme="000")
        user_acc.save({"username": "user"})
        self.user = user_acc.user
        self.user.set_password("user")
        self.user.save()

        team_acc = Account(trigramme="100")
        team_acc.save({"username": "team"})
        self.team = team_acc.user
        self.team.set_password("team")
        self.team.save()
        self.team.user_permissions.add(
            Permission.objects.get(content_type__app_label="kfet", codename="is_team")
        )

        self.url = reverse("kfet.login.generic")
        self.generic_user = get_kfet_generic_user()

    def test_url(self):
        self.assertEqual(self.url, "/k-fet/login/generic")

    def test_notoken_get(self):
        """
        Send confirmation for user to emit POST request, instead of GET.
        """
        self.client.login(username="team", password="team")

        r = self.client.get(self.url)

        self.assertEqual(r.status_code, 200)
        self.assertTemplateUsed(r, "kfet/confirm_form.html")

    def test_notoken_post(self):
        """
        POST request without token in COOKIES sets a token and redirects to
        logout url.
        """
        self.client.login(username="team", password="team")

        r = self.client.post(self.url)

        self.assertRedirects(
            r, "/gestion/logout?next={}".format(self.url), fetch_redirect_response=False
        )

    def test_notoken_not_team(self):
        """
        Logged in user must be a team user to initiate login as generic user.
        """
        self.client.login(username="user", password="user")

        # With GET.
        r = self.client.get(self.url)
        self.assertRedirects(
            r, "/gestion/login?next={}".format(self.url), fetch_redirect_response=False
        )

        # Also with POST.
        r = self.client.post(self.url)
        self.assertRedirects(
            r, "/gestion/login?next={}".format(self.url), fetch_redirect_response=False
        )

    def _set_signed_cookie(self, client, key, value):
        signed_value = signing.get_cookie_signer(salt=key).sign(value)
        client.cookies.load({key: signed_value})

    def _is_cookie_deleted(self, client, key):
        try:
            self.assertNotIn(key, client.cookies)
        except AssertionError:
            try:
                cookie = client.cookies[key]
                # It also can be emptied.
                self.assertEqual(cookie.value, "")
                self.assertEqual(cookie["expires"], "Thu, 01 Jan 1970 00:00:00 GMT")
                self.assertEqual(cookie["max-age"], 0)
            except AssertionError:
                raise AssertionError("The cookie '%s' still exists." % key)

    def test_withtoken_valid(self):
        """
        The kfet generic user is logged in.
        """
        token = GenericTeamToken.objects.create(token="valid")
        self._set_signed_cookie(
            self.client, GenericLoginView.TOKEN_COOKIE_NAME, "valid"
        )

        r = self.client.get(self.url)

        self.assertRedirects(r, reverse("kfet.kpsul"))
        self.assertEqual(r.wsgi_request.user, self.generic_user)
        self._is_cookie_deleted(self.client, GenericLoginView.TOKEN_COOKIE_NAME)
        with self.assertRaises(GenericTeamToken.DoesNotExist):
            token.refresh_from_db()

    def test_withtoken_invalid(self):
        """
        If token is invalid, delete it and try again.
        """
        self._set_signed_cookie(
            self.client, GenericLoginView.TOKEN_COOKIE_NAME, "invalid"
        )

        r = self.client.get(self.url)

        self.assertRedirects(r, self.url, fetch_redirect_response=False)
        self.assertEqual(r.wsgi_request.user, AnonymousUser())
        self._is_cookie_deleted(self.client, GenericLoginView.TOKEN_COOKIE_NAME)

    def test_flow_ok(self):
        """
        A team user is logged in as the kfet generic user.
        """
        self.client.login(username="team", password="team")
        next_url = "/k-fet/"

        r = self.client.post("{}?next={}".format(self.url, next_url), follow=True)

        self.assertEqual(r.wsgi_request.user, self.generic_user)
        self.assertEqual(r.wsgi_request.path, "/k-fet/")


##
# Temporary authentication
#
# Includes:
#   - TemporaryAuthMiddleware
#   - temporary_auth context processor
##


class TemporaryAuthTests(TestCase):
    def setUp(self):
        patcher_messages = mock.patch("gestioncof.signals.messages")
        patcher_messages.start()
        self.addCleanup(patcher_messages.stop)

        self.factory = RequestFactory()

        user1_acc = Account(trigramme="000")
        user1_acc.change_pwd("kfet_user1")
        user1_acc.save({"username": "user1"})
        self.user1 = user1_acc.user
        self.user1.set_password("user1")
        self.user1.save()

        user2_acc = Account(trigramme="100")
        user2_acc.change_pwd("kfet_user2")
        user2_acc.save({"username": "user2"})
        self.user2 = user2_acc.user
        self.user2.set_password("user2")
        self.user2.save()

        self.perm = Permission.objects.get(
            content_type__app_label="kfet", codename="is_team"
        )
        self.perm2 = Permission.objects.get(
            content_type__app_label="kfet", codename="can_force_close"
        )
        self.user1.user_permissions.add(self.perm)
        self.user2.user_permissions.add(self.perm, self.perm2)

    def test_context_processor(self):
        """
        Context variables give the real authenticated user and his permissions.
        """
        self.client.login(username="user1", password="user1")

        r = self.client.post("/k-fet/accounts/000/edit", HTTP_KFETPASSWORD="kfet_user2")

        self.assertEqual(r.context["user"], self.user1)
        self.assertNotIn("kfet.can_force_close", r.context["perms"])

    def test_auth_not_persistent(self):
        """
        The authentication is temporary, i.e. for one request.
        """
        self.client.login(username="user1", password="user1")

        r1 = self.client.post(
            "/k-fet/accounts/000/edit", HTTP_KFETPASSWORD="kfet_user2"
        )
        self.assertEqual(r1.wsgi_request.user, self.user2)

        r2 = self.client.post("/k-fet/accounts/000/edit")
        self.assertEqual(r2.wsgi_request.user, self.user1)