authentik.endpoints.tests.test_tasks 

 1from unittest.mock import PropertyMock, patch
 2
 3from rest_framework.test import APITestCase
 4
 5from authentik.endpoints.controller import BaseController, Capabilities
 6from authentik.endpoints.models import Connector
 7from authentik.endpoints.tasks import endpoints_sync
 8from authentik.lib.generators import generate_id
 9
10
11class TestEndpointTasks(APITestCase):
12    def test_agent_sync(self):
13        class controller(BaseController):
14            def capabilities(self):
15                return [Capabilities.ENROLL_AUTOMATIC_API]
16
17            def sync_endpoints(self):
18                pass
19
20        with patch.object(Connector, "controller", PropertyMock(return_value=controller)):
21            connector = Connector.objects.create(name=generate_id())
22            self.assertEqual(len(connector.schedule_specs), 1)
23
24            endpoints_sync.send(connector.pk).get_result(block=True)
25
26    def test_agent_no_sync(self):
27        class controller(BaseController):
28            def capabilities(self):
29                return []
30
31        with patch.object(Connector, "controller", PropertyMock(return_value=controller)):
32            connector = Connector.objects.create(name=generate_id())
33            self.assertEqual(len(connector.schedule_specs), 0)
34
35            endpoints_sync.send(connector.pk).get_result(block=True)
class TestEndpointTasks(rest_framework.test.APITestCase): 
12class TestEndpointTasks(APITestCase):
13    def test_agent_sync(self):
14        class controller(BaseController):
15            def capabilities(self):
16                return [Capabilities.ENROLL_AUTOMATIC_API]
17
18            def sync_endpoints(self):
19                pass
20
21        with patch.object(Connector, "controller", PropertyMock(return_value=controller)):
22            connector = Connector.objects.create(name=generate_id())
23            self.assertEqual(len(connector.schedule_specs), 1)
24
25            endpoints_sync.send(connector.pk).get_result(block=True)
26
27    def test_agent_no_sync(self):
28        class controller(BaseController):
29            def capabilities(self):
30                return []
31
32        with patch.object(Connector, "controller", PropertyMock(return_value=controller)):
33            connector = Connector.objects.create(name=generate_id())
34            self.assertEqual(len(connector.schedule_specs), 0)
35
36            endpoints_sync.send(connector.pk).get_result(block=True)

Similar to TransactionTestCase, but use transaction.atomic() to achieve test isolation.

In most situations, TestCase should be preferred to TransactionTestCase as it allows faster execution. However, there are some situations where using TransactionTestCase might be necessary (e.g. testing some transactional behavior).

On database backends with no transaction support, TestCase behaves as TransactionTestCase.

def test_agent_sync(self): 
13    def test_agent_sync(self):
14        class controller(BaseController):
15            def capabilities(self):
16                return [Capabilities.ENROLL_AUTOMATIC_API]
17
18            def sync_endpoints(self):
19                pass
20
21        with patch.object(Connector, "controller", PropertyMock(return_value=controller)):
22            connector = Connector.objects.create(name=generate_id())
23            self.assertEqual(len(connector.schedule_specs), 1)
24
25            endpoints_sync.send(connector.pk).get_result(block=True)
def test_agent_no_sync(self): 
27    def test_agent_no_sync(self):
28        class controller(BaseController):
29            def capabilities(self):
30                return []
31
32        with patch.object(Connector, "controller", PropertyMock(return_value=controller)):
33            connector = Connector.objects.create(name=generate_id())
34            self.assertEqual(len(connector.schedule_specs), 0)
35
36            endpoints_sync.send(connector.pk).get_result(block=True)