authentik.sources.plex.tests

plex Source tests

  1"""plex Source tests"""
  2
  3from django.test import TestCase
  4from requests.exceptions import RequestException
  5from requests_mock import Mocker
  6
  7from authentik.events.models import Event, EventAction
  8from authentik.lib.generators import generate_key
  9from authentik.sources.plex.models import PlexSource
 10from authentik.sources.plex.plex import PlexAuth
 11from authentik.sources.plex.tasks import check_plex_token
 12
 13USER_INFO_RESPONSE = {
 14    "id": 1234123419,
 15    "uuid": "qwerqewrqewrqwr",
 16    "username": "username",
 17    "title": "title",
 18    "email": "foo@bar.baz",
 19}
 20RESOURCES_RESPONSE = [
 21    {
 22        "name": "foo",
 23        "clientIdentifier": "allowed",
 24        "provides": "server",
 25    },
 26    {
 27        "name": "foo",
 28        "clientIdentifier": "denied",
 29        "provides": "server",
 30    },
 31]
 32
 33
 34class TestPlexSource(TestCase):
 35    """plex Source tests"""
 36
 37    def setUp(self):
 38        self.source: PlexSource = PlexSource.objects.create(
 39            name="test",
 40            slug="test",
 41        )
 42
 43    def test_login_challenge(self):
 44        """Test login_challenge"""
 45        ui_login_button = self.source.ui_login_button(None)
 46        self.assertTrue(ui_login_button.challenge.is_valid(raise_exception=True))
 47
 48    def test_get_user_info(self):
 49        """Test get_user_info"""
 50        token = generate_key()
 51        api = PlexAuth(self.source, token)
 52        with Mocker() as mocker:
 53            mocker.get("https://plex.tv/api/v2/user", json=USER_INFO_RESPONSE)
 54            self.assertEqual(
 55                api.get_user_info(),
 56                (
 57                    USER_INFO_RESPONSE,
 58                    1234123419,
 59                ),
 60            )
 61
 62    def test_check_server_overlap(self):
 63        """Test check_server_overlap"""
 64        token = generate_key()
 65        api = PlexAuth(self.source, token)
 66        with Mocker() as mocker:
 67            mocker.get("https://plex.tv/api/v2/resources", json=RESOURCES_RESPONSE)
 68            self.assertFalse(api.check_server_overlap())
 69        self.source.allowed_servers = ["allowed"]
 70        self.source.save()
 71        with Mocker() as mocker:
 72            mocker.get("https://plex.tv/api/v2/resources", json=RESOURCES_RESPONSE)
 73            self.assertTrue(api.check_server_overlap())
 74
 75    def test_check_task(self):
 76        """Test token check task"""
 77        with Mocker() as mocker:
 78            mocker.get("https://plex.tv/api/v2/user", json=USER_INFO_RESPONSE)
 79            check_plex_token.send(self.source.pk)
 80            self.assertFalse(Event.objects.filter(action=EventAction.CONFIGURATION_ERROR).exists())
 81        with Mocker() as mocker:
 82            mocker.get("https://plex.tv/api/v2/user", exc=RequestException())
 83            check_plex_token.send(self.source.pk)
 84            self.assertTrue(Event.objects.filter(action=EventAction.CONFIGURATION_ERROR).exists())
 85
 86    def test_user_base_properties(self):
 87        """Test user base properties"""
 88        properties = self.source.get_base_user_properties(info=USER_INFO_RESPONSE)
 89        self.assertEqual(
 90            properties,
 91            {
 92                "username": "username",
 93                "name": "title",
 94                "email": "foo@bar.baz",
 95            },
 96        )
 97
 98    def test_group_base_properties(self):
 99        """Test group base properties"""
100        for group_id in ["group 1", "group 2"]:
101            properties = self.source.get_base_group_properties(group_id=group_id)
102            self.assertEqual(properties, {"name": group_id})
USER_INFO_RESPONSE = {'id': 1234123419, 'uuid': 'qwerqewrqewrqwr', 'username': 'username', 'title': 'title', 'email': 'foo@bar.baz'}
RESOURCES_RESPONSE = [{'name': 'foo', 'clientIdentifier': 'allowed', 'provides': 'server'}, {'name': 'foo', 'clientIdentifier': 'denied', 'provides': 'server'}]
class TestPlexSource(django.test.testcases.TestCase):
 35class TestPlexSource(TestCase):
 36    """plex Source tests"""
 37
 38    def setUp(self):
 39        self.source: PlexSource = PlexSource.objects.create(
 40            name="test",
 41            slug="test",
 42        )
 43
 44    def test_login_challenge(self):
 45        """Test login_challenge"""
 46        ui_login_button = self.source.ui_login_button(None)
 47        self.assertTrue(ui_login_button.challenge.is_valid(raise_exception=True))
 48
 49    def test_get_user_info(self):
 50        """Test get_user_info"""
 51        token = generate_key()
 52        api = PlexAuth(self.source, token)
 53        with Mocker() as mocker:
 54            mocker.get("https://plex.tv/api/v2/user", json=USER_INFO_RESPONSE)
 55            self.assertEqual(
 56                api.get_user_info(),
 57                (
 58                    USER_INFO_RESPONSE,
 59                    1234123419,
 60                ),
 61            )
 62
 63    def test_check_server_overlap(self):
 64        """Test check_server_overlap"""
 65        token = generate_key()
 66        api = PlexAuth(self.source, token)
 67        with Mocker() as mocker:
 68            mocker.get("https://plex.tv/api/v2/resources", json=RESOURCES_RESPONSE)
 69            self.assertFalse(api.check_server_overlap())
 70        self.source.allowed_servers = ["allowed"]
 71        self.source.save()
 72        with Mocker() as mocker:
 73            mocker.get("https://plex.tv/api/v2/resources", json=RESOURCES_RESPONSE)
 74            self.assertTrue(api.check_server_overlap())
 75
 76    def test_check_task(self):
 77        """Test token check task"""
 78        with Mocker() as mocker:
 79            mocker.get("https://plex.tv/api/v2/user", json=USER_INFO_RESPONSE)
 80            check_plex_token.send(self.source.pk)
 81            self.assertFalse(Event.objects.filter(action=EventAction.CONFIGURATION_ERROR).exists())
 82        with Mocker() as mocker:
 83            mocker.get("https://plex.tv/api/v2/user", exc=RequestException())
 84            check_plex_token.send(self.source.pk)
 85            self.assertTrue(Event.objects.filter(action=EventAction.CONFIGURATION_ERROR).exists())
 86
 87    def test_user_base_properties(self):
 88        """Test user base properties"""
 89        properties = self.source.get_base_user_properties(info=USER_INFO_RESPONSE)
 90        self.assertEqual(
 91            properties,
 92            {
 93                "username": "username",
 94                "name": "title",
 95                "email": "foo@bar.baz",
 96            },
 97        )
 98
 99    def test_group_base_properties(self):
100        """Test group base properties"""
101        for group_id in ["group 1", "group 2"]:
102            properties = self.source.get_base_group_properties(group_id=group_id)
103            self.assertEqual(properties, {"name": group_id})

plex Source tests

def setUp(self):
38    def setUp(self):
39        self.source: PlexSource = PlexSource.objects.create(
40            name="test",
41            slug="test",
42        )

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

def test_login_challenge(self):
44    def test_login_challenge(self):
45        """Test login_challenge"""
46        ui_login_button = self.source.ui_login_button(None)
47        self.assertTrue(ui_login_button.challenge.is_valid(raise_exception=True))

Test login_challenge

def test_get_user_info(self):
49    def test_get_user_info(self):
50        """Test get_user_info"""
51        token = generate_key()
52        api = PlexAuth(self.source, token)
53        with Mocker() as mocker:
54            mocker.get("https://plex.tv/api/v2/user", json=USER_INFO_RESPONSE)
55            self.assertEqual(
56                api.get_user_info(),
57                (
58                    USER_INFO_RESPONSE,
59                    1234123419,
60                ),
61            )

Test get_user_info

def test_check_server_overlap(self):
63    def test_check_server_overlap(self):
64        """Test check_server_overlap"""
65        token = generate_key()
66        api = PlexAuth(self.source, token)
67        with Mocker() as mocker:
68            mocker.get("https://plex.tv/api/v2/resources", json=RESOURCES_RESPONSE)
69            self.assertFalse(api.check_server_overlap())
70        self.source.allowed_servers = ["allowed"]
71        self.source.save()
72        with Mocker() as mocker:
73            mocker.get("https://plex.tv/api/v2/resources", json=RESOURCES_RESPONSE)
74            self.assertTrue(api.check_server_overlap())

Test check_server_overlap

def test_check_task(self):
76    def test_check_task(self):
77        """Test token check task"""
78        with Mocker() as mocker:
79            mocker.get("https://plex.tv/api/v2/user", json=USER_INFO_RESPONSE)
80            check_plex_token.send(self.source.pk)
81            self.assertFalse(Event.objects.filter(action=EventAction.CONFIGURATION_ERROR).exists())
82        with Mocker() as mocker:
83            mocker.get("https://plex.tv/api/v2/user", exc=RequestException())
84            check_plex_token.send(self.source.pk)
85            self.assertTrue(Event.objects.filter(action=EventAction.CONFIGURATION_ERROR).exists())

Test token check task

def test_user_base_properties(self):
87    def test_user_base_properties(self):
88        """Test user base properties"""
89        properties = self.source.get_base_user_properties(info=USER_INFO_RESPONSE)
90        self.assertEqual(
91            properties,
92            {
93                "username": "username",
94                "name": "title",
95                "email": "foo@bar.baz",
96            },
97        )

Test user base properties

def test_group_base_properties(self):
 99    def test_group_base_properties(self):
100        """Test group base properties"""
101        for group_id in ["group 1", "group 2"]:
102            properties = self.source.get_base_group_properties(group_id=group_id)
103            self.assertEqual(properties, {"name": group_id})

Test group base properties