authentik.blueprints.tests.test_v1_tasks

  1"""Test blueprints v1 tasks"""
  2
  3from hashlib import sha512
  4from tempfile import NamedTemporaryFile, mkdtemp
  5
  6from django.test import TransactionTestCase
  7from yaml import dump
  8
  9from authentik.blueprints.models import BlueprintInstance, BlueprintInstanceStatus
 10from authentik.blueprints.v1.tasks import apply_blueprint, blueprints_discovery, blueprints_find
 11from authentik.lib.config import CONFIG
 12from authentik.lib.generators import generate_id
 13
 14TMP = mkdtemp("authentik-blueprints")
 15
 16
 17class TestBlueprintsV1Tasks(TransactionTestCase):
 18    """Test Blueprints v1 Tasks"""
 19
 20    @CONFIG.patch("blueprints_dir", TMP)
 21    def test_invalid_file_syntax(self):
 22        """Test syntactically invalid file"""
 23        with NamedTemporaryFile(suffix=".yaml", dir=TMP) as file:
 24            file.write(b"{")
 25            file.flush()
 26            blueprints = blueprints_find()
 27            self.assertEqual(blueprints, [])
 28
 29    @CONFIG.patch("blueprints_dir", TMP)
 30    def test_invalid_file_version(self):
 31        """Test invalid file"""
 32        with NamedTemporaryFile(suffix=".yaml", dir=TMP) as file:
 33            file.write(b"version: 2")
 34            file.flush()
 35            blueprints = blueprints_find()
 36            self.assertEqual(blueprints, [])
 37
 38    @CONFIG.patch("blueprints_dir", TMP)
 39    def test_valid(self):
 40        """Test valid file"""
 41        blueprint_id = generate_id()
 42        with NamedTemporaryFile(mode="w+", suffix=".yaml", dir=TMP) as file:
 43            file.write(
 44                dump(
 45                    {
 46                        "version": 1,
 47                        "entries": [],
 48                        "metadata": {
 49                            "name": blueprint_id,
 50                        },
 51                    }
 52                )
 53            )
 54            file.seek(0)
 55            file_hash = sha512(file.read().encode()).hexdigest()
 56            file.flush()
 57            blueprints_discovery.send()
 58            instance = BlueprintInstance.objects.filter(name=blueprint_id).first()
 59            self.assertEqual(instance.last_applied_hash, file_hash)
 60            self.assertEqual(
 61                instance.metadata,
 62                {
 63                    "name": blueprint_id,
 64                    "labels": {},
 65                },
 66            )
 67
 68    @CONFIG.patch("blueprints_dir", TMP)
 69    def test_valid_updated(self):
 70        """Test valid file"""
 71        BlueprintInstance.objects.filter(name="foo").delete()
 72        with NamedTemporaryFile(mode="w+", suffix=".yaml", dir=TMP) as file:
 73            file.write(
 74                dump(
 75                    {
 76                        "version": 1,
 77                        "entries": [],
 78                        "metadata": {
 79                            "name": "foo",
 80                        },
 81                    }
 82                )
 83            )
 84            file.flush()
 85            blueprints_discovery.send()
 86            blueprint = BlueprintInstance.objects.filter(name="foo").first()
 87            self.assertEqual(
 88                blueprint.last_applied_hash,
 89                (
 90                    "b86ec439b3857350714f070d2833490e736d9155d3d97b2cac13f3b352223e5a"
 91                    "1adbf8ec56fa616d46090cc4773ff9e46c4e509fde96b97de87dd21fa329ca1a"
 92                ),
 93            )
 94            self.assertEqual(blueprint.metadata, {"labels": {}, "name": "foo"})
 95            file.write(
 96                dump(
 97                    {
 98                        "version": 1,
 99                        "entries": [],
100                        "metadata": {
101                            "name": "foo",
102                            "labels": {
103                                "foo": "bar",
104                            },
105                        },
106                    }
107                )
108            )
109            file.flush()
110            blueprints_discovery.send()
111            blueprint.refresh_from_db()
112            self.assertEqual(
113                blueprint.last_applied_hash,
114                (
115                    "87b68b10131d2c9751ed308bba38f04734b9e2cdf8532ed617bc52979b063c49"
116                    "2564f33f3d20ab9d5f0fd9e6eb77a13942e060199f147789cb7afab9690e72b5"
117                ),
118            )
119            self.assertEqual(
120                blueprint.metadata,
121                {
122                    "name": "foo",
123                    "labels": {"foo": "bar"},
124                },
125            )
126
127    @CONFIG.patch("blueprints_dir", TMP)
128    def test_valid_disabled(self):
129        """Test valid file"""
130        with NamedTemporaryFile(mode="w+", suffix=".yaml", dir=TMP) as file:
131            file.write(
132                dump(
133                    {
134                        "version": 1,
135                        "entries": [],
136                    }
137                )
138            )
139            file.flush()
140            instance: BlueprintInstance = BlueprintInstance.objects.create(
141                name=generate_id(),
142                path=file.name,
143                enabled=False,
144                status=BlueprintInstanceStatus.UNKNOWN,
145            )
146            instance.refresh_from_db()
147            self.assertEqual(instance.last_applied_hash, "")
148            self.assertEqual(
149                instance.status,
150                BlueprintInstanceStatus.UNKNOWN,
151            )
152            apply_blueprint.send(instance.pk).get_result(block=True)
153            instance.refresh_from_db()
154            self.assertEqual(instance.last_applied_hash, "")
155            self.assertEqual(
156                instance.status,
157                BlueprintInstanceStatus.UNKNOWN,
158            )