mirror of
https://github.com/goauthentik/authentik.git
synced 2026-06-17 19:09:11 +03:00
tests: add mixin to launch traefik for tests requiring SSL (#22011)
Signed-off-by: Jens Langhammer <jens@goauthentik.io>
This commit is contained in:
@@ -1,10 +1,15 @@
|
||||
from os import unlink, write
|
||||
from sys import stderr
|
||||
from tempfile import mkstemp
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from channels.testing import ChannelsLiveServerTestCase
|
||||
from django.apps import apps
|
||||
from django.contrib.staticfiles.testing import StaticLiveServerTestCase
|
||||
from django.urls import reverse
|
||||
from dramatiq import get_broker
|
||||
from structlog.stdlib import get_logger
|
||||
from yaml import safe_dump
|
||||
|
||||
from authentik.core.apps import Setup
|
||||
from authentik.core.models import User
|
||||
@@ -47,6 +52,78 @@ class E2ETestMixin(DockerTestCase):
|
||||
print("::endgroup::", file=stderr)
|
||||
super().tearDown()
|
||||
|
||||
def url(self, view: str, query: dict | None = None, **kwargs) -> str:
|
||||
"""reverse `view` with `**kwargs` into full URL using live_server_url"""
|
||||
url = self.live_server_url + reverse(view, kwargs=kwargs)
|
||||
if query:
|
||||
return url + "?" + urlencode(query)
|
||||
return url
|
||||
|
||||
|
||||
class SSLLiveMixin(DockerTestCase):
|
||||
"""Mixin to provide an SSL-enabled webserver for integration/e2e tests that require it.
|
||||
|
||||
Overrides `live_server_url` and as such other all usual helper functions will return an HTTPS
|
||||
URL. Certificate is self-signed and random on each run."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self._setup_traefik()
|
||||
|
||||
def tearDown(self):
|
||||
super().tearDown()
|
||||
unlink(self._traefik_config)
|
||||
|
||||
@property
|
||||
def live_server_url(self):
|
||||
return f"https://{self.host}:{self._traefik_port}"
|
||||
|
||||
def _setup_traefik(self):
|
||||
config = {
|
||||
"http": {
|
||||
"routers": {
|
||||
"authentik": {
|
||||
"rule": "PathPrefix(`/`)",
|
||||
"entryPoints": ["websecure"],
|
||||
"service": "authentik",
|
||||
"tls": {},
|
||||
}
|
||||
},
|
||||
"services": {
|
||||
"authentik": {"loadBalancer": {"servers": [{"url": super().live_server_url}]}}
|
||||
},
|
||||
}
|
||||
}
|
||||
fd, self._traefik_config = mkstemp()
|
||||
write(fd, safe_dump(config).encode())
|
||||
traefik = self.run_container(
|
||||
image="docker.io/library/traefik:3.1",
|
||||
command=[
|
||||
"--providers.file.filename=/etc/traefik/dynamic.yml",
|
||||
"--providers.file.watch=true",
|
||||
"--entrypoints.websecure.address=:9443",
|
||||
"--log.level=DEBUG",
|
||||
"--api=true",
|
||||
"--api.dashboard=true",
|
||||
"--api.insecure=true",
|
||||
],
|
||||
ports={
|
||||
"9443": None,
|
||||
},
|
||||
volumes={
|
||||
self._traefik_config: {
|
||||
"bind": "/etc/traefik/dynamic.yml",
|
||||
}
|
||||
},
|
||||
)
|
||||
# {
|
||||
# "8443/tcp": [
|
||||
# {"HostIp": "0.0.0.0", "HostPort": "8443"},
|
||||
# {"HostIp": "::", "HostPort": "8443"},
|
||||
# ],
|
||||
# }
|
||||
self._traefik_port = traefik.ports["9443/tcp"][0]["HostPort"]
|
||||
|
||||
|
||||
class E2ETestCase(E2ETestMixin, StaticLiveServerTestCase):
|
||||
"""E2E Test case with django static live server"""
|
||||
|
||||
@@ -5,10 +5,8 @@ from json import JSONDecodeError, dumps, loads
|
||||
from pathlib import Path
|
||||
from tempfile import gettempdir
|
||||
from time import sleep
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from django.contrib.staticfiles.testing import StaticLiveServerTestCase
|
||||
from django.urls import reverse
|
||||
from docker.models.containers import Container
|
||||
from requests import RequestException
|
||||
from selenium import webdriver
|
||||
@@ -131,13 +129,6 @@ class SeleniumTestMixin(E2ETestMixin):
|
||||
f"HTML: {self.driver.page_source[:1000]}"
|
||||
) from exc
|
||||
|
||||
def url(self, view: str, query: dict | None = None, **kwargs) -> str:
|
||||
"""reverse `view` with `**kwargs` into full URL using live_server_url"""
|
||||
url = self.live_server_url + reverse(view, kwargs=kwargs)
|
||||
if query:
|
||||
return url + "?" + urlencode(query)
|
||||
return url
|
||||
|
||||
def if_user_url(self, path: str | None = None) -> str:
|
||||
"""same as self.url() but show URL in shell"""
|
||||
url = self.url("authentik_core:if-user")
|
||||
|
||||
Reference in New Issue
Block a user