stages/email: add tests, cleanup
This commit is contained in:
Jens Langhammer
parent
3219cffb52
commit
6676e95011
|
|
@ -12,7 +12,7 @@ class EmailStageSendForm(forms.Form):
|
|||
|
||||
|
||||
class EmailStageForm(forms.ModelForm):
|
||||
"""Form to create/edit Dummy Stage"""
|
||||
"""Form to create/edit E-Mail Stage"""
|
||||
|
||||
class Meta:
|
||||
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""email stage models"""
|
||||
from django.core.mail.backends.smtp import EmailBackend
|
||||
from django.core.mail import get_connection
|
||||
from django.core.mail.backends.base import BaseEmailBackend
|
||||
from django.db import models
|
||||
from django.utils.translation import gettext as _
|
||||
|
||||
|
|
@ -27,9 +28,9 @@ class EmailStage(Stage):
|
|||
form = "passbook.stages.email.forms.EmailStageForm"
|
||||
|
||||
@property
|
||||
def backend(self) -> EmailBackend:
|
||||
def backend(self) -> BaseEmailBackend:
|
||||
"""Get fully configured EMail Backend instance"""
|
||||
return EmailBackend(
|
||||
return get_connection(
|
||||
host=self.host,
|
||||
port=self.port,
|
||||
username=self.username,
|
||||
|
|
|
|||
|
|
@ -65,9 +65,4 @@ class EmailStageView(FormView, AuthenticationStage):
|
|||
send_mails(self.executor.current_stage, message)
|
||||
# We can't call stage_ok yet, as we're still waiting
|
||||
# for the user to click the link in the email
|
||||
# return self.executor.stage_ok()
|
||||
return super().form_invalid(form)
|
||||
|
||||
# def post(self, request: HttpRequest):
|
||||
# """Just redirect to next stage"""
|
||||
# return self.executor.()
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ def send_mails(stage: EmailStage, *messages: List[EmailMultiAlternatives]):
|
|||
@CELERY_APP.task(
|
||||
bind=True, autoretry_for=(SMTPException, ConnectionError,), retry_backoff=True
|
||||
)
|
||||
# pylint: disable=unused-argument
|
||||
def _send_mail_task(self, email_stage_pk: int, message: Dict[Any, Any]):
|
||||
"""Send E-Mail according to EmailStage parameters from background worker.
|
||||
Automatically retries if message couldn't be sent."""
|
||||
|
|
@ -38,6 +39,4 @@ def _send_mail_task(self, email_stage_pk: int, message: Dict[Any, Any]):
|
|||
setattr(message_object, key, value)
|
||||
message_object.from_email = stage.from_address
|
||||
LOGGER.debug("Sending mail", to=message_object.to)
|
||||
num_sent = stage.backend.send_messages([message_object])
|
||||
if num_sent != 1:
|
||||
raise self.retry()
|
||||
stage.backend.send_messages([message_object])
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@
|
|||
<head>
|
||||
<meta name="viewport" content="width=device-width" />
|
||||
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
|
||||
<title>Simple Transactional Email</title>
|
||||
<title></title>
|
||||
<style>{% inline_static_ascii "stages/email/css/base.css" %}</style>
|
||||
</head>
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,5 @@
|
|||
"""passbook core inlining template tags"""
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from django import template
|
||||
from django.contrib.staticfiles import finders
|
||||
|
|
@ -10,21 +8,17 @@ register = template.Library()
|
|||
|
||||
|
||||
@register.simple_tag()
|
||||
def inline_static_ascii(path: str) -> Optional[str]:
|
||||
def inline_static_ascii(path: str) -> str:
|
||||
"""Inline static asset. Doesn't check file contents, plain text is assumed"""
|
||||
result = finders.find(path)
|
||||
if os.path.exists(result):
|
||||
with open(result) as _file:
|
||||
return _file.read()
|
||||
return None
|
||||
with open(result) as _file:
|
||||
return _file.read()
|
||||
|
||||
|
||||
@register.simple_tag()
|
||||
def inline_static_binary(path: str) -> Optional[str]:
|
||||
def inline_static_binary(path: str) -> str:
|
||||
"""Inline static asset. Uses file extension for base64 block"""
|
||||
result = finders.find(path)
|
||||
suffix = Path(path).suffix
|
||||
if os.path.exists(result):
|
||||
with open(result) as _file:
|
||||
return f"data:image/{suffix};base64," + _file.read()
|
||||
return None
|
||||
with open(result) as _file:
|
||||
return f"data:image/{suffix};base64," + _file.read()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,88 @@
|
|||
"""email tests"""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.core import mail
|
||||
from django.shortcuts import reverse
|
||||
from django.test import Client, TestCase
|
||||
|
||||
from passbook.core.models import Nonce, User
|
||||
from passbook.flows.models import Flow, FlowDesignation, FlowStageBinding
|
||||
from passbook.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
|
||||
from passbook.flows.views import SESSION_KEY_PLAN
|
||||
from passbook.stages.email.models import EmailStage
|
||||
from passbook.stages.email.stage import QS_KEY_TOKEN
|
||||
|
||||
|
||||
class TestEmailStage(TestCase):
|
||||
"""Email tests"""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.user = User.objects.create_user(
|
||||
username="unittest", email="test@beryju.org"
|
||||
)
|
||||
self.client = Client()
|
||||
|
||||
self.flow = Flow.objects.create(
|
||||
name="test-email",
|
||||
slug="test-email",
|
||||
designation=FlowDesignation.AUTHENTICATION,
|
||||
)
|
||||
self.stage = EmailStage.objects.create(name="email",)
|
||||
FlowStageBinding.objects.create(flow=self.flow, stage=self.stage, order=2)
|
||||
|
||||
def test_rendering(self):
|
||||
"""Test with pending user"""
|
||||
plan = FlowPlan(flow_pk=self.flow.pk.hex, stages=[self.stage])
|
||||
plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
|
||||
session = self.client.session
|
||||
session[SESSION_KEY_PLAN] = plan
|
||||
session.save()
|
||||
|
||||
url = reverse(
|
||||
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
|
||||
)
|
||||
response = self.client.get(url)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_pending_user(self):
|
||||
"""Test with pending user"""
|
||||
plan = FlowPlan(flow_pk=self.flow.pk.hex, stages=[self.stage])
|
||||
plan.context[PLAN_CONTEXT_PENDING_USER] = self.user
|
||||
session = self.client.session
|
||||
session[SESSION_KEY_PLAN] = plan
|
||||
session.save()
|
||||
|
||||
url = reverse(
|
||||
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
|
||||
)
|
||||
with self.settings(
|
||||
EMAIL_BACKEND="django.core.mail.backends.locmem.EmailBackend"
|
||||
):
|
||||
response = self.client.post(url)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(len(mail.outbox), 1)
|
||||
self.assertEqual(mail.outbox[0].subject, "passbook - Password Recovery")
|
||||
|
||||
def test_token(self):
|
||||
"""Test with token"""
|
||||
# Make sure token exists
|
||||
self.test_pending_user()
|
||||
plan = FlowPlan(flow_pk=self.flow.pk.hex, stages=[self.stage])
|
||||
session = self.client.session
|
||||
session[SESSION_KEY_PLAN] = plan
|
||||
session.save()
|
||||
|
||||
with patch("passbook.flows.views.FlowExecutorView.cancel", MagicMock()):
|
||||
url = reverse(
|
||||
"passbook_flows:flow-executor", kwargs={"flow_slug": self.flow.slug}
|
||||
)
|
||||
token = Nonce.objects.get(user=self.user)
|
||||
url += f"?{QS_KEY_TOKEN}={token.pk.hex}"
|
||||
response = self.client.get(url)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertEqual(response.url, reverse("passbook_core:overview"))
|
||||
|
||||
session = self.client.session
|
||||
plan: FlowPlan = session[SESSION_KEY_PLAN]
|
||||
self.assertEqual(plan.context[PLAN_CONTEXT_PENDING_USER], self.user)
|
||||
Reference in New Issue