Merge branch 'master' of github.com:openedx/edx-enterprise into ENT-9…

…200/new-role-in-lms

CHANGELOG.rst

@@ -17,9 +17,13 @@ Unreleased
 ----------
 * nothing unreleased
 
+[4.21.10]
+----------
+* created migration to create a system-wide enterprise role named `enterprise_provisioning_admin`.
+
 [4.21.9]
 ---------
-* created migration to `update_or_create` a system-wide enterprise role named `enterprise_provisioning_admin.
+* fix: fixed search fetch crashing because of server taking too long for api request logs.
 
 [4.21.8]
 ---------

enterprise/__init__.py

@@ -2,4 +2,4 @@
 Your project description goes here.
 """
 
-__version__ = "4.21.8"
+__version__ = "4.21.9"

enterprise/roles_api.py

@@ -8,6 +8,7 @@
     ENTERPRISE_LEARNER_ROLE,
     ENTERPRISE_OPERATOR_ROLE,
     SYSTEM_ENTERPRISE_CATALOG_ADMIN_ROLE,
+    SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE,
 )
 from enterprise.models import SystemWideEnterpriseRole, SystemWideEnterpriseUserRoleAssignment
 
@@ -46,6 +47,11 @@ def catalog_admin_role():
     return get_or_create_system_wide_role(SYSTEM_ENTERPRISE_CATALOG_ADMIN_ROLE)
 
 
+def provisioning_admin_role():
+    """ Returns the provisioning admin role. """
+    return get_or_create_system_wide_role(SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE)
+
+
 def roles_by_name():
     """
     Returns a mapping of system wide roles by name.
@@ -55,6 +61,7 @@ def roles_by_name():
         ENTERPRISE_LEARNER_ROLE: learner_role(),
         ENTERPRISE_OPERATOR_ROLE: openedx_operator_role(),
         SYSTEM_ENTERPRISE_CATALOG_ADMIN_ROLE: catalog_admin_role(),
+        SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE: provisioning_admin_role(),
     }
 
 

integrated_channels/integrated_channel/admin/__init__.py

@@ -104,14 +104,10 @@ class IntegratedChannelAPIRequestLogAdmin(admin.ModelAdmin):
         "status_code",
     ]
     search_fields = [
-        "status_code",
-        "enterprise_customer__name",
-        "enterprise_customer__uuid",
-        "enterprise_customer_configuration_id",
-        "endpoint",
-        "time_taken",
-        "response_body",
-        "payload",
+        "enterprise_customer__name__icontains",
+        "enterprise_customer__uuid__iexact",
+        "enterprise_customer_configuration_id__iexact",
+        "endpoint__icontains",
     ]
     readonly_fields = [
         "status_code",
@@ -122,12 +118,25 @@ class IntegratedChannelAPIRequestLogAdmin(admin.ModelAdmin):
         "response_body",
         "payload",
     ]
+    list_filter = ('status_code',)
 
     list_per_page = 20
 
     def get_queryset(self, request):
+        """
+        Optimize queryset by selecting related 'enterprise_customer' and limiting fields.
+        """
         queryset = super().get_queryset(request)
-        return queryset.select_related('enterprise_customer')
+        return queryset.select_related('enterprise_customer').only(
+            'id',
+            'endpoint',
+            'enterprise_customer_id',
+            'time_taken',
+            'status_code',
+            'enterprise_customer__name',
+            'enterprise_customer__uuid',
+            'enterprise_customer_configuration_id'
+        )
 
     class Meta:
         model = IntegratedChannelAPIRequestLogs

tests/test_admin/test_admin.py

@@ -0,0 +1,77 @@
+"""
+Tests for the IntegratedChannelAPIRequest admin module in `edx-enterprise`.
+"""
+
+from django.contrib.admin.sites import AdminSite
+from django.db import connection
+from django.test import TestCase
+from django.test.utils import CaptureQueriesContext
+
+from integrated_channels.integrated_channel.admin import IntegratedChannelAPIRequestLogAdmin
+from integrated_channels.integrated_channel.models import IntegratedChannelAPIRequestLogs
+from test_utils import factories
+
+
+class IntegratedChannelAPIRequestLogAdminTest(TestCase):
+    """
+    Test the admin functionality for the IntegratedChannelAPIRequestLogs model.
+    """
+
+    def setUp(self):
+        """
+        Set up the test environment by creating a test admin instance and sample data.
+        """
+        self.site = AdminSite()
+        self.admin = IntegratedChannelAPIRequestLogAdmin(IntegratedChannelAPIRequestLogs, self.site)
+
+        self.enterprise_customer = factories.EnterpriseCustomerFactory(name='Test Enterprise')
+        self.log_entry = IntegratedChannelAPIRequestLogs.objects.create(
+            enterprise_customer=self.enterprise_customer,
+            enterprise_customer_configuration_id=1,
+            endpoint="http://test.com",
+            payload="test payload",
+            time_taken=1.0,
+            status_code=200,
+            response_body="test response"
+        )
+
+    def test_get_queryset_optimization(self):
+        """
+        Test that the get_queryset method optimizes query by using select_related and only selecting specified fields.
+        """
+        request = None
+
+        with CaptureQueriesContext(connection) as queries:
+            queryset = self.admin.get_queryset(request)
+
+            list(queryset)
+
+        self.assertEqual(len(queries), 1)
+
+        query = queries[0]['sql']
+        self.assertIn('integrated_channel_integratedchannelapirequestlogs', query)
+        self.assertIn('enterprise_enterprisecustomer', query)
+
+        self.assertIn('"integrated_channel_integratedchannelapirequestlogs"."id"', query)
+        self.assertIn('"integrated_channel_integratedchannelapirequestlogs"."endpoint"', query)
+        self.assertIn('"integrated_channel_integratedchannelapirequestlogs"."enterprise_customer_id"', query)
+        self.assertIn('"integrated_channel_integratedchannelapirequestlogs"."time_taken"', query)
+        self.assertIn('"integrated_channel_integratedchannelapirequestlogs"."status_code"', query)
+        self.assertIn('"enterprise_enterprisecustomer"."name"', query)
+        self.assertIn('"enterprise_enterprisecustomer"."uuid"', query)
+        self.assertIn(
+            '"integrated_channel_integratedchannelapirequestlogs"."enterprise_customer_configuration_id"', query
+        )
+
+        self.assertNotIn('payload', query)
+        self.assertNotIn('response_body', query)
+
+        log_entry = queryset.get(id=self.log_entry.id)
+        self.assertEqual(log_entry.endpoint, "http://test.com")
+        self.assertEqual(log_entry.enterprise_customer.name, "Test Enterprise")
+
+        # Verify that accessing an unselected field causes an additional query
+        with CaptureQueriesContext(connection) as queries:
+            _ = log_entry.payload
+
+        self.assertEqual(len(queries), 1, "Accessing unselected field should cause exactly one additional query")

tests/test_models.py

@@ -2368,6 +2368,22 @@ def test_get_context_applies_to_all_contexts(self):
 
         assert ['*'] == enterprise_role_assignment.get_context()
 
+    def test_get_context_applies_for_provisioning_admins(self):
+        """
+        Verify that having a PA role assignment with ``applies_to_all_contexts`` set to True gives
+        the user a context of "*".
+        """
+        enterprise_customer = factories.EnterpriseCustomerFactory(uuid='47130371-0b6d-43f5-01de-71942664de2b')
+        user = self._create_and_link_user('edx@example.com', enterprise_customer)
+
+        enterprise_role_assignment, __ = SystemWideEnterpriseUserRoleAssignment.objects.get_or_create(
+            user=user,
+            role=roles_api.provisioning_admin_role(),
+            applies_to_all_contexts=True,
+        )
+
+        assert ['*'] == enterprise_role_assignment.get_context()
+
     @ddt.data(
         {
             'role_name': ENTERPRISE_LEARNER_ROLE,

tests/test_roles_api.py

@@ -9,6 +9,7 @@
     ENTERPRISE_LEARNER_ROLE,
     ENTERPRISE_OPERATOR_ROLE,
     SYSTEM_ENTERPRISE_CATALOG_ADMIN_ROLE,
+    SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE
 )
 from enterprise.models import SystemWideEnterpriseRole
 
@@ -22,6 +23,7 @@ class TestUpdateRoleAssignmentsCommand(TestCase):
         ENTERPRISE_LEARNER_ROLE,
         ENTERPRISE_OPERATOR_ROLE,
         SYSTEM_ENTERPRISE_CATALOG_ADMIN_ROLE,
+        SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE
     )
 
     def setUp(self):