trustchain-oc1-orchestral
/
authentik
Archived
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

tests: use create_test_flow where possible (#3606) 

* use create_test_flow where possible

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

* fix and add more tests

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

* remove unused websocket stuff

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

* Revert "remove unused websocket stuff"

This reverts commit fc05f80951.

* keepdb for make test

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

* fix more

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

* add tests for notification transports

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>
This commit is contained in:
Jens L 2022-09-17 13:16:53 +02:00 committed by GitHub
parent 09795fa6fb
commit 2bd10dbdee
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 217 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -28,7 +28,7 @@ test-docker:
rm -f .env
test:
coverage run manage.py test authentik
coverage run manage.py test --keepdb authentik
coverage html
coverage report

4
authentik/core/management/commands/shell.py
View File

@ -7,7 +7,7 @@ from django.core.management.base import BaseCommand
from django.db.models import Model
from django.db.models.signals import post_save, pre_delete
from authentik import __version__
from authentik import get_full_version
from authentik.core.models import User
from authentik.events.middleware import should_log_model
from authentik.events.models import Event, EventAction
@ -18,7 +18,7 @@ BANNER_TEXT = """### authentik shell ({authentik})
node=platform.node(),
python=platform.python_version(),
arch=platform.machine(),
authentik=__version__,
authentik=get_full_version(),
)

8
authentik/core/tests/test_applications_api.py
View File

@ -5,8 +5,7 @@ from django.urls import reverse
from rest_framework.test import APITestCase
from authentik.core.models import Application
from authentik.core.tests.utils import create_test_admin_user
from authentik.flows.models import Flow
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.policies.dummy.models import DummyPolicy
from authentik.policies.models import PolicyBinding
from authentik.providers.oauth2.models import OAuth2Provider
@ -20,10 +19,7 @@ class TestApplicationsAPI(APITestCase):
self.provider = OAuth2Provider.objects.create(
name="test",
redirect_uris="http://some-other-domain",
authorization_flow=Flow.objects.create(
name="test",
slug="test",
),
authorization_flow=create_test_flow(),
)
self.allowed = Application.objects.create(
name="allowed",

15
authentik/core/tests/test_applications_views.py
View File

@ -4,8 +4,7 @@ from unittest.mock import MagicMock, patch
from django.urls import reverse
from authentik.core.models import Application
from authentik.core.tests.utils import create_test_admin_user, create_test_tenant
from authentik.flows.models import Flow, FlowDesignation
from authentik.core.tests.utils import create_test_admin_user, create_test_flow, create_test_tenant
from authentik.flows.tests import FlowTestCase
from authentik.tenants.models import Tenant
@ -21,11 +20,7 @@ class TestApplicationsViews(FlowTestCase):
def test_check_redirect(self):
"""Test redirect"""
empty_flow = Flow.objects.create(
name="foo",
slug="foo",
designation=FlowDesignation.AUTHENTICATION,
)
empty_flow = create_test_flow()
tenant: Tenant = create_test_tenant()
tenant.flow_authentication = empty_flow
tenant.save()
@ -49,11 +44,7 @@ class TestApplicationsViews(FlowTestCase):
def test_check_redirect_auth(self):
"""Test redirect"""
self.client.force_login(self.user)
empty_flow = Flow.objects.create(
name="foo",
slug="foo",
designation=FlowDesignation.AUTHENTICATION,
)
empty_flow = create_test_flow()
tenant: Tenant = create_test_tenant()
tenant.flow_authentication = empty_flow
tenant.save()

6
authentik/core/tests/test_source_flow_manager.py
View File

@ -6,7 +6,7 @@ from guardian.utils import get_anonymous_user
from authentik.core.models import SourceUserMatchingModes, User
from authentik.core.sources.flow_manager import Action
from authentik.flows.models import Flow, FlowDesignation
from authentik.core.tests.utils import create_test_flow
from authentik.lib.generators import generate_id
from authentik.lib.tests.utils import get_request
from authentik.policies.denied import AccessDeniedResponse
@ -152,9 +152,7 @@ class TestSourceFlowManager(TestCase):
"""Test error handling when a source selected flow is non-applicable due to a policy"""
self.source.user_matching_mode = SourceUserMatchingModes.USERNAME_LINK
flow = Flow.objects.create(
name="test", slug="test", title="test", designation=FlowDesignation.ENROLLMENT
)
flow = create_test_flow()
policy = ExpressionPolicy.objects.create(
name="false", expression="""ak_message("foo");return False"""
)

16
authentik/events/models.py
View File

@ -22,14 +22,20 @@ from django.utils.translation import gettext as _
from requests import RequestException
from structlog.stdlib import get_logger
from authentik import __version__
from authentik import get_full_version
from authentik.core.middleware import (
SESSION_KEY_IMPERSONATE_ORIGINAL_USER,
SESSION_KEY_IMPERSONATE_USER,
)
from authentik.core.models import ExpiringModel, Group, PropertyMapping, User
from authentik.events.geo import GEOIP_READER
from authentik.events.utils import cleanse_dict, get_user, model_to_dict, sanitize_dict
from authentik.events.utils import (
cleanse_dict,
get_user,
model_to_dict,
sanitize_dict,
sanitize_item,
)
from authentik.lib.models import DomainlessURLValidator, SerializerModel
from authentik.lib.sentry import SentryIgnoredException
from authentik.lib.utils.http import get_client_ip, get_http_session
@ -355,11 +361,13 @@ class NotificationTransport(SerializerModel):
"user_username": notification.user.username,
}
if self.webhook_mapping:
default_body = self.webhook_mapping.evaluate(
default_body = sanitize_item(
self.webhook_mapping.evaluate(
user=notification.user,
request=None,
notification=notification,
)
)
try:
response = get_http_session().post(
self.webhook_url,
@ -406,7 +414,7 @@ class NotificationTransport(SerializerModel):
"title": notification.body,
"color": "#fd4b2d",
"fields": fields,
"footer": f"authentik v{__version__}",
"footer": f"authentik {get_full_version()}",
}
],
}

22
authentik/events/tests/test_notifications.py
View File

@ -31,8 +31,8 @@ class TestEventsNotifications(TestCase):
def test_trigger_empty(self):
"""Test trigger without any policies attached"""
transport = NotificationTransport.objects.create(name="transport")
trigger = NotificationRule.objects.create(name="trigger", group=self.group)
transport = NotificationTransport.objects.create(name=generate_id())
trigger = NotificationRule.objects.create(name=generate_id(), group=self.group)
trigger.transports.add(transport)
trigger.save()
@ -43,8 +43,8 @@ class TestEventsNotifications(TestCase):
def test_trigger_single(self):
"""Test simple transport triggering"""
transport = NotificationTransport.objects.create(name="transport")
trigger = NotificationRule.objects.create(name="trigger", group=self.group)
transport = NotificationTransport.objects.create(name=generate_id())
trigger = NotificationRule.objects.create(name=generate_id(), group=self.group)
trigger.transports.add(transport)
trigger.save()
matcher = EventMatcherPolicy.objects.create(
@ -59,7 +59,7 @@ class TestEventsNotifications(TestCase):
def test_trigger_no_group(self):
"""Test trigger without group"""
trigger = NotificationRule.objects.create(name="trigger")
trigger = NotificationRule.objects.create(name=generate_id())
matcher = EventMatcherPolicy.objects.create(
name="matcher", action=EventAction.CUSTOM_PREFIX
)
@ -72,9 +72,9 @@ class TestEventsNotifications(TestCase):
def test_policy_error_recursive(self):
"""Test Policy error which would cause recursion"""
transport = NotificationTransport.objects.create(name="transport")
transport = NotificationTransport.objects.create(name=generate_id())
NotificationRule.objects.filter(name__startswith="default").delete()
trigger = NotificationRule.objects.create(name="trigger", group=self.group)
trigger = NotificationRule.objects.create(name=generate_id(), group=self.group)
trigger.transports.add(transport)
trigger.save()
matcher = EventMatcherPolicy.objects.create(
@ -95,9 +95,9 @@ class TestEventsNotifications(TestCase):
self.group.users.add(user2)
self.group.save()
transport = NotificationTransport.objects.create(name="transport", send_once=True)
transport = NotificationTransport.objects.create(name=generate_id(), send_once=True)
NotificationRule.objects.filter(name__startswith="default").delete()
trigger = NotificationRule.objects.create(name="trigger", group=self.group)
trigger = NotificationRule.objects.create(name=generate_id(), group=self.group)
trigger.transports.add(transport)
trigger.save()
matcher = EventMatcherPolicy.objects.create(
@ -118,10 +118,10 @@ class TestEventsNotifications(TestCase):
)
transport = NotificationTransport.objects.create(
name="transport", webhook_mapping=mapping, mode=TransportMode.LOCAL
name=generate_id(), webhook_mapping=mapping, mode=TransportMode.LOCAL
)
NotificationRule.objects.filter(name__startswith="default").delete()
trigger = NotificationRule.objects.create(name="trigger", group=self.group)
trigger = NotificationRule.objects.create(name=generate_id(), group=self.group)
trigger.transports.add(transport)
matcher = EventMatcherPolicy.objects.create(
name="matcher", action=EventAction.CUSTOM_PREFIX

131
authentik/events/tests/test_transports.py Normal file
View File

@ -0,0 +1,131 @@
"""transport tests"""
from unittest.mock import PropertyMock, patch
from django.core import mail
from django.core.mail.backends.locmem import EmailBackend
from django.test import TestCase
from requests_mock import Mocker
from authentik import get_full_version
from authentik.core.tests.utils import create_test_admin_user
from authentik.events.models import (
Event,
Notification,
NotificationSeverity,
NotificationTransport,
NotificationWebhookMapping,
TransportMode,
)
from authentik.lib.generators import generate_id
class TestEventTransports(TestCase):
"""Test Event Transports"""
def setUp(self) -> None:
self.user = create_test_admin_user()
self.event = Event.new("foo", "testing", foo="bar,").set_user(self.user)
self.event.save()
self.notification = Notification.objects.create(
severity=NotificationSeverity.ALERT,
body="foo",
event=self.event,
user=self.user,
)
def test_transport_webhook(self):
"""Test webhook transport"""
transport: NotificationTransport = NotificationTransport.objects.create(
name=generate_id(),
mode=TransportMode.WEBHOOK,
webhook_url="http://localhost:1234/test",
)
with Mocker() as mocker:
mocker.post("http://localhost:1234/test")
transport.send(self.notification)
self.assertEqual(mocker.call_count, 1)
self.assertEqual(mocker.request_history[0].method, "POST")
self.assertJSONEqual(
mocker.request_history[0].body.decode(),
{
"body": "foo",
"severity": "alert",
"user_email": self.user.email,
"user_username": self.user.username,
},
)
def test_transport_webhook_mapping(self):
"""Test webhook transport with custom mapping"""
mapping = NotificationWebhookMapping.objects.create(
name=generate_id(), expression="return request.user"
)
transport: NotificationTransport = NotificationTransport.objects.create(
name=generate_id(),
mode=TransportMode.WEBHOOK,
webhook_url="http://localhost:1234/test",
webhook_mapping=mapping,
)
with Mocker() as mocker:
mocker.post("http://localhost:1234/test")
transport.send(self.notification)
self.assertEqual(mocker.call_count, 1)
self.assertEqual(mocker.request_history[0].method, "POST")
self.assertJSONEqual(
mocker.request_history[0].body.decode(),
{"email": self.user.email, "pk": self.user.pk, "username": self.user.username},
)
def test_transport_webhook_slack(self):
"""Test webhook transport (slack)"""
transport: NotificationTransport = NotificationTransport.objects.create(
name=generate_id(),
mode=TransportMode.WEBHOOK_SLACK,
webhook_url="http://localhost:1234/test",
)
with Mocker() as mocker:
mocker.post("http://localhost:1234/test")
transport.send(self.notification)
self.assertEqual(mocker.call_count, 1)
self.assertEqual(mocker.request_history[0].method, "POST")
self.assertJSONEqual(
mocker.request_history[0].body.decode(),
{
"username": "authentik",
"icon_url": "https://goauthentik.io/img/icon.png",
"attachments": [
{
"author_name": "authentik",
"author_link": "https://goauthentik.io",
"author_icon": "https://goauthentik.io/img/icon.png",
"title": "custom_foo",
"color": "#fd4b2d",
"fields": [
{"title": "Severity", "value": "alert", "short": True},
{
"title": "Dispatched for user",
"value": self.user.username,
"short": True,
},
{"title": "foo", "value": "bar,"},
],
"footer": f"authentik {get_full_version()}",
}
],
},
)
def test_transport_email(self):
"""Test email transport"""
transport: NotificationTransport = NotificationTransport.objects.create(
name=generate_id(),
mode=TransportMode.EMAIL,
)
with patch(
"authentik.stages.email.models.EmailStage.backend_class",
PropertyMock(return_value=EmailBackend),
):
transport.send(self.notification)
self.assertEqual(len(mail.outbox), 1)
self.assertEqual(mail.outbox[0].subject, "authentik Notification: custom_foo")
self.assertIn(self.notification.body, mail.outbox[0].alternatives[0][0])

13
authentik/flows/tests/test_inspector.py
View File

@ -6,10 +6,9 @@ from django.test.client import RequestFactory
from django.urls.base import reverse
from rest_framework.test import APITestCase
from authentik.core.tests.utils import create_test_admin_user
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.challenge import ChallengeTypes
from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding, InvalidResponseAction
from authentik.lib.generators import generate_id
from authentik.flows.models import FlowDesignation, FlowStageBinding, InvalidResponseAction
from authentik.stages.dummy.models import DummyStage
from authentik.stages.identification.models import IdentificationStage, UserFields
@ -24,11 +23,7 @@ class TestFlowInspector(APITestCase):
def test(self):
"""test inspector"""
flow = Flow.objects.create(
name=generate_id(),
slug=generate_id(),
designation=FlowDesignation.AUTHENTICATION,
)
flow = create_test_flow(FlowDesignation.AUTHENTICATION)
# Stage 1 is an identification stage
ident_stage = IdentificationStage.objects.create(
@ -55,7 +50,7 @@ class TestFlowInspector(APITestCase):
"flow_info": {
"background": flow.background_url,
"cancel_url": reverse("authentik_flows:cancel"),
"title": "",
"title": flow.title,
"layout": "stacked",
},
"type": ChallengeTypes.NATIVE.value,

39
authentik/flows/tests/test_planner.py
View File

@ -8,9 +8,10 @@ from django.urls import reverse
from guardian.shortcuts import get_anonymous_user
from authentik.core.models import User
from authentik.core.tests.utils import create_test_flow
from authentik.flows.exceptions import EmptyFlowException, FlowNonApplicableException
from authentik.flows.markers import ReevaluateMarker, StageMarker
from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
from authentik.flows.models import FlowDesignation, FlowStageBinding
from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlanner, cache_key
from authentik.lib.tests.utils import dummy_get_response
from authentik.policies.dummy.models import DummyPolicy
@ -32,11 +33,7 @@ class TestFlowPlanner(TestCase):
def test_empty_plan(self):
"""Test that empty plan raises exception"""
flow = Flow.objects.create(
name="test-empty",
slug="test-empty",
designation=FlowDesignation.AUTHENTICATION,
)
flow = create_test_flow()
request = self.request_factory.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
)
@ -52,11 +49,7 @@ class TestFlowPlanner(TestCase):
)
def test_non_applicable_plan(self):
"""Test that empty plan raises exception"""
flow = Flow.objects.create(
name="test-empty",
slug="test-empty",
designation=FlowDesignation.AUTHENTICATION,
)
flow = create_test_flow()
request = self.request_factory.get(
reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
)
@ -69,11 +62,7 @@ class TestFlowPlanner(TestCase):
@patch("authentik.flows.planner.cache", CACHE_MOCK)
def test_planner_cache(self):
"""Test planner cache"""
flow = Flow.objects.create(
name="test-cache",
slug="test-cache",
designation=FlowDesignation.AUTHENTICATION,
)
flow = create_test_flow(FlowDesignation.AUTHENTICATION)
FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy"), order=0
)
@ -92,11 +81,7 @@ class TestFlowPlanner(TestCase):
def test_planner_default_context(self):
"""Test planner with default_context"""
flow = Flow.objects.create(
name="test-default-context",
slug="test-default-context",
designation=FlowDesignation.AUTHENTICATION,
)
flow = create_test_flow()
FlowStageBinding.objects.create(
target=flow, stage=DummyStage.objects.create(name="dummy"), order=0
)
@ -113,11 +98,7 @@ class TestFlowPlanner(TestCase):
def test_planner_marker_reevaluate(self):
"""Test that the planner creates the proper marker"""
flow = Flow.objects.create(
name="test-default-context",
slug="test-default-context",
designation=FlowDesignation.AUTHENTICATION,
)
flow = create_test_flow()
FlowStageBinding.objects.create(
target=flow,
@ -138,11 +119,7 @@ class TestFlowPlanner(TestCase):
def test_planner_reevaluate_actual(self):
"""Test planner with re-evaluate"""
flow = Flow.objects.create(
name="test-default-context",
slug="test-default-context",
designation=FlowDesignation.AUTHENTICATION,
)
flow = create_test_flow()
false_policy = DummyPolicy.objects.create(result=False, wait_min=1, wait_max=2)
binding = FlowStageBinding.objects.create(

6
authentik/outposts/tests/test_ws.py
View File

@ -6,7 +6,7 @@ from channels.testing import WebsocketCommunicator
from django.test import TransactionTestCase
from authentik import __version__
from authentik.flows.models import Flow, FlowDesignation
from authentik.core.tests.utils import create_test_flow
from authentik.outposts.channels import WebsocketMessage, WebsocketMessageInstruction
from authentik.outposts.models import Outpost, OutpostType
from authentik.providers.proxy.models import ProxyProvider
@ -21,9 +21,7 @@ class TestOutpostWS(TransactionTestCase):
name="test",
internal_host="http://localhost",
external_host="http://localhost",
authorization_flow=Flow.objects.create(
name="foo", slug="foo", designation=FlowDesignation.AUTHORIZATION
),
authorization_flow=create_test_flow(),
)
self.outpost: Outpost = Outpost.objects.create(
name="test",

4
authentik/root/monitoring.py
View File

@ -48,11 +48,11 @@ class ReadyView(View):
try:
db_conn = connections["default"]
_ = db_conn.cursor()
except OperationalError:
except OperationalError: # pragma: no cover
return HttpResponse(status=503)
try:
redis_conn = get_redis_connection()
redis_conn.ping()
except RedisError:
except RedisError: # pragma: no cover
return HttpResponse(status=503)
return HttpResponse(status=204)

8
authentik/root/tests.py
View File

@ -20,3 +20,11 @@ class TestRoot(TestCase):
auth_headers = {"HTTP_AUTHORIZATION": creds}
response = self.client.get(reverse("metrics"), **auth_headers)
self.assertEqual(response.status_code, 200)
def test_monitoring_live(self):
"""Test LiveView"""
self.assertEqual(self.client.get(reverse("health-live")).status_code, 204)
def test_monitoring_ready(self):
"""Test ReadyView"""
self.assertEqual(self.client.get(reverse("health-ready")).status_code, 204)

10
authentik/stages/authenticator_validate/tests/test_sms.py
View File

@ -4,8 +4,8 @@ from unittest.mock import MagicMock, patch
from django.test.client import RequestFactory
from django.urls.base import reverse
from authentik.core.tests.utils import create_test_admin_user
from authentik.flows.models import Flow, FlowStageBinding, NotConfiguredAction
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.models import FlowStageBinding, NotConfiguredAction
from authentik.flows.tests import FlowTestCase
from authentik.lib.generators import generate_id
from authentik.stages.authenticator_sms.models import AuthenticatorSMSStage, SMSDevice, SMSProviders
@ -47,7 +47,7 @@ class AuthenticatorValidateStageSMSTests(FlowTestCase):
device_classes=[DeviceClasses.SMS],
)
stage.configuration_stages.set([ident_stage])
flow = Flow.objects.create(name="test", slug="test", title="test")
flow = create_test_flow()
FlowStageBinding.objects.create(target=flow, stage=ident_stage, order=0)
FlowStageBinding.objects.create(target=flow, stage=stage, order=1)
@ -84,7 +84,7 @@ class AuthenticatorValidateStageSMSTests(FlowTestCase):
device_classes=[DeviceClasses.SMS],
)
stage.configuration_stages.set([ident_stage])
flow = Flow.objects.create(name="test", slug="test", title="test")
flow = create_test_flow()
FlowStageBinding.objects.create(target=flow, stage=ident_stage, order=0)
FlowStageBinding.objects.create(target=flow, stage=stage, order=1)
@ -140,7 +140,7 @@ class AuthenticatorValidateStageSMSTests(FlowTestCase):
device_classes=[DeviceClasses.SMS],
)
stage.configuration_stages.set([ident_stage])
flow = Flow.objects.create(name="test", slug="test", title="test")
flow = create_test_flow()
FlowStageBinding.objects.create(target=flow, stage=ident_stage, order=0)
FlowStageBinding.objects.create(target=flow, stage=stage, order=1)

6
authentik/stages/authenticator_validate/tests/test_stage.py
View File

@ -4,8 +4,8 @@ from django.test.client import RequestFactory
from django.urls.base import reverse
from rest_framework.exceptions import ValidationError
from authentik.core.tests.utils import create_test_admin_user
from authentik.flows.models import Flow, FlowStageBinding, NotConfiguredAction
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.models import FlowStageBinding, NotConfiguredAction
from authentik.flows.stage import StageView
from authentik.flows.tests import FlowTestCase
from authentik.flows.views.executor import FlowExecutorView
@ -40,7 +40,7 @@ class AuthenticatorValidateStageTests(FlowTestCase):
not_configured_action=NotConfiguredAction.CONFIGURE,
)
stage.configuration_stages.set([conf_stage])
flow = Flow.objects.create(name="test", slug="test", title="test")
flow = create_test_flow()
FlowStageBinding.objects.create(target=flow, stage=conf_stage, order=0)
FlowStageBinding.objects.create(target=flow, stage=stage, order=1)

4
authentik/stages/authenticator_validate/tests/test_webauthn.py
View File

@ -8,7 +8,7 @@ from webauthn.helpers.base64url_to_bytes import base64url_to_bytes
from webauthn.helpers.bytes_to_base64url import bytes_to_base64url
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.models import Flow, FlowStageBinding, NotConfiguredAction
from authentik.flows.models import FlowStageBinding, NotConfiguredAction
from authentik.flows.stage import StageView
from authentik.flows.tests import FlowTestCase
from authentik.flows.views.executor import FlowExecutorView
@ -54,7 +54,7 @@ class AuthenticatorValidateStageWebAuthnTests(FlowTestCase):
)
sleep(1)
stage.configuration_stages.set([ident_stage])
flow = Flow.objects.create(name="test", slug="test", title="test")
flow = create_test_flow()
FlowStageBinding.objects.create(target=flow, stage=ident_stage, order=0)
FlowStageBinding.objects.create(target=flow, stage=stage, order=1)

3
authentik/stages/email/tasks.py
View File

@ -67,8 +67,7 @@ def send_mail(self: MonitoredTask, message: dict[Any, Any], email_stage_pk: Opti
try:
backend = stage.backend
except ValueError as exc:
# pyright: reportGeneralTypeIssues=false
LOGGER.warning(exc)
LOGGER.warning("failed to get email backend", exc=exc)
self.set_status(TaskResult(TaskResultStatus.ERROR).with_error(exc))
return
backend.open()

19
authentik/stages/identification/tests.py
View File

@ -3,7 +3,7 @@ from django.urls import reverse
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.challenge import ChallengeTypes
from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
from authentik.flows.models import FlowDesignation, FlowStageBinding
from authentik.flows.tests import FlowTestCase
from authentik.sources.oauth.models import OAuthSource
from authentik.stages.identification.models import IdentificationStage, UserFields
@ -148,12 +148,7 @@ class TestIdentificationStage(FlowTestCase):
def test_enrollment_flow(self):
"""Test that enrollment flow is linked correctly"""
flow = Flow.objects.create(
name="enroll-test",
slug="unique-enrollment-string",
title="unique-enrollment-string",
designation=FlowDesignation.ENROLLMENT,
)
flow = create_test_flow()
self.stage.enrollment_flow = flow
self.stage.save()
FlowStageBinding.objects.create(
@ -173,7 +168,7 @@ class TestIdentificationStage(FlowTestCase):
password_fields=False,
enroll_url=reverse(
"authentik_core:if-flow",
kwargs={"flow_slug": "unique-enrollment-string"},
kwargs={"flow_slug": flow.slug},
),
show_source_labels=False,
primary_action="Log in",
@ -192,11 +187,7 @@ class TestIdentificationStage(FlowTestCase):
def test_recovery_flow(self):
"""Test that recovery flow is linked correctly"""
flow = Flow.objects.create(
name="recovery-test",
slug="unique-recovery-string",
designation=FlowDesignation.RECOVERY,
)
flow = create_test_flow()
self.stage.recovery_flow = flow
self.stage.save()
FlowStageBinding.objects.create(
@ -215,7 +206,7 @@ class TestIdentificationStage(FlowTestCase):
password_fields=False,
recovery_url=reverse(
"authentik_core:if-flow",
kwargs={"flow_slug": "unique-recovery-string"},
kwargs={"flow_slug": flow.slug},
),
show_source_labels=False,
primary_action="Log in",

10
authentik/stages/prompt/tests.py
View File

@ -5,9 +5,9 @@ from django.test import RequestFactory
from django.urls import reverse
from rest_framework.exceptions import ErrorDetail, ValidationError
from authentik.core.tests.utils import create_test_admin_user
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.markers import StageMarker
from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
from authentik.flows.models import FlowStageBinding
from authentik.flows.planner import FlowPlan
from authentik.flows.tests import FlowTestCase
from authentik.flows.views.executor import SESSION_KEY_PLAN
@ -24,11 +24,7 @@ class TestPromptStage(FlowTestCase):
super().setUp()
self.user = create_test_admin_user()
self.factory = RequestFactory()
self.flow = Flow.objects.create(
name="test-prompt",
slug="test-prompt",
designation=FlowDesignation.AUTHENTICATION,
)
self.flow = create_test_flow()
username_prompt = Prompt.objects.create(
field_key="username_prompt",
label="USERNAME_LABEL",

10
authentik/stages/user_write/tests.py
View File

@ -7,9 +7,9 @@ from django.urls import reverse
from authentik.core.models import USER_ATTRIBUTE_SOURCES, Group, Source, User, UserSourceConnection
from authentik.core.sources.stage import PLAN_CONTEXT_SOURCES_CONNECTION
from authentik.core.tests.utils import create_test_admin_user
from authentik.core.tests.utils import create_test_admin_user, create_test_flow
from authentik.flows.markers import StageMarker
from authentik.flows.models import Flow, FlowDesignation, FlowStageBinding
from authentik.flows.models import FlowStageBinding
from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
from authentik.flows.tests import FlowTestCase
from authentik.flows.tests.test_executor import TO_STAGE_RESPONSE_MOCK
@ -24,11 +24,7 @@ class TestUserWriteStage(FlowTestCase):
def setUp(self):
super().setUp()
self.flow = Flow.objects.create(
name="test-write",
slug="test-write",
designation=FlowDesignation.AUTHENTICATION,
)
self.flow = create_test_flow()
self.group = Group.objects.create(name="test-group")
self.other_group = Group.objects.create(name="other-group")
self.stage = UserWriteStage.objects.create(

6
tests/integration/test_outpost_docker.py
View File

@ -10,8 +10,8 @@ from docker.models.containers import Container
from docker.types.healthcheck import Healthcheck
from authentik import __version__
from authentik.core.tests.utils import create_test_flow
from authentik.crypto.models import CertificateKeyPair
from authentik.flows.models import Flow, FlowDesignation
from authentik.outposts.controllers.docker import DockerController
from authentik.outposts.models import (
DockerServiceConnection,
@ -64,9 +64,7 @@ class OutpostDockerTests(ChannelsLiveServerTestCase):
name="test",
internal_host="http://localhost",
external_host="http://localhost",
authorization_flow=Flow.objects.create(
name="foo", slug="foo", designation=FlowDesignation.AUTHORIZATION
),
authorization_flow=create_test_flow(),
)
authentication_kp = CertificateKeyPair.objects.create(
name="docker-authentication",

6
tests/integration/test_proxy_docker.py
View File

@ -10,8 +10,8 @@ from docker.models.containers import Container
from docker.types.healthcheck import Healthcheck
from authentik import __version__
from authentik.core.tests.utils import create_test_flow
from authentik.crypto.models import CertificateKeyPair
from authentik.flows.models import Flow, FlowDesignation
from authentik.outposts.models import (
DockerServiceConnection,
Outpost,
@ -64,9 +64,7 @@ class TestProxyDocker(ChannelsLiveServerTestCase):
name="test",
internal_host="http://localhost",
external_host="http://localhost",
authorization_flow=Flow.objects.create(
name="foo", slug="foo", designation=FlowDesignation.AUTHORIZATION
),
authorization_flow=create_test_flow(),
)
authentication_kp = CertificateKeyPair.objects.create(
name="docker-authentication",