OpenConceptLab/ocl_issues#1732 | Global/Org/User URL Registry List/Se…

…arch/Create APIs
OpenConceptLab · Jan 11, 2024 · cc041ed · cc041ed
1 parent bcd9dc3
commit cc041ed
Show file tree

Hide file tree

Showing 17 changed files with 722 additions and 2 deletions.
diff --git a/core/common/views.py b/core/common/views.py
@@ -77,7 +77,7 @@ def has_parent_scope(self):
         return self.has_owner_scope() and self.has_concept_container_scope()
 
     def _should_exclude_retired_from_search_results(self):
-        if self.is_owner_document_model() or 'expansion' in self.kwargs:
+        if self.is_owner_document_model() or 'expansion' in self.kwargs or self.is_url_registry_document():
             return False
 
         params = get(self, 'params') or self.request.query_params.dict()
@@ -88,7 +88,8 @@ def should_include_inactive(self):
         return self.request.query_params.get(INCLUDE_INACTIVE) in TRUTHY
 
     def _should_include_private(self):
-        return self.is_user_document() or self.request.user.is_staff or self.is_user_scope()
+        return (self.is_user_document() or self.request.user.is_staff or
+                self.is_user_scope() or self.is_url_registry_document())
 
     def is_verbose(self):
         return self.request.query_params.get(VERBOSE_PARAM, False) in TRUTHY
@@ -468,6 +469,10 @@ def is_user_document(self):
         from core.users.documents import UserProfileDocument
         return self.document_model == UserProfileDocument
 
+    def is_url_registry_document(self):
+        from core.url_registry.documents import URLRegistryDocument
+        return self.document_model == URLRegistryDocument
+
     def is_concept_document(self):
         from core.concepts.documents import ConceptDocument
         return self.document_model == ConceptDocument

diff --git a/core/integration_tests/tests_url_registries.py b/core/integration_tests/tests_url_registries.py
@@ -0,0 +1,302 @@
+from rest_framework.exceptions import ErrorDetail
+
+from core.common.tests import OCLAPITestCase
+from core.orgs.tests.factories import OrganizationFactory
+from core.url_registry.factories import GlobalURLRegistryFactory, OrganizationURLRegistryFactory, UserURLRegistryFactory
+from core.url_registry.models import URLRegistry
+from core.users.tests.factories import UserProfileFactory
+
+
+class URLRegistriesViewTest(OCLAPITestCase):
+    def test_post_global_registry(self):
+        user = UserProfileFactory()
+        response = self.client.post(
+            '/url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'url': 'https://foo.bar.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, organization__isnull=True, user__isnull=True, url='https://foo.bar.com',
+                namespace__isnull=True
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 1)
+
+        # duplicate entry
+        response = self.client.post(
+            '/url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'url': 'https://foo.bar.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(
+            response.data,
+            {'non_fields_error': [ErrorDetail(string='This entry already exists.', code='invalid')]}
+        )
+        self.assertEqual(URLRegistry.objects.count(), 1)
+
+        # entry with same namespace and url but different name is duplicate
+        response = self.client.post(
+            '/url-registry/',
+            {
+                'name': 'GlobalRegistry2',
+                'namespace': '/registry-1/',
+                'url': 'https://foo.bar.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(
+            response.data,
+            {'non_fields_error': [ErrorDetail(string='This entry already exists.', code='invalid')]}
+        )
+        self.assertEqual(URLRegistry.objects.count(), 1)
+
+        # entry with same name and namespace but different url is not duplicate
+        response = self.client.post(
+            '/url-registry/',
+            {
+                'name': 'GlobalRegistry1',
+                'namespace': '/registry-1/',
+                'url': 'https://foo.bar.1.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, organization__isnull=True, user__isnull=True, url='https://foo.bar.1.com',
+                namespace__isnull=True
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 2)
+
+    def test_post_org_registry(self):
+        org = OrganizationFactory()
+        user = org.created_by
+        response = self.client.post(
+            org.uri + 'url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'namespace': 'Foobar',  # will be set correctly
+                'url': 'https://foo.bar.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertEqual(response.data['namespace'], org.uri)
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, organization=org, url='https://foo.bar.com'
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 1)
+
+        # duplicate entry
+        response = self.client.post(
+            org.uri + 'url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'url': 'https://foo.bar.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(
+            response.data,
+            {'non_fields_error': [ErrorDetail(string='This entry already exists.', code='invalid')]}
+        )
+        self.assertEqual(URLRegistry.objects.count(), 1)
+
+        response = self.client.post(
+            org.uri + 'url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'url': 'https://foo.bar.1.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertEqual(response.data['namespace'], org.uri)
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, organization=org, url='https://foo.bar.1.com'
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 2)
+
+        response = self.client.post(
+            '/url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'namespace': org.uri,
+                'url': 'https://foo.bar.2.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertEqual(response.data['namespace'], org.uri)
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, organization=org, url='https://foo.bar.2.com'
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 3)
+
+    def test_post_user_registry(self):
+        user = UserProfileFactory()
+        response = self.client.post(
+            user.uri + 'url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'namespace': 'Foobar',  # will be set correctly
+                'url': 'https://foo.bar.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertEqual(response.data['namespace'], user.uri)
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, user=user, url='https://foo.bar.com'
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 1)
+
+        # duplicate entry
+        response = self.client.post(
+            user.uri + 'url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'url': 'https://foo.bar.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(
+            response.data,
+            {'non_fields_error': [ErrorDetail(string='This entry already exists.', code='invalid')]}
+        )
+        self.assertEqual(URLRegistry.objects.count(), 1)
+
+        response = self.client.post(
+            user.uri + 'url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'url': 'https://foo.bar.1.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertEqual(response.data['namespace'], user.uri)
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, user=user, url='https://foo.bar.1.com'
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 2)
+
+        response = self.client.post(
+            '/url-registry/',
+            {
+                'name': 'GlobalRegistry',
+                'namespace': user.uri,
+                'url': 'https://foo.bar.2.com',
+            },
+            HTTP_AUTHORIZATION=f"Token {user.get_token()}",
+            format='json',
+        )
+
+        self.assertEqual(response.status_code, 201)
+        self.assertIsNotNone(response.data['id'])
+        self.assertEqual(response.data['namespace'], user.uri)
+        self.assertTrue(
+            URLRegistry.objects.filter(
+                is_active=True, user=user, url='https://foo.bar.2.com'
+            ).exists()
+        )
+        self.assertEqual(URLRegistry.objects.count(), 3)
+
+    def test_get(self):
+        global_registry = GlobalURLRegistryFactory(name='global')
+        org_registry = OrganizationURLRegistryFactory(name='org')
+        user_registry = UserURLRegistryFactory(name='user')
+
+        response = self.client.get('/url-registry/')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]['name'], 'global')
+        self.assertEqual(response.data[0]['id'], global_registry.id)
+
+        response = self.client.get(org_registry.owner.uri + 'url-registry/')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]['name'], 'org')
+        self.assertEqual(response.data[0]['id'], org_registry.id)
+
+        response = self.client.get(user_registry.owner.uri + 'url-registry/')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]['name'], 'user')
+        self.assertEqual(response.data[0]['id'], user_registry.id)
+
+        response = self.client.get(user_registry.owner.uri + 'orgs/url-registry/')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 0)
+
+        org_registry.organization.members.add(user_registry.user)
+
+        response = self.client.get(user_registry.owner.uri + 'orgs/url-registry/')
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]['name'], 'org')
+        self.assertEqual(response.data[0]['id'], org_registry.id)
+
+        response = self.client.get(
+            '/user/orgs/url-registry/',
+            HTTP_AUTHORIZATION=f"Token {user_registry.user.get_token()}",
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]['name'], 'org')
+        self.assertEqual(response.data[0]['id'], org_registry.id)
diff --git a/core/orgs/urls.py b/core/orgs/urls.py
@@ -51,6 +51,9 @@
         views.OrganizationMemberView.as_view(),
         name='organization-member-detail'
     ),
+    re_path(
+        r'^(?P<org>' + NAMESPACE_PATTERN + ')/url-registry/', include('core.url_registry.urls'),
+        name='org-url-registry-urls'),
     re_path(r'^(?P<org>' + NAMESPACE_PATTERN + ')/repos/', include('core.repos.urls'), name='org-repos-urls'),
     re_path(r'^(?P<org>' + NAMESPACE_PATTERN + ')/sources/', include('core.sources.urls')),
     re_path(r'^(?P<org>' + NAMESPACE_PATTERN + ')/CodeSystem/', include('core.code_systems.urls'),

diff --git a/core/settings.py b/core/settings.py
@@ -100,6 +100,7 @@
     'core.tasks',
     'core.toggles',
     'core.repos',
+    'core.url_registry',
 ]
 REST_FRAMEWORK = {
     'DEFAULT_PERMISSION_CLASSES': (

diff --git a/core/url_registry/__init__.py b/core/url_registry/__init__.py
@@ -0,0 +1,4 @@
+from django.conf import settings
+
+
+__version__ = settings.VERSION