From e8e42261e3f603049c21811eb5dda82ed48a9f18 Mon Sep 17 00:00:00 2001 From: Jens Langhammer Date: Thu, 16 Jan 2025 01:02:14 +0100 Subject: [PATCH] simplify and patch enterprise Signed-off-by: Jens Langhammer --- tests/e2e/_process.py | 54 +++++++++++++++++-------------------------- 1 file changed, 21 insertions(+), 33 deletions(-) diff --git a/tests/e2e/_process.py b/tests/e2e/_process.py index a86712f7b9..5cdbfe5651 100644 --- a/tests/e2e/_process.py +++ b/tests/e2e/_process.py @@ -1,10 +1,15 @@ """authentik e2e testing utilities""" -import traceback +from datetime import timedelta +from time import mktime +from unittest.mock import MagicMock, patch -from daphne.testing import DaphneProcess, _reinstall_reactor +from daphne.testing import DaphneProcess from django import setup as django_setup from django.conf import settings +from django.utils.timezone import now + +from authentik.lib.generators import generate_id class TestDatabaseProcess(DaphneProcess): @@ -12,41 +17,24 @@ class TestDatabaseProcess(DaphneProcess): https://github.com/django/channels/issues/2048""" def run(self): - # OK, now we are in a forked child process, and want to use the reactor. - # However, FreeBSD systems like MacOS do not fork the underlying Kqueue, - # which asyncio (hence asyncioreactor) is built on. - # Therefore, we should uninstall the broken reactor and install a new one. - _reinstall_reactor() - if not settings.configured: # Fix For raise AppRegistryNotReady("Apps aren't loaded yet.") django_setup() # Ensure Django is fully set up before using settings - - from daphne.endpoints import build_endpoint_description_strings - from daphne.server import Server - from twisted.internet import reactor - - application = self.get_application() if not settings.DATABASES[list(settings.DATABASES.keys())[0]]["NAME"].startswith("test_"): for _, db_settings in settings.DATABASES.items(): db_settings["NAME"] = f"test_{db_settings['NAME']}" settings.TEST = True + from authentik.enterprise.license import LicenseKey - try: - # Create the server class - endpoints = build_endpoint_description_strings(host=self.host, port=0) - self.server = Server( - application=application, endpoints=endpoints, signal_handlers=False, **self.kwargs - ) - # Set up a poller to look for the port - reactor.callLater(0.1, self.resolve_port) - # Run with setup/teardown - if self.setup is not None: - self.setup() - try: - self.server.run() - finally: - if self.teardown is not None: - self.teardown() - except BaseException as e: - # Put the error on our queue so the parent gets it - self.errors.put((e, traceback.format_exc())) + with patch( + "authentik.enterprise.license.LicenseKey.validate", + MagicMock( + return_value=LicenseKey( + aud="", + exp=int(mktime((now() + timedelta(days=3000)).timetuple())), + name=generate_id(), + internal_users=100, + external_users=100, + ) + ), + ): + return super().run()