Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions component_catalog/tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,6 @@
EMAIL_HOST="localhost",
EMAIL_PORT=25,
DEFAULT_FROM_EMAIL="webmaster@localhost",
SITE_URL="server.dejacode.nexb",
)
class ComponentAPITestCase(TestCase):
def setUp(self):
Expand Down Expand Up @@ -315,6 +314,8 @@ def test_api_component_detail_endpoint(self):
self.assertContains(response, self.component1_detail_url)
self.assertEqual(str(self.component1), response.data["display_name"])
self.assertIn(self.component1_detail_url, response.data["api_url"])
expected_url = f"http://testserver{self.component1.get_absolute_url()}"
self.assertEqual(expected_url, response.data["absolute_url"])
self.assertEqual(str(self.component1.uuid), response.data["uuid"])
self.assertEqual(self.component1.name, response.data["name"])
self.assertEqual(self.component1.type, response.data["type"])
Expand Down Expand Up @@ -923,7 +924,6 @@ def test_api_component_protected_fields_as_read_only(self):
EMAIL_HOST="localhost",
EMAIL_PORT=25,
DEFAULT_FROM_EMAIL="webmaster@localhost",
SITE_URL="server.dejacode.nexb",
)
class PackageAPITestCase(MaxQueryMixin, TestCase):
def setUp(self):
Expand Down Expand Up @@ -1069,6 +1069,8 @@ def test_api_package_detail_endpoint(self):

self.assertContains(response, self.package1_detail_url)
self.assertIn(self.package1_detail_url, response.data["api_url"])
expected_url = f"http://testserver{self.package1.get_absolute_url()}"
self.assertEqual(expected_url, response.data["absolute_url"])
self.assertEqual(str(self.package1.uuid), response.data["uuid"])
self.assertEqual(self.package1.filename, response.data["filename"])
self.assertEqual(self.package1.size, response.data["size"])
Expand Down
8 changes: 4 additions & 4 deletions dje/admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -300,7 +300,7 @@ def log_addition(self, request, object, change_message=None):
history_entry = History.log_addition(request.user, object)

if ADDITION in getattr(self, "email_notification_on", []):
send_notification_email(request.user, object, ADDITION)
send_notification_email(request, object, ADDITION)

return history_entry

Expand All @@ -319,7 +319,7 @@ def log_change(self, request, object, message):
# Expending the base message with details
changes_details = getattr(request, "changes_details", {})
message += construct_changes_details_message(changes_details)
send_notification_email(request.user, object, CHANGE, message)
send_notification_email(request, object, CHANGE, message)

return history_entry

Expand Down Expand Up @@ -764,14 +764,14 @@ def delete_model(self, request, obj):
History.log_deletion(request.user, obj)
super().delete_model(request, obj)
if DELETION in self.email_notification_on:
send_notification_email(request.user, obj, DELETION)
send_notification_email(request, obj, DELETION)

def delete_queryset(self, request, queryset):
"""
Add the email notification on bulk deletion through the default django
'delete_selected' action.
"""
send_notification_email_on_queryset(request.user, queryset, DELETION)
send_notification_email_on_queryset(request, queryset, DELETION)
super().delete_queryset(request, queryset)

def get_urls(self):
Expand Down
18 changes: 11 additions & 7 deletions dje/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
from contextlib import suppress
from urllib.parse import urlparse

from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ImproperlyConfigured
from django.core.exceptions import ObjectDoesNotExist
Expand Down Expand Up @@ -177,7 +176,7 @@ def perform_create(self, serializer):

History.log_addition(user, serializer.instance)
if History.ADDITION in self.email_notification_on:
send_notification_email(user, serializer.instance, History.ADDITION)
send_notification_email(self.request, serializer.instance, History.ADDITION)

def perform_update(self, serializer):
"""Add the CHANGE History."""
Expand Down Expand Up @@ -214,7 +213,9 @@ def perform_update(self, serializer):
change_message += construct_changes_details_message(
{serializer.instance: changes_details}
)
send_notification_email(user, serializer.instance, History.CHANGE, change_message)
send_notification_email(
self.request, serializer.instance, History.CHANGE, change_message
)


class ExtraPermissionsViewSetMixin:
Expand Down Expand Up @@ -317,14 +318,17 @@ def get_fields(self):

def get_absolute_url(self, obj):
"""
Return a fully qualified URL (includes the schema and domains) of the object.
Combining the settings site URL and the get_absolute_url() method of the object.
Return the object fully qualified URL including the schema and domain.

Usage:
absolute_url = serializers.SerializerMethodField()
"""
site = settings.SITE_URL.rstrip("/")
return f"{site}{obj.get_absolute_url()}"
absolute_url = obj.get_absolute_url()

if request := self.context.get("request", None):
return request.build_absolute_uri(location=absolute_url)

return absolute_url

def apply_tabs_permission(self, fields, user):
model_tabset = get_tabset_for_model(self.Meta.model)
Expand Down
29 changes: 16 additions & 13 deletions dje/notification.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ def has_email_settings():
return False


def send_notification_email(user, instance, action, message=""):
def send_notification_email(request, instance, action, message=""):
if not has_email_settings():
return

Expand All @@ -57,6 +57,7 @@ def send_notification_email(user, instance, action, message=""):
if not recipients:
return

user = request.user
verbose_name = instance._meta.verbose_name.capitalize()
verbose_action = VERBOSE_ACTION[action]
subject = f'{verbose_action} {verbose_name}: "{instance}"'
Expand All @@ -70,22 +71,22 @@ def send_notification_email(user, instance, action, message=""):
return
body += f"\n\n{message}"

if action is not History.DELETION and settings.SITE_URL:
site_url = settings.SITE_URL.rstrip("/")
body += f"\n\n{site_url}{instance.get_admin_url()}"
if action is not History.DELETION:
absolute_url = request.build_absolute_uri(location=instance.get_admin_url())
body += f"\n\n{absolute_url}"

send_mail_task.delay(subject, body, settings.DEFAULT_FROM_EMAIL, recipients)


def send_notification_email_on_queryset(user, queryset, action, message=""):
def send_notification_email_on_queryset(request, queryset, action, message=""):
if not has_email_settings():
return

if not queryset:
return

if len(queryset) == 1:
return send_notification_email(user, queryset[0], action, message)
return send_notification_email(request, queryset[0], action, message)

first = queryset[0]
if not hasattr(first, "dataspace"):
Expand All @@ -95,6 +96,7 @@ def send_notification_email_on_queryset(user, queryset, action, message=""):
if not recipients:
return

user = request.user
verbose_name_plural = first._meta.verbose_name_plural.capitalize()
verbose_action = VERBOSE_ACTION[action]

Expand All @@ -107,9 +109,9 @@ def send_notification_email_on_queryset(user, queryset, action, message=""):
for instance in queryset:
body += f"\n- {instance}"

if action is not History.DELETION and settings.SITE_URL:
site_url = settings.SITE_URL.rstrip("/")
body += f" {site_url}{instance.get_admin_url()}"
if action is not History.DELETION:
absolute_url = request.build_absolute_uri(location=instance.get_admin_url())
body += f" {absolute_url}"

if message:
body += f"\n\n{message}"
Expand Down Expand Up @@ -158,7 +160,7 @@ def successful_mass_update(sender, action, request, queryset, modeladmin, form,
message = "Changes details:\n\n"
message += "\n\n".join(f"* {field}\nNew value: {value}" for field, value in changes)

send_notification_email_on_queryset(request.user, queryset, History.CHANGE, message)
send_notification_email_on_queryset(request, queryset, History.CHANGE, message)


def send_password_changed_email(user):
Expand All @@ -178,9 +180,9 @@ def notify_on_user_locked_out(request, username, **kwargs):
Webhook defined in the reference Dataspace when the `user_locked_out`
signal is triggered.
"""
site_url = settings.SITE_URL.rstrip("/")
access_attempt_url = reverse("admin:axes_accessattempt_changelist")
access_attempt_link = f"{site_url}{access_attempt_url}?q={username}"
access_attempt_absolute_url = request.build_absolute_uri(location=access_attempt_url)
access_attempt_link = f"{access_attempt_absolute_url}?q={username}"
user = DejacodeUser.objects.get_or_none(username=username)

subject = "[DejaCode] Login attempt on locked account requires review!"
Expand All @@ -198,7 +200,8 @@ def notify_on_user_locked_out(request, username, **kwargs):

if user:
user_list_url = reverse("admin:dje_dejacodeuser_changelist")
user_list_link = f"{site_url}{user_list_url}?q={username}"
user_list_absolute_url = request.build_absolute_uri(location=user_list_url)
user_list_link = f"{user_list_absolute_url}?q={username}"
message += (
f'"{username}" is an existing DejaCode user in Dataspace '
f'"{user.dataspace.name}": {user_list_link}\n'
Expand Down
40 changes: 28 additions & 12 deletions dje/tests/test_notification.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core import mail
from django.test import RequestFactory
from django.test import TestCase
from django.test import override_settings
from django.urls import reverse
Expand All @@ -38,6 +39,7 @@
)
class EmailNotificationTest(TestCase):
def setUp(self):
self.factory = RequestFactory()
self.dataspace = Dataspace.objects.create(name="Dataspace")
self.user = get_user_model().objects.create_superuser(
"test", "test@test.com", "t3st", self.dataspace, data_email_notification=True
Expand Down Expand Up @@ -72,65 +74,73 @@ def test_get_data_update_recipients(self):
)

def test_send_notification_email(self):
request = self.factory.get("")
request.user = self.user

# No recipient, no mail
self.user.data_email_notification = False
self.user.save()
send_notification_email(self.user, self.owner, notification.ADDITION)
send_notification_email(request, self.owner, notification.ADDITION)
self.assertEqual(len(mail.outbox), 0)

# Proper object and user, notification is sent
self.user.data_email_notification = True
self.user.save()
send_notification_email(self.user, self.owner, notification.ADDITION)
send_notification_email(request, self.owner, notification.ADDITION)
self.assertEqual(len(mail.outbox), 1)
self.assertEqual(mail.outbox[0].subject, f'Added Owner: "{self.owner}"')
self.assertIn(f"http://testserver{self.owner.get_admin_url()}", mail.outbox[0].body)

# Sending a notification on a empty string object
# Nothing is sent as it's not dataspace related
send_notification_email(self.user, "", notification.ADDITION)
send_notification_email(request, "", notification.ADDITION)
self.assertEqual(len(mail.outbox), 1)

# Sending a change notification with the 'No fields changed.' message
# Nothing is sent
send_notification_email(self.user, self.owner, notification.CHANGE, "No fields changed.")
send_notification_email(request, self.owner, notification.CHANGE, "No fields changed.")
self.assertEqual(len(mail.outbox), 1)

# Sending a change notification with a change message
send_notification_email(self.user, self.owner, notification.CHANGE, "Some changes...")
send_notification_email(request, self.owner, notification.CHANGE, "Some changes...")

self.assertEqual(len(mail.outbox), 2)

def test_send_notification_email_on_queryset(self):
request = self.factory.get("")
request.user = self.user

self.assertEqual(len(mail.outbox), 0)
queryset = Owner.objects.all()

# No recipient, no mail
self.user.data_email_notification = False
self.user.save()
send_notification_email_on_queryset(self.user, queryset, notification.CHANGE)
send_notification_email_on_queryset(request, queryset, notification.CHANGE)
self.assertEqual(len(mail.outbox), 0)

# Proper object and user, notification is sent
self.user.data_email_notification = True
self.user.save()
send_notification_email_on_queryset(self.user, queryset, notification.CHANGE)
send_notification_email_on_queryset(request, queryset, notification.CHANGE)
self.assertEqual(len(mail.outbox), 1)
self.assertEqual(len(queryset), 1)
self.assertEqual(mail.outbox[0].subject, f'Updated Owner: "{self.owner}"')
self.assertIn(f"http://testserver{self.owner.get_admin_url()}", mail.outbox[0].body)

# Using an empty queryset, nothing is sent
send_notification_email_on_queryset(self.user, [], notification.CHANGE)
send_notification_email_on_queryset(request, [], notification.CHANGE)
self.assertEqual(len(mail.outbox), 1)

# Using a queryset of non-dataspace related object (empty strings)
send_notification_email_on_queryset(self.user, ["", ""], notification.CHANGE)
send_notification_email_on_queryset(request, ["", ""], notification.CHANGE)
self.assertEqual(len(mail.outbox), 1)

Owner.objects.create(name="Organization2", dataspace=self.dataspace)
queryset = Owner.objects.all()

self.assertEqual(len(queryset), 2)
send_notification_email_on_queryset(self.user, queryset, notification.CHANGE)
send_notification_email_on_queryset(request, queryset, notification.CHANGE)
self.assertEqual(mail.outbox[1].subject, "Multiple Owners updated")
self.assertIn(str(self.owner), mail.outbox[1].body)

Expand All @@ -140,10 +150,13 @@ def test_notification_with_body_containing_unicode_chars(self):
Owner.objects.create(name="Test2 Organization", dataspace=self.dataspace)
queryset = Owner.objects.all()

request = self.factory.get("")
request.user = self.user

# Proper object and user, notification is sent
self.user.data_email_notification = True
self.user.save()
send_notification_email_on_queryset(self.user, queryset, notification.CHANGE, line)
send_notification_email_on_queryset(request, queryset, notification.CHANGE, line)

self.assertEqual(len(mail.outbox), 1)
self.assertEqual(len(queryset), 2)
Expand All @@ -164,7 +177,10 @@ def test_notification_multiline_subject(self):
instance._meta.verbose_name = multiline_name
instance.dataspace = self.dataspace

send_notification_email(self.user, instance, notification.DELETION)
request = self.factory.get("")
request.user = self.user

send_notification_email(request, instance, notification.DELETION)
self.assertEqual(len(mail.outbox), 1)
subject = mail.outbox[0].subject
self.assertTrue("zstream.h" in multiline_name)
Expand Down
3 changes: 2 additions & 1 deletion license_library/tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -282,7 +282,6 @@ def test_api_licenseannotation_get_filters(self):
EMAIL_HOST="localhost",
EMAIL_PORT=25,
DEFAULT_FROM_EMAIL="webmaster@localhost",
SITE_URL="server.dejacode.nexb",
)
class LicenseAPITestCase(MaxQueryMixin, TestCase):
def setUp(self):
Expand Down Expand Up @@ -435,6 +434,8 @@ def test_api_license_detail_endpoint(self):
self.assertContains(response, self.license1_detail_url)
self.assertEqual(str(self.license1), response.data["display_name"])
self.assertIn(self.license1_detail_url, response.data["api_url"])
expected_url = f"http://testserver{self.license1.get_absolute_url()}"
self.assertEqual(expected_url, response.data["absolute_url"])
self.assertEqual(str(self.license1.uuid), response.data["uuid"])
self.assertEqual(self.license1.name, response.data["name"])
self.assertEqual(self.license1.key, response.data["key"])
Expand Down
3 changes: 2 additions & 1 deletion organization/tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
EMAIL_HOST="localhost",
EMAIL_PORT=25,
DEFAULT_FROM_EMAIL="webmaster@localhost",
SITE_URL="server.dejacode.nexb",
)
class OwnerAPITestCase(MaxQueryMixin, TestCase):
def setUp(self):
Expand Down Expand Up @@ -174,6 +173,8 @@ def test_api_owner_detail_endpoint(self):

self.assertContains(response, self.owner1_detail_url)
self.assertIn(self.owner1_detail_url, response.data["api_url"])
expected_url = f"http://testserver{self.owner1.get_absolute_url()}"
self.assertEqual(expected_url, response.data["absolute_url"])
self.assertEqual(str(self.owner1.uuid), response.data["uuid"])
self.assertEqual(self.owner1.name, response.data["name"])
self.assertEqual(self.owner1.type, response.data["type"])
Expand Down
3 changes: 2 additions & 1 deletion reporting/tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,8 @@ def test_api_report_detail_endpoint(self):
self.assertEqual(str(self.report1.uuid), response.data["uuid"])
self.assertEqual(self.report1.column_template.name, response.data["column_template"])
self.assertEqual(self.report1.description, response.data["description"])
self.assertIn(self.report1.get_absolute_url(), response.data["absolute_url"])
expected_url = f"http://testserver{self.report1.get_absolute_url()}"
self.assertEqual(expected_url, response.data["absolute_url"])
self.assertEqual(self.report1.name, response.data["name"])
self.assertEqual(self.report1.query.content_type.model, response.data["content_type"])
self.assertEqual(self.report1.query.name, response.data["query"])
Expand Down
Loading