Remove the need to define SITE_URL for several features (#53)

Signed-off-by: tdruez <tdruez@nexb.com>

component_catalog/tests/test_api.py

@@ -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):
@@ -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"])
@@ -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):
@@ -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"])

dje/admin.py

@@ -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
 
@@ -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
 
@@ -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):

dje/api.py

@@ -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
@@ -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."""
@@ -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:
@@ -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)

dje/notification.py

@@ -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
 
@@ -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}"'
@@ -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"):
@@ -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]
 
@@ -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}"
@@ -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):
@@ -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!"
@@ -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'

dje/tests/test_notification.py

@@ -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
@@ -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
@@ -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)
 
@@ -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)
@@ -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)

license_library/tests/test_api.py

@@ -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):
@@ -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"])

organization/tests/test_api.py

@@ -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):
@@ -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"])

reporting/tests/test_api.py

@@ -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"])