offloaded some apps.lyric.views responsibilities to new Celery depend fn in .tasks; core.celery created for celery config; CELERY_BROKER_URL added to .settings & throughout project; some lyric view IT responsibility now accordingly covered by task UT domain
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
This commit is contained in:
24
src/apps/lyric/tasks.py
Normal file
24
src/apps/lyric/tasks.py
Normal file
@@ -0,0 +1,24 @@
|
||||
import requests
|
||||
|
||||
from celery import shared_task
|
||||
from django.conf import settings
|
||||
|
||||
|
||||
@shared_task
|
||||
def send_login_email_task(email, url):
|
||||
message_body = f"Use this magic link to login to your Dashboard:\n\n{url}"
|
||||
# Send mail via Mailgun HTTP API
|
||||
response = requests.post(
|
||||
f"https://api.mailgun.net/v3/{settings.MAILGUN_DOMAIN}/messages",
|
||||
auth=("api", settings.MAILGUN_API_KEY),
|
||||
data={
|
||||
"from": "adman@howdy.earthmanrpg.com",
|
||||
"to": email,
|
||||
"subject": "A magic login link to your Dashboard",
|
||||
"text": message_body,
|
||||
}
|
||||
)
|
||||
|
||||
# Log any errors
|
||||
if response.status_code != 200:
|
||||
print(f"Mailgun API error: {response.status_code}: {response.text}")
|
||||
@@ -5,27 +5,23 @@ from unittest import mock
|
||||
from apps.lyric.models import Token
|
||||
|
||||
|
||||
|
||||
@mock.patch("apps.lyric.views.send_login_email_task.delay")
|
||||
class SendLoginEmailViewTest(TestCase):
|
||||
def test_redirects_to_home_page(self):
|
||||
def test_redirects_to_home_page(self, mock_delay):
|
||||
response = self.client.post(
|
||||
"/apps/lyric/send_login_email", data={"email": "discoman@example.com"}
|
||||
)
|
||||
self.assertRedirects(response, "/")
|
||||
|
||||
@mock.patch("apps.lyric.views.requests.post")
|
||||
def test_sends_mail_to_address_from_post(self, mock_post):
|
||||
def test_sends_mail_to_address_from_post(self, mock_delay):
|
||||
self.client.post(
|
||||
"/apps/lyric/send_login_email", data={"email": "discoman@example.com"}
|
||||
)
|
||||
|
||||
self.assertEqual(mock_post.called, True)
|
||||
data = mock_post.call_args.kwargs["data"]
|
||||
self.assertEqual(data["subject"], "A magic login link to your Dashboard")
|
||||
self.assertEqual(data["from"], "adman@howdy.earthmanrpg.com")
|
||||
self.assertEqual(data["to"], "discoman@example.com")
|
||||
self.assertEqual(mock_delay.called, True)
|
||||
self.assertEqual(mock_delay.call_args.args[0], "discoman@example.com")
|
||||
|
||||
def test_adds_success_message(self):
|
||||
def test_adds_success_message(self, mock_delay):
|
||||
response = self.client.post(
|
||||
"/apps/lyric/send_login_email",
|
||||
data={"email": "discoman@example.com"},
|
||||
@@ -39,24 +35,21 @@ class SendLoginEmailViewTest(TestCase):
|
||||
)
|
||||
self.assertEqual(message.tags, "success")
|
||||
|
||||
def test_creates_token_associated_with_email(self):
|
||||
def test_creates_token_associated_with_email(self, mock_delay):
|
||||
self.client.post(
|
||||
"/apps/lyric/send_login_email", data={"email": "discoman@example.com"}
|
||||
)
|
||||
token = Token.objects.get()
|
||||
self.assertEqual(token.email, "discoman@example.com")
|
||||
|
||||
|
||||
@mock.patch("apps.lyric.views.requests.post")
|
||||
def test_sends_link_to_login_using_token_uid(self, mock_post):
|
||||
def test_sends_link_to_login_using_token_uid(self, mock_delay):
|
||||
self.client.post(
|
||||
"/apps/lyric/send_login_email", data={"email": "discoman@example.com"}
|
||||
)
|
||||
|
||||
token = Token.objects.get()
|
||||
expected_url = f"http://testserver/apps/lyric/login?token={token.uid}"
|
||||
data = mock_post.call_args.kwargs["data"]
|
||||
self.assertIn(expected_url, data["text"])
|
||||
self.assertEqual(mock_delay.call_args.args[1], expected_url)
|
||||
|
||||
class LoginViewTest(TestCase):
|
||||
def test_redirects_to_home_page(self):
|
||||
|
||||
16
src/apps/lyric/tests/unit/test_tasks.py
Normal file
16
src/apps/lyric/tests/unit/test_tasks.py
Normal file
@@ -0,0 +1,16 @@
|
||||
from django.test import SimpleTestCase
|
||||
from unittest import mock
|
||||
|
||||
from apps.lyric.tasks import send_login_email_task
|
||||
|
||||
|
||||
class SendLoginEmailTaskTest(SimpleTestCase):
|
||||
@mock.patch("apps.lyric.tasks.requests.post")
|
||||
def test_sends_mail_via_mailgun(self, mock_post):
|
||||
send_login_email_task("discoman@example.com", "http://example.com/login?token=abc123")
|
||||
self.assertEqual(mock_post.called, True)
|
||||
data = mock_post.call_args.kwargs["data"]
|
||||
self.assertEqual(data["subject"], "A magic login link to your Dashboard")
|
||||
self.assertEqual(data["from"], "adman@howdy.earthmanrpg.com")
|
||||
self.assertEqual(data["to"], "discoman@example.com")
|
||||
self.assertIn("http://example.com/login?token=abc123", data["text"])
|
||||
@@ -1,10 +1,10 @@
|
||||
import requests
|
||||
from django.contrib import auth, messages
|
||||
from django.conf import settings
|
||||
# from django.core.mail import send_mail
|
||||
from django.shortcuts import redirect
|
||||
from django.urls import reverse
|
||||
|
||||
from .models import Token
|
||||
from .tasks import send_login_email_task
|
||||
|
||||
|
||||
def send_login_email(request):
|
||||
email = request.POST["email"]
|
||||
@@ -12,26 +12,13 @@ def send_login_email(request):
|
||||
url = request.build_absolute_uri(
|
||||
reverse("login") + "?token=" + str(token.uid),
|
||||
)
|
||||
message_body = f"Use this magic link to login to your Dashboard:\n\n{url}"
|
||||
# Send mail via Mailgun HTTP API
|
||||
response = requests.post(
|
||||
f"https://api.mailgun.net/v3/{settings.MAILGUN_DOMAIN}/messages",
|
||||
auth=("api", settings.MAILGUN_API_KEY),
|
||||
data={
|
||||
"from": "adman@howdy.earthmanrpg.com",
|
||||
"to": email,
|
||||
"subject": "A magic login link to your Dashboard",
|
||||
"text": message_body,
|
||||
},
|
||||
)
|
||||
# Log any errors
|
||||
if response.status_code != 200:
|
||||
print(f"Mailgun API error: {response.status_code}: {response.text}")
|
||||
|
||||
send_login_email_task.delay(email, url)
|
||||
messages.success(
|
||||
request,
|
||||
"Check your email!—there you'll find a magic login link. But hurry… it's only temporary!",
|
||||
)
|
||||
|
||||
return redirect("/")
|
||||
|
||||
def login(request):
|
||||
|
||||
Reference in New Issue
Block a user