authentik.outposts.tests.test_controller_docker

  1"""Docker controller tests"""
  2
  3from django.test import TestCase
  4from docker.models.containers import Container
  5
  6from authentik.blueprints.tests import reconcile_app
  7from authentik.outposts.apps import MANAGED_OUTPOST
  8from authentik.outposts.controllers.base import ControllerException
  9from authentik.outposts.controllers.docker import DockerController
 10from authentik.outposts.models import DockerServiceConnection, Outpost, OutpostType
 11from authentik.providers.proxy.controllers.docker import ProxyDockerController
 12
 13
 14class DockerControllerTests(TestCase):
 15    """Docker controller tests"""
 16
 17    @reconcile_app("authentik_outposts")
 18    def setUp(self) -> None:
 19        self.outpost = Outpost.objects.create(
 20            name="test",
 21            type=OutpostType.PROXY,
 22        )
 23        self.integration = DockerServiceConnection(name="test")
 24
 25    def test_init_managed(self):
 26        """Docker controller shouldn't do anything for managed outpost"""
 27        controller = DockerController(
 28            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(), self.integration
 29        )
 30        self.assertIsNone(controller.up())
 31        self.assertIsNone(controller.down())
 32
 33    def test_init_invalid(self):
 34        """Ensure init fails with invalid client"""
 35        with self.assertRaises(ControllerException):
 36            DockerController(self.outpost, self.integration)
 37
 38    def test_env_valid(self):
 39        """Test environment check"""
 40        controller = DockerController(
 41            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(), self.integration
 42        )
 43        env = [f"{key}={value}" for key, value in controller._get_env().items()]
 44        container = Container(attrs={"Config": {"Env": env}})
 45        self.assertFalse(controller._comp_env(container))
 46
 47    def test_env_invalid(self):
 48        """Test environment check"""
 49        controller = DockerController(
 50            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(), self.integration
 51        )
 52        container = Container(attrs={"Config": {"Env": []}})
 53        self.assertTrue(controller._comp_env(container))
 54
 55    def test_label_valid(self):
 56        """Test label check"""
 57        controller = DockerController(
 58            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(), self.integration
 59        )
 60        container = Container(attrs={"Config": {"Labels": controller._get_labels()}})
 61        self.assertFalse(controller._comp_labels(container))
 62
 63    def test_label_invalid(self):
 64        """Test label check"""
 65        controller = DockerController(
 66            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(), self.integration
 67        )
 68        container = Container(attrs={"Config": {"Labels": {}}})
 69        self.assertTrue(controller._comp_labels(container))
 70        container = Container(attrs={"Config": {"Labels": {"io.goauthentik.outpost-uuid": "foo"}}})
 71        self.assertTrue(controller._comp_labels(container))
 72
 73    def test_port_valid(self):
 74        """Test port check"""
 75        controller = ProxyDockerController(
 76            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(), self.integration
 77        )
 78        container = Container(
 79            attrs={
 80                "NetworkSettings": {
 81                    "Ports": {
 82                        "9000/tcp": [{"HostIp": "", "HostPort": "9000"}],
 83                        "9443/tcp": [{"HostIp": "", "HostPort": "9443"}],
 84                    }
 85                },
 86                "State": "",
 87            }
 88        )
 89        with self.settings(TEST=False):
 90            self.assertFalse(controller._comp_ports(container))
 91            container.attrs["State"] = "running"
 92            self.assertFalse(controller._comp_ports(container))
 93
 94    def test_port_invalid(self):
 95        """Test port check"""
 96        controller = ProxyDockerController(
 97            Outpost.objects.filter(managed=MANAGED_OUTPOST).first(), self.integration
 98        )
 99        container_no_ports = Container(
100            attrs={"NetworkSettings": {"Ports": None}, "State": "running"}
101        )
102        container_missing_port = Container(
103            attrs={
104                "NetworkSettings": {
105                    "Ports": {
106                        "9443/tcp": [{"HostIp": "", "HostPort": "9443"}],
107                    }
108                },
109                "State": "running",
110            }
111        )
112        container_mismatched_host = Container(
113            attrs={
114                "NetworkSettings": {
115                    "Ports": {
116                        "9443/tcp": [{"HostIp": "", "HostPort": "123"}],
117                    }
118                },
119                "State": "running",
120            }
121        )
122        with self.settings(TEST=False):
123            self.assertFalse(controller._comp_ports(container_no_ports))
124            self.assertTrue(controller._comp_ports(container_missing_port))
125            self.assertTrue(controller._comp_ports(container_mismatched_host))