authentik.root.tests.test_ws_client

View Source

  1from unittest.mock import patch
  2
  3from asgiref.sync import sync_to_async
  4from channels.routing import URLRouter
  5from channels.testing import WebsocketCommunicator
  6from django.http import HttpRequest
  7from django.test import TransactionTestCase
  8
  9from authentik.core.tests.utils import create_test_user
 10from authentik.events.models import (
 11    Event,
 12    EventAction,
 13    Notification,
 14    NotificationTransport,
 15    TransportMode,
 16)
 17from authentik.flows.apps import RefreshOtherFlowsAfterAuthentication
 18from authentik.lib.generators import generate_id
 19from authentik.root import websocket
 20from authentik.stages.password import BACKEND_INBUILT
 21from authentik.stages.user_login.stage import COOKIE_NAME_KNOWN_DEVICE
 22from authentik.tenants.utils import get_current_tenant
 23
 24
 25class TestClientWS(TransactionTestCase):
 26
 27    def setUp(self):
 28        tenant = get_current_tenant()
 29        tenant.flags[RefreshOtherFlowsAfterAuthentication().key] = True
 30        tenant.save()
 31        self.user = create_test_user()
 32
 33    async def _alogin_cookie(self, user, **kwargs):
 34        """Similar to `client.aforce_login` but allow setting of cookies"""
 35        from django.contrib.auth import alogin
 36
 37        # Create a fake request to store login details.
 38        request = HttpRequest()
 39        session = await self.client.asession()
 40        request.session = session
 41        request.COOKIES.update(kwargs)
 42
 43        await alogin(request, user, BACKEND_INBUILT)
 44        # Save the session values.
 45        await request.session.asave()
 46        self.client._set_login_cookies(request)
 47
 48    async def test_auth_blank(self):
 49        dev_id = generate_id()
 50        communicator = WebsocketCommunicator(
 51            URLRouter(websocket.websocket_urlpatterns),
 52            "/ws/client/",
 53            headers=[(b"cookie", f"{COOKIE_NAME_KNOWN_DEVICE}={dev_id}".encode())],
 54        )
 55        connected, _ = await communicator.connect()
 56        self.assertTrue(connected)
 57
 58        await self._alogin_cookie(self.user, **{COOKIE_NAME_KNOWN_DEVICE: dev_id})
 59
 60        await communicator.receive_nothing()
 61        await communicator.receive_json_from()
 62        await communicator.disconnect()
 63
 64    async def test_tab_refresh(self):
 65        dev_id = generate_id()
 66        communicator = WebsocketCommunicator(
 67            URLRouter(websocket.websocket_urlpatterns),
 68            "/ws/client/",
 69            headers=[(b"cookie", f"{COOKIE_NAME_KNOWN_DEVICE}={dev_id}".encode())],
 70        )
 71        connected, _ = await communicator.connect()
 72        self.assertTrue(connected)
 73
 74        with patch("authentik.flows.apps.RefreshOtherFlowsAfterAuthentication.get") as flag:
 75            flag.return_value = True
 76            await self._alogin_cookie(self.user, **{COOKIE_NAME_KNOWN_DEVICE: dev_id})
 77
 78        evt = await communicator.receive_json_from()
 79        self.assertEqual(
 80            evt, {"message_type": "session.authenticated", "type": "event.session.authenticated"}
 81        )
 82
 83        await communicator.disconnect()
 84
 85    async def test_notification(self):
 86        communicator = WebsocketCommunicator(
 87            URLRouter(websocket.websocket_urlpatterns), "/ws/client/"
 88        )
 89        communicator.scope["user"] = self.user
 90        connected, _ = await communicator.connect()
 91        self.assertTrue(connected)
 92
 93        transport = await NotificationTransport.objects.acreate(
 94            name=generate_id(), mode=TransportMode.LOCAL
 95        )
 96        event = await sync_to_async(Event.new)(EventAction.LOGIN)
 97        event.set_user(self.user)
 98        await event.asave()
 99        notification = Notification(
100            user=self.user,
101            body="foo",
102            event=event,
103            hyperlink="goauthentik.io",
104            hyperlink_label="a link",
105        )
106        await sync_to_async(transport.send_local)(notification)
107
108        evt = await communicator.receive_json_from(timeout=5)
109        self.assertEqual(evt["message_type"], "notification.new")
110        self.assertEqual(evt["id"], str(notification.pk))
111        self.assertEqual(evt["data"]["pk"], str(notification.pk))
112        self.assertEqual(evt["data"]["body"], "foo")
113        self.assertEqual(evt["data"]["event"]["pk"], str(event.pk))
114
115        await communicator.disconnect()

class TestClientWS(django.test.testcases.TransactionTestCase): View Source

 26class TestClientWS(TransactionTestCase):
 27
 28    def setUp(self):
 29        tenant = get_current_tenant()
 30        tenant.flags[RefreshOtherFlowsAfterAuthentication().key] = True
 31        tenant.save()
 32        self.user = create_test_user()
 33
 34    async def _alogin_cookie(self, user, **kwargs):
 35        """Similar to `client.aforce_login` but allow setting of cookies"""
 36        from django.contrib.auth import alogin
 37
 38        # Create a fake request to store login details.
 39        request = HttpRequest()
 40        session = await self.client.asession()
 41        request.session = session
 42        request.COOKIES.update(kwargs)
 43
 44        await alogin(request, user, BACKEND_INBUILT)
 45        # Save the session values.
 46        await request.session.asave()
 47        self.client._set_login_cookies(request)
 48
 49    async def test_auth_blank(self):
 50        dev_id = generate_id()
 51        communicator = WebsocketCommunicator(
 52            URLRouter(websocket.websocket_urlpatterns),
 53            "/ws/client/",
 54            headers=[(b"cookie", f"{COOKIE_NAME_KNOWN_DEVICE}={dev_id}".encode())],
 55        )
 56        connected, _ = await communicator.connect()
 57        self.assertTrue(connected)
 58
 59        await self._alogin_cookie(self.user, **{COOKIE_NAME_KNOWN_DEVICE: dev_id})
 60
 61        await communicator.receive_nothing()
 62        await communicator.receive_json_from()
 63        await communicator.disconnect()
 64
 65    async def test_tab_refresh(self):
 66        dev_id = generate_id()
 67        communicator = WebsocketCommunicator(
 68            URLRouter(websocket.websocket_urlpatterns),
 69            "/ws/client/",
 70            headers=[(b"cookie", f"{COOKIE_NAME_KNOWN_DEVICE}={dev_id}".encode())],
 71        )
 72        connected, _ = await communicator.connect()
 73        self.assertTrue(connected)
 74
 75        with patch("authentik.flows.apps.RefreshOtherFlowsAfterAuthentication.get") as flag:
 76            flag.return_value = True
 77            await self._alogin_cookie(self.user, **{COOKIE_NAME_KNOWN_DEVICE: dev_id})
 78
 79        evt = await communicator.receive_json_from()
 80        self.assertEqual(
 81            evt, {"message_type": "session.authenticated", "type": "event.session.authenticated"}
 82        )
 83
 84        await communicator.disconnect()
 85
 86    async def test_notification(self):
 87        communicator = WebsocketCommunicator(
 88            URLRouter(websocket.websocket_urlpatterns), "/ws/client/"
 89        )
 90        communicator.scope["user"] = self.user
 91        connected, _ = await communicator.connect()
 92        self.assertTrue(connected)
 93
 94        transport = await NotificationTransport.objects.acreate(
 95            name=generate_id(), mode=TransportMode.LOCAL
 96        )
 97        event = await sync_to_async(Event.new)(EventAction.LOGIN)
 98        event.set_user(self.user)
 99        await event.asave()
100        notification = Notification(
101            user=self.user,
102            body="foo",
103            event=event,
104            hyperlink="goauthentik.io",
105            hyperlink_label="a link",
106        )
107        await sync_to_async(transport.send_local)(notification)
108
109        evt = await communicator.receive_json_from(timeout=5)
110        self.assertEqual(evt["message_type"], "notification.new")
111        self.assertEqual(evt["id"], str(notification.pk))
112        self.assertEqual(evt["data"]["pk"], str(notification.pk))
113        self.assertEqual(evt["data"]["body"], "foo")
114        self.assertEqual(evt["data"]["event"]["pk"], str(event.pk))
115
116        await communicator.disconnect()

A class whose instances are single test cases.

By default, the test code itself should be placed in a method named 'runTest'.

If the fixture may be used for many test cases, create as many test methods as are needed. When instantiating such a TestCase subclass, specify in the constructor arguments the name of the test method that the instance is to execute.

Test authors should subclass TestCase for their own tests. Construction and deconstruction of the test's environment ('fixture') can be implemented by overriding the 'setUp' and 'tearDown' methods respectively.

If it is necessary to override the __init__ method, the base class __init__ method must always be called. It is important that subclasses should not change the signature of their __init__ method, since instances of the classes are instantiated automatically by parts of the framework in order to be run.

When subclassing TestCase, you can set these attributes:

failureException: determines which exception will be raised when the instance's assertion methods fail; test methods raising this exception will be deemed to have 'failed' rather than 'errored'.
longMessage: determines whether long messages (including repr of objects used in assert methods) will be printed on failure in addition to any explicit message passed.
maxDiff: sets the maximum length of a diff in failure messages by assert methods using difflib. It is looked up as an instance attribute so can be configured by individual tests if required.

def setUp(self): View Source

28    def setUp(self):
29        tenant = get_current_tenant()
30        tenant.flags[RefreshOtherFlowsAfterAuthentication().key] = True
31        tenant.save()
32        self.user = create_test_user()

Hook method for setting up the test fixture before exercising it.

async def test_auth_blank(self): View Source

49    async def test_auth_blank(self):
50        dev_id = generate_id()
51        communicator = WebsocketCommunicator(
52            URLRouter(websocket.websocket_urlpatterns),
53            "/ws/client/",
54            headers=[(b"cookie", f"{COOKIE_NAME_KNOWN_DEVICE}={dev_id}".encode())],
55        )
56        connected, _ = await communicator.connect()
57        self.assertTrue(connected)
58
59        await self._alogin_cookie(self.user, **{COOKIE_NAME_KNOWN_DEVICE: dev_id})
60
61        await communicator.receive_nothing()
62        await communicator.receive_json_from()
63        await communicator.disconnect()

async def test_tab_refresh(self): View Source

65    async def test_tab_refresh(self):
66        dev_id = generate_id()
67        communicator = WebsocketCommunicator(
68            URLRouter(websocket.websocket_urlpatterns),
69            "/ws/client/",
70            headers=[(b"cookie", f"{COOKIE_NAME_KNOWN_DEVICE}={dev_id}".encode())],
71        )
72        connected, _ = await communicator.connect()
73        self.assertTrue(connected)
74
75        with patch("authentik.flows.apps.RefreshOtherFlowsAfterAuthentication.get") as flag:
76            flag.return_value = True
77            await self._alogin_cookie(self.user, **{COOKIE_NAME_KNOWN_DEVICE: dev_id})
78
79        evt = await communicator.receive_json_from()
80        self.assertEqual(
81            evt, {"message_type": "session.authenticated", "type": "event.session.authenticated"}
82        )
83
84        await communicator.disconnect()

async def test_notification(self): View Source

 86    async def test_notification(self):
 87        communicator = WebsocketCommunicator(
 88            URLRouter(websocket.websocket_urlpatterns), "/ws/client/"
 89        )
 90        communicator.scope["user"] = self.user
 91        connected, _ = await communicator.connect()
 92        self.assertTrue(connected)
 93
 94        transport = await NotificationTransport.objects.acreate(
 95            name=generate_id(), mode=TransportMode.LOCAL
 96        )
 97        event = await sync_to_async(Event.new)(EventAction.LOGIN)
 98        event.set_user(self.user)
 99        await event.asave()
100        notification = Notification(
101            user=self.user,
102            body="foo",
103            event=event,
104            hyperlink="goauthentik.io",
105            hyperlink_label="a link",
106        )
107        await sync_to_async(transport.send_local)(notification)
108
109        evt = await communicator.receive_json_from(timeout=5)
110        self.assertEqual(evt["message_type"], "notification.new")
111        self.assertEqual(evt["id"], str(notification.pk))
112        self.assertEqual(evt["data"]["pk"], str(notification.pk))
113        self.assertEqual(evt["data"]["body"], "foo")
114        self.assertEqual(evt["data"]["event"]["pk"], str(event.pk))
115
116        await communicator.disconnect()