authentik.endpoints.tests.test_facts

 1from rest_framework.test import APITestCase
 2
 3from authentik.endpoints.models import Connector, Device, DeviceConnection
 4from authentik.lib.generators import generate_id
 5
 6
 7class TestEndpointFacts(APITestCase):
 8
 9    def test_facts_cache_purge(self):
10        """Test that creating a snapshot for a new connection purges the facts cache"""
11        device = Device.objects.create(
12            identifier=generate_id(),
13            name=generate_id(),
14        )
15        self.assertEqual(device.cached_facts.data, {})
16        connector = Connector.objects.create(name=generate_id())
17        connection = DeviceConnection.objects.create(
18            device=device,
19            connector=connector,
20        )
21        connection.create_snapshot({"vendor": {"goauthentik.io/testing": {"foo": "bar"}}})
22        self.assertEqual(
23            device.cached_facts.data, {"vendor": {"goauthentik.io/testing": {"foo": "bar"}}}
24        )
25
26    def test_facts_merge(self):
27        """test facts merging"""
28        device = Device.objects.create(
29            identifier=generate_id(),
30            name=generate_id(),
31        )
32        connection_a = DeviceConnection.objects.create(
33            device=device,
34            connector=Connector.objects.create(name=generate_id()),
35        )
36        connection_a.create_snapshot(
37            {
38                "software": [
39                    {
40                        "name": "software-a",
41                        "version": "1.2.3.4",
42                        "source": "package",
43                    }
44                ]
45            }
46        )
47
48        connection_a = DeviceConnection.objects.create(
49            device=device,
50            connector=Connector.objects.create(name=generate_id()),
51        )
52        connection_a.create_snapshot(
53            {
54                "software": [
55                    {
56                        "name": "software-b",
57                        "version": "5.6.7.8",
58                        "source": "package",
59                    }
60                ]
61            }
62        )
63        self.assertCountEqual(
64            device.cached_facts.data["software"],
65            [
66                {
67                    "name": "software-a",
68                    "version": "1.2.3.4",
69                    "source": "package",
70                },
71                {
72                    "name": "software-b",
73                    "version": "5.6.7.8",
74                    "source": "package",
75                },
76            ],
77        )
class TestEndpointFacts(rest_framework.test.APITestCase):
 8class TestEndpointFacts(APITestCase):
 9
10    def test_facts_cache_purge(self):
11        """Test that creating a snapshot for a new connection purges the facts cache"""
12        device = Device.objects.create(
13            identifier=generate_id(),
14            name=generate_id(),
15        )
16        self.assertEqual(device.cached_facts.data, {})
17        connector = Connector.objects.create(name=generate_id())
18        connection = DeviceConnection.objects.create(
19            device=device,
20            connector=connector,
21        )
22        connection.create_snapshot({"vendor": {"goauthentik.io/testing": {"foo": "bar"}}})
23        self.assertEqual(
24            device.cached_facts.data, {"vendor": {"goauthentik.io/testing": {"foo": "bar"}}}
25        )
26
27    def test_facts_merge(self):
28        """test facts merging"""
29        device = Device.objects.create(
30            identifier=generate_id(),
31            name=generate_id(),
32        )
33        connection_a = DeviceConnection.objects.create(
34            device=device,
35            connector=Connector.objects.create(name=generate_id()),
36        )
37        connection_a.create_snapshot(
38            {
39                "software": [
40                    {
41                        "name": "software-a",
42                        "version": "1.2.3.4",
43                        "source": "package",
44                    }
45                ]
46            }
47        )
48
49        connection_a = DeviceConnection.objects.create(
50            device=device,
51            connector=Connector.objects.create(name=generate_id()),
52        )
53        connection_a.create_snapshot(
54            {
55                "software": [
56                    {
57                        "name": "software-b",
58                        "version": "5.6.7.8",
59                        "source": "package",
60                    }
61                ]
62            }
63        )
64        self.assertCountEqual(
65            device.cached_facts.data["software"],
66            [
67                {
68                    "name": "software-a",
69                    "version": "1.2.3.4",
70                    "source": "package",
71                },
72                {
73                    "name": "software-b",
74                    "version": "5.6.7.8",
75                    "source": "package",
76                },
77            ],
78        )

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_facts_cache_purge(self):
10    def test_facts_cache_purge(self):
11        """Test that creating a snapshot for a new connection purges the facts cache"""
12        device = Device.objects.create(
13            identifier=generate_id(),
14            name=generate_id(),
15        )
16        self.assertEqual(device.cached_facts.data, {})
17        connector = Connector.objects.create(name=generate_id())
18        connection = DeviceConnection.objects.create(
19            device=device,
20            connector=connector,
21        )
22        connection.create_snapshot({"vendor": {"goauthentik.io/testing": {"foo": "bar"}}})
23        self.assertEqual(
24            device.cached_facts.data, {"vendor": {"goauthentik.io/testing": {"foo": "bar"}}}
25        )

Test that creating a snapshot for a new connection purges the facts cache

def test_facts_merge(self):
27    def test_facts_merge(self):
28        """test facts merging"""
29        device = Device.objects.create(
30            identifier=generate_id(),
31            name=generate_id(),
32        )
33        connection_a = DeviceConnection.objects.create(
34            device=device,
35            connector=Connector.objects.create(name=generate_id()),
36        )
37        connection_a.create_snapshot(
38            {
39                "software": [
40                    {
41                        "name": "software-a",
42                        "version": "1.2.3.4",
43                        "source": "package",
44                    }
45                ]
46            }
47        )
48
49        connection_a = DeviceConnection.objects.create(
50            device=device,
51            connector=Connector.objects.create(name=generate_id()),
52        )
53        connection_a.create_snapshot(
54            {
55                "software": [
56                    {
57                        "name": "software-b",
58                        "version": "5.6.7.8",
59                        "source": "package",
60                    }
61                ]
62            }
63        )
64        self.assertCountEqual(
65            device.cached_facts.data["software"],
66            [
67                {
68                    "name": "software-a",
69                    "version": "1.2.3.4",
70                    "source": "package",
71                },
72                {
73                    "name": "software-b",
74                    "version": "5.6.7.8",
75                    "source": "package",
76                },
77            ],
78        )

test facts merging