This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/tests/integration/test_proxy_kubernetes.py

98 lines
3.6 KiB
Python
Raw Permalink Normal View History

"""Test Controllers"""
from typing import Optional
import pytest
import yaml
from django.test import TestCase
from structlog.stdlib import get_logger
from authentik.core.tests.utils import create_test_flow
from authentik.outposts.controllers.kubernetes import KubernetesController
2020-12-05 21:08:42 +00:00
from authentik.outposts.models import KubernetesServiceConnection, Outpost, OutpostType
from authentik.outposts.tasks import outpost_connection_discovery
from authentik.providers.proxy.controllers.k8s.ingress import IngressReconciler
2020-12-05 21:08:42 +00:00
from authentik.providers.proxy.controllers.kubernetes import ProxyKubernetesController
from authentik.providers.proxy.models import ProxyMode, ProxyProvider
LOGGER = get_logger()
class TestProxyKubernetes(TestCase):
"""Test Controllers"""
controller: Optional[KubernetesController]
def setUp(self):
# Ensure that local connection have been created
outpost_connection_discovery() # pylint: disable=no-value-for-parameter
self.controller = None
def tearDown(self) -> None:
if self.controller:
for log in self.controller.down_with_logs():
LOGGER.info(log)
return super().tearDown()
@pytest.mark.timeout(120)
def test_kubernetes_controller_static(self):
"""Test Kubernetes Controller"""
provider: ProxyProvider = ProxyProvider.objects.create(
name="test",
internal_host="http://localhost",
external_host="http://localhost",
authorization_flow=create_test_flow(),
)
service_connection = KubernetesServiceConnection.objects.first()
outpost: Outpost = Outpost.objects.create(
name="test",
type=OutpostType.PROXY,
service_connection=service_connection,
)
outpost.providers.add(provider)
outpost.save()
self.controller = ProxyKubernetesController(outpost, service_connection)
manifest = self.controller.get_static_deployment()
self.assertEqual(len(list(yaml.load_all(manifest, Loader=yaml.SafeLoader))), 4)
@pytest.mark.timeout(120)
def test_kubernetes_controller_ingress(self):
"""Test Kubernetes Controller's Ingress"""
provider: ProxyProvider = ProxyProvider.objects.create(
name="test",
internal_host="http://localhost",
external_host="https://localhost",
authorization_flow=create_test_flow(),
)
provider2: ProxyProvider = ProxyProvider.objects.create(
name="test2",
internal_host="http://otherhost",
external_host="https://otherhost",
mode=ProxyMode.FORWARD_SINGLE,
authorization_flow=create_test_flow(),
)
service_connection = KubernetesServiceConnection.objects.first()
outpost: Outpost = Outpost.objects.create(
name="test",
type=OutpostType.PROXY,
service_connection=service_connection,
)
outpost.providers.add(provider)
self.controller = ProxyKubernetesController(outpost, service_connection)
ingress_rec = IngressReconciler(self.controller)
ingress = ingress_rec.retrieve()
self.assertEqual(len(ingress.spec.rules), 1)
self.assertEqual(ingress.spec.rules[0].host, "localhost")
# add provider, check again
outpost.providers.add(provider2)
ingress = ingress_rec.retrieve()
self.assertEqual(len(ingress.spec.rules), 2)
self.assertEqual(ingress.spec.rules[0].host, "localhost")
self.assertEqual(ingress.spec.rules[1].host, "otherhost")