authentik.outposts.tests.test_ws

Websocket tests

 1"""Websocket tests"""
 2
 3from dataclasses import asdict
 4
 5from channels.routing import URLRouter
 6from channels.testing import WebsocketCommunicator
 7from django.test import TransactionTestCase
 8
 9from authentik import authentik_version
10from authentik.core.tests.utils import create_test_flow
11from authentik.outposts.consumer import WebsocketMessage, WebsocketMessageInstruction
12from authentik.outposts.models import Outpost, OutpostType
13from authentik.providers.proxy.models import ProxyProvider
14from authentik.root import websocket
15
16
17class TestOutpostWS(TransactionTestCase):
18    """Websocket tests"""
19
20    def setUp(self) -> None:
21        self.provider: ProxyProvider = ProxyProvider.objects.create(
22            name="test",
23            internal_host="http://localhost",
24            external_host="http://localhost",
25            authorization_flow=create_test_flow(),
26        )
27        self.outpost: Outpost = Outpost.objects.create(
28            name="test",
29            type=OutpostType.PROXY,
30        )
31        self.outpost.providers.add(self.provider)
32        self.token = self.outpost.token.key
33
34    async def test_auth(self):
35        """Test auth without token"""
36        communicator = WebsocketCommunicator(
37            URLRouter(websocket.websocket_urlpatterns), f"/ws/outpost/{self.outpost.pk}/"
38        )
39        connected, _ = await communicator.connect()
40        self.assertFalse(connected)
41        await communicator.disconnect()
42
43    async def test_auth_valid(self):
44        """Test auth with token"""
45        communicator = WebsocketCommunicator(
46            URLRouter(websocket.websocket_urlpatterns),
47            f"/ws/outpost/{self.outpost.pk}/",
48            [(b"authorization", f"Bearer {self.token}".encode())],
49        )
50        connected, _ = await communicator.connect()
51        self.assertTrue(connected)
52        await communicator.disconnect()
53
54    async def test_send(self):
55        """Test sending of Hello"""
56        communicator = WebsocketCommunicator(
57            URLRouter(websocket.websocket_urlpatterns),
58            f"/ws/outpost/{self.outpost.pk}/",
59            [(b"authorization", f"Bearer {self.token}".encode())],
60        )
61        connected, _ = await communicator.connect()
62        self.assertTrue(connected)
63        await communicator.send_json_to(
64            asdict(
65                WebsocketMessage(
66                    instruction=WebsocketMessageInstruction.HELLO,
67                    args={
68                        "version": authentik_version(),
69                        "buildHash": "foo",
70                        "uuid": "123",
71                    },
72                )
73            )
74        )
75        response = await communicator.receive_json_from()
76        self.assertEqual(
77            response, asdict(WebsocketMessage(instruction=WebsocketMessageInstruction.ACK, args={}))
78        )
79        await communicator.disconnect()
80
81    async def test_send_ack(self):
82        """Test sending of ACK"""
83        communicator = WebsocketCommunicator(
84            URLRouter(websocket.websocket_urlpatterns),
85            f"/ws/outpost/{self.outpost.pk}/",
86            [(b"authorization", f"Bearer {self.token}".encode())],
87        )
88        connected, _ = await communicator.connect()
89        self.assertTrue(connected)
90        await communicator.send_json_to(
91            asdict(
92                WebsocketMessage(
93                    instruction=WebsocketMessageInstruction.ACK,
94                    args={},
95                )
96            )
97        )
98        await communicator.disconnect()
class TestOutpostWS(django.test.testcases.TransactionTestCase): 
18class TestOutpostWS(TransactionTestCase):
19    """Websocket tests"""
20
21    def setUp(self) -> None:
22        self.provider: ProxyProvider = ProxyProvider.objects.create(
23            name="test",
24            internal_host="http://localhost",
25            external_host="http://localhost",
26            authorization_flow=create_test_flow(),
27        )
28        self.outpost: Outpost = Outpost.objects.create(
29            name="test",
30            type=OutpostType.PROXY,
31        )
32        self.outpost.providers.add(self.provider)
33        self.token = self.outpost.token.key
34
35    async def test_auth(self):
36        """Test auth without token"""
37        communicator = WebsocketCommunicator(
38            URLRouter(websocket.websocket_urlpatterns), f"/ws/outpost/{self.outpost.pk}/"
39        )
40        connected, _ = await communicator.connect()
41        self.assertFalse(connected)
42        await communicator.disconnect()
43
44    async def test_auth_valid(self):
45        """Test auth with token"""
46        communicator = WebsocketCommunicator(
47            URLRouter(websocket.websocket_urlpatterns),
48            f"/ws/outpost/{self.outpost.pk}/",
49            [(b"authorization", f"Bearer {self.token}".encode())],
50        )
51        connected, _ = await communicator.connect()
52        self.assertTrue(connected)
53        await communicator.disconnect()
54
55    async def test_send(self):
56        """Test sending of Hello"""
57        communicator = WebsocketCommunicator(
58            URLRouter(websocket.websocket_urlpatterns),
59            f"/ws/outpost/{self.outpost.pk}/",
60            [(b"authorization", f"Bearer {self.token}".encode())],
61        )
62        connected, _ = await communicator.connect()
63        self.assertTrue(connected)
64        await communicator.send_json_to(
65            asdict(
66                WebsocketMessage(
67                    instruction=WebsocketMessageInstruction.HELLO,
68                    args={
69                        "version": authentik_version(),
70                        "buildHash": "foo",
71                        "uuid": "123",
72                    },
73                )
74            )
75        )
76        response = await communicator.receive_json_from()
77        self.assertEqual(
78            response, asdict(WebsocketMessage(instruction=WebsocketMessageInstruction.ACK, args={}))
79        )
80        await communicator.disconnect()
81
82    async def test_send_ack(self):
83        """Test sending of ACK"""
84        communicator = WebsocketCommunicator(
85            URLRouter(websocket.websocket_urlpatterns),
86            f"/ws/outpost/{self.outpost.pk}/",
87            [(b"authorization", f"Bearer {self.token}".encode())],
88        )
89        connected, _ = await communicator.connect()
90        self.assertTrue(connected)
91        await communicator.send_json_to(
92            asdict(
93                WebsocketMessage(
94                    instruction=WebsocketMessageInstruction.ACK,
95                    args={},
96                )
97            )
98        )
99        await communicator.disconnect()

Websocket tests

def setUp(self) -> None: 
21    def setUp(self) -> None:
22        self.provider: ProxyProvider = ProxyProvider.objects.create(
23            name="test",
24            internal_host="http://localhost",
25            external_host="http://localhost",
26            authorization_flow=create_test_flow(),
27        )
28        self.outpost: Outpost = Outpost.objects.create(
29            name="test",
30            type=OutpostType.PROXY,
31        )
32        self.outpost.providers.add(self.provider)
33        self.token = self.outpost.token.key

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

async def test_auth(self): 
35    async def test_auth(self):
36        """Test auth without token"""
37        communicator = WebsocketCommunicator(
38            URLRouter(websocket.websocket_urlpatterns), f"/ws/outpost/{self.outpost.pk}/"
39        )
40        connected, _ = await communicator.connect()
41        self.assertFalse(connected)
42        await communicator.disconnect()

Test auth without token

async def test_auth_valid(self): 
44    async def test_auth_valid(self):
45        """Test auth with token"""
46        communicator = WebsocketCommunicator(
47            URLRouter(websocket.websocket_urlpatterns),
48            f"/ws/outpost/{self.outpost.pk}/",
49            [(b"authorization", f"Bearer {self.token}".encode())],
50        )
51        connected, _ = await communicator.connect()
52        self.assertTrue(connected)
53        await communicator.disconnect()

Test auth with token

async def test_send(self): 
55    async def test_send(self):
56        """Test sending of Hello"""
57        communicator = WebsocketCommunicator(
58            URLRouter(websocket.websocket_urlpatterns),
59            f"/ws/outpost/{self.outpost.pk}/",
60            [(b"authorization", f"Bearer {self.token}".encode())],
61        )
62        connected, _ = await communicator.connect()
63        self.assertTrue(connected)
64        await communicator.send_json_to(
65            asdict(
66                WebsocketMessage(
67                    instruction=WebsocketMessageInstruction.HELLO,
68                    args={
69                        "version": authentik_version(),
70                        "buildHash": "foo",
71                        "uuid": "123",
72                    },
73                )
74            )
75        )
76        response = await communicator.receive_json_from()
77        self.assertEqual(
78            response, asdict(WebsocketMessage(instruction=WebsocketMessageInstruction.ACK, args={}))
79        )
80        await communicator.disconnect()

Test sending of Hello

async def test_send_ack(self): 
82    async def test_send_ack(self):
83        """Test sending of ACK"""
84        communicator = WebsocketCommunicator(
85            URLRouter(websocket.websocket_urlpatterns),
86            f"/ws/outpost/{self.outpost.pk}/",
87            [(b"authorization", f"Bearer {self.token}".encode())],
88        )
89        connected, _ = await communicator.connect()
90        self.assertTrue(connected)
91        await communicator.send_json_to(
92            asdict(
93                WebsocketMessage(
94                    instruction=WebsocketMessageInstruction.ACK,
95                    args={},
96                )
97            )
98        )
99        await communicator.disconnect()

Test sending of ACK