public/django_aws_ses
7
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

updates to README.md addition of CONTRIBUTING.md work on tests and requirments-dev.txt

Browse Source
This commit is contained in:
Raymond Jessop
2025-04-18 16:31:05 -05:00
parent 3113fb953f
commit 39a972d688
8 changed files with 414 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
django_aws_ses/static/django_aws_ses/css/unsubscribe.css
View File

@@ -15,10 +15,17 @@ p {
color: #555;
margin: 10px 0;
}
a {
a, button {
color: #007bff;
text-decoration: none;
padding: 8px 16px;
margin: 5px;
border: 1px solid #007bff;
border-radius: 4px;
background: none;
cursor: pointer;
}
a:hover {
text-decoration: underline;
a:hover, button:hover {
background-color: #007bff;
color: #fff;
}

20
django_aws_ses/templates/django_aws_ses/unsubscribe.html
View File

@@ -5,13 +5,23 @@
<link rel="stylesheet" href="{% static 'django_aws_ses/css/unsubscribe.css' %}">
{% endblock %}
{% block title1 %}Unsubscribe{% endblock title1 %}
{% block title1 %}Email Subscription{% endblock title1 %}
{% block content %}
<div class="unsubscribe-container">
<h3>{{ unsubscribe_message|escape }}</h3>
<p>You have been successfully unsubscribed from our email list.</p>
<p>Changed your mind? <a href="{% url 'django_aws_ses:aws_ses_unsubscribe' uuid=uuid hash=hash %}?resubscribe=1">Re-subscribe</a></p>
<p>Return to <a href="{{ home_url|default:'/' }}">home</a>.</p>
{% if user_email %}
<h3>{{ unsubscribe_message|escape }}</h3>
<p>{{ user_email|escape }} has been {% if request.GET.resubscribe %}re-subscribed{% else %}unsubscribed{% endif %} from our email list.</p>
<p>Return to <a href="{{ home_url|default:'/' }}">home</a>.</p>
{% else %}
<h3>{{ confirmation_message|escape }}</h3>
<p>Please confirm your subscription preference for your email.</p>
<form method="post" action="">
{% csrf_token %}
<button type="submit" name="action" value="unsubscribe">Unsubscribe</button>
<button type="submit" name="action" value="resubscribe">Re-subscribe</button>
</form>
<p>Return to <a href="{{ home_url|default:'/' }}">home</a>.</p>
{% endif %}
</div>
{% endblock content %}

137
django_aws_ses/tests.py Normal file
View File

@@ -0,0 +1,137 @@
from django.test import TestCase, RequestFactory, override_settings
from django.core.mail import EmailMessage
from django.contrib.auth import get_user_model
from django.urls import reverse
from django.utils.http import urlsafe_base64_encode
from django.utils.encoding import force_bytes
from unittest.mock import patch, Mock
import json
from .backends import SESBackend
from .views import handle_bounce, HandleUnsubscribe
from .utils import filter_recipients
from .models import BounceRecord, ComplaintRecord, AwsSesUserAddon
User = get_user_model()
@override_settings(
AWS_SES_ACCESS_KEY_ID='test-key',
AWS_SES_SECRET_ACCESS_KEY='test-secret',
AWS_SES_REGION_NAME='us-east-1',
AWS_SES_REGION_ENDPOINT='email.us-east-1.amazonaws.com',
EMAIL_BACKEND='django_aws_ses.backends.SESBackend',
SES_BOUNCE_LIMIT=1,
)
class DjangoAwsSesTests(TestCase):
def setUp(self):
self.factory = RequestFactory()
self.user = User.objects.create_user(
username='testuser', email='test@example.com', password='testpass'
)
self.ses_addon = AwsSesUserAddon.objects.create(user=self.user)
@patch('boto3.client')
def test_email_sending(self, mock_boto_client):
"""Test sending an email via SESBackend."""
mock_ses = Mock()
mock_ses.send_raw_email.return_value = {
'MessageId': 'test-id',
'ResponseMetadata': {'RequestId': 'test-request-id'}
}
mock_boto_client.return_value = mock_ses
backend = SESBackend()
message = EmailMessage(
subject='Test',
body='Hello',
from_email='from@example.com',
to=['to@example.com']
)
sent, _ = backend.send_messages([message])
self.assertEqual(sent, 1)
self.assertEqual(message.extra_headers['message_id'], 'test-id')
def test_bounce_handling(self):
"""Test handling an SNS bounce notification."""
notification = {
'Type': 'Notification',
'Message': json.dumps({
'notificationType': 'Bounce',
'mail': {'destination': ['test@example.com']},
'bounce': {
'feedbackId': 'test-feedback',
'bounceType': 'Permanent',
'bounceSubType': 'General',
'bouncedRecipients': [{'emailAddress': 'test@example.com'}]
}
})
}
request = self.factory.post('/aws_ses/bounce/', data=notification, content_type='application/json')
with patch('django_aws_ses.utils.verify_bounce_message', return_value=True):
response = handle_bounce(request)
self.assertEqual(response.status_code, 200)
self.assertTrue(BounceRecord.objects.filter(email='test@example.com').exists())
def test_complaint_handling(self):
"""Test handling an SNS complaint notification."""
notification = {
'Type': 'Notification',
'Message': json.dumps({
'notificationType': 'Complaint',
'mail': {'destination': ['test@example.com']},
'complaint': {
'feedbackId': 'test-feedback',
'complaintFeedbackType': 'abuse',
'complainedRecipients': [{'emailAddress': 'test@example.com'}]
}
})
}
request = self.factory.post('/aws_ses/bounce/', data=notification, content_type='application/json')
with patch('django_aws_ses.utils.verify_bounce_message', return_value=True):
response = handle_bounce(request)
self.assertEqual(response.status_code, 200)
self.assertTrue(ComplaintRecord.objects.filter(email='test@example.com').exists())
def test_unsubscribe_confirmation(self):
"""Test unsubscribe confirmation page and action."""
uuid = urlsafe_base64_encode(force_bytes(str(self.user.pk)))
hash_value = self.ses_addon.unsubscribe_hash_generator()
url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid, 'hash': hash_value})
# Test GET (confirmation page)
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'Please confirm your subscription preference')
self.assertFalse(self.ses_addon.unsubscribe)
# Test POST (unsubscribe)
response = self.client.post(url, {'action': 'unsubscribe'})
self.ses_addon.refresh_from_db()
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'You have been unsubscribed')
self.assertTrue(self.ses_addon.unsubscribe)
def test_resubscribe_confirmation(self):
"""Test re-subscribe confirmation action."""
self.ses_addon.unsubscribe = True
self.ses_addon.save()
uuid = urlsafe_base64_encode(force_bytes(str(self.user.pk)))
hash_value = self.ses_addon.unsubscribe_hash_generator()
url = reverse('django_aws_ses:aws_ses_unsubscribe', kwargs={'uuid': uuid, 'hash': hash_value})
response = self.client.post(url, {'action': 'resubscribe'})
self.ses_addon.refresh_from_db()
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'You have been re-subscribed')
self.assertFalse(self.ses_addon.unsubscribe)
def test_recipient_filtering(self):
"""Test recipient filtering for blacklisted emails."""
BounceRecord.objects.create(email='bounce@example.com')
ComplaintRecord.objects.create(email='complaint@example.com')
recipients = ['test@example.com', 'bounce@example.com', 'complaint@example.com']
filtered = filter_recipients(recipients)
self.assertEqual(filtered, ['test@example.com'])

68
django_aws_ses/views.py
View File

@@ -13,6 +13,7 @@ from django.utils.encoding import force_bytes, force_str
from django.utils.http import urlsafe_base64_decode, urlsafe_base64_encode
from django.views.decorators.http import require_POST
from django.views.generic.base import TemplateView
from django.views.decorators.csrf import csrf_protect
from . import settings
from . import signals
@@ -287,21 +288,25 @@ def handle_bounce(request):
class HandleUnsubscribe(TemplateView):
"""View to handle email unsubscribe requests."""
http_method_names = ['get']
"""View to handle email unsubscribe and re-subscribe requests with confirmation."""
http_method_names = ['get', 'post']
template_name = settings.UNSUBSCRIBE_TEMPLATE
base_template_name = settings.BASE_TEMPLATE
unsubscribe_message = "We Have Unsubscribed the Following Email"
confirmation_message = "Please confirm your email subscription preference"
unsubscribe_message = "You have been unsubscribed"
resubscribe_message = "You have been re-subscribed"
def get_context_data(self, **kwargs):
"""Add base template and unsubscribe message to context."""
"""Add base template and appropriate message to context."""
context = super().get_context_data(**kwargs)
context['base_template_name'] = self.base_template_name
context['unsubscribe_message'] = self.unsubscribe_message
context['confirmation_message'] = self.confirmation_message
context['unsubscribe_message'] = self.resubscribe_message if self.request.GET.get('resubscribe') else self.unsubscribe_message
context['user_email'] = getattr(self, 'user_email', '')
return context
def get(self, request, *args, **kwargs):
"""Process unsubscribe request and redirect if invalid."""
"""Show confirmation page for unsubscribe or re-subscribe."""
uuid = self.kwargs['uuid']
hash_value = self.kwargs['hash']
@@ -317,12 +322,51 @@ class HandleUnsubscribe(TemplateView):
except AwsSesUserAddon.DoesNotExist:
ses = AwsSesUserAddon.objects.create(user=user)
if user and ses.check_unsubscribe_hash(hash_value):
ses.unsubscribe = True
ses.save()
logger.info(f"Unsubscribed user: {user.email}")
else:
if not user or not ses.check_unsubscribe_hash(hash_value):
logger.warning(f"Invalid unsubscribe hash for user: {user.email}")
return redirect(settings.HOME_URL)
return super().get(request, *args, **kwargs)
self.user_email = user.email
return super().get(request, *args, **kwargs)
@csrf_protect
def post(self, request, *args, **kwargs):
"""Process unsubscribe or re-subscribe request."""
uuid = self.kwargs['uuid']
hash_value = self.kwargs['hash']
action = request.POST.get('action')
try:
uuid = force_str(urlsafe_base64_decode(uuid))
user = User.objects.get(pk=uuid)
except (TypeError, ValueError, OverflowError, User.DoesNotExist) as e:
logger.warning(f"Invalid unsubscribe UUID: {e}")
return redirect(settings.HOME_URL)
try:
ses = user.aws_ses
except AwsSesUserAddon.DoesNotExist:
ses = AwsSesUserAddon.objects.create(user=user)
if not user or not ses.check_unsubscribe_hash(hash_value):
logger.warning(f"Invalid unsubscribe hash for user: {user.email}")
return redirect(settings.HOME_URL)
if action == 'unsubscribe':
ses.unsubscribe = True
ses.save()
logger.info(f"Unsubscribed user: {user.email}")
self.user_email = user.email
self.template_name = settings.UNSUBSCRIBE_TEMPLATE
return self.get(request, *args, **kwargs)
elif action == 'resubscribe':
ses.unsubscribe = False
ses.save()
logger.info(f"Re-subscribed user: {user.email}")
self.user_email = user.email
self.request.GET = request.GET.copy()
self.request.GET['resubscribe'] = '1'
return self.get(request, *args, **kwargs)
logger.warning(f"Invalid action for user: {user.email}")
return redirect(settings.HOME_URL)