OpenConceptLab/ocl_issues#2348 | Expansions newer repo version from resolved diff API

core/collections/models.py

@@ -1614,6 +1614,27 @@ def link_repo_versions(self):
             if version:
                 self.explicit_source_versions.add(version)
 
+    def get_resolved_repo_version_diff_with_latest_updates(self):
+        rels = [
+            'explicit_source_versions', 'evaluated_source_versions',
+            'explicit_collection_versions', 'evaluated_collection_versions'
+        ]
+        diff = {}
+
+        def update_diffs(rel):
+            for version in get(self, rel).filter():
+                if version.url in diff:
+                    continue
+                repo_url = drop_version(version.url)
+                resolved, _ = ConceptContainerModel.resolve_reference_expression(repo_url)
+                if resolved and resolved.url != version.url:
+                    diff[version.url] = resolved.url
+
+        for relation in rels:
+            update_diffs(relation)
+
+        return diff
+
 
 class ExpansionParameters:
     ACTIVE = 'activeOnly'

core/collections/tests/tests.py

@@ -1499,6 +1499,40 @@ def test_link_repo_versions(self):
         self.assertTrue(expansion.explicit_source_versions.exists())
         self.assertEqual(expansion.explicit_source_versions.first(), concept.parent)
 
+    def test_get_resolved_repo_version_diff_with_latest_updates(self):
+        collection = OrganizationCollectionFactory()
+        expansion = ExpansionFactory(collection_version=collection)
+
+        # No repo versions linked yet — diff should be empty
+        self.assertEqual(expansion.get_resolved_repo_version_diff_with_latest_updates(), {})
+
+        # HEAD must exist so resolve_reference_expression can find the latest released version
+        source_head = OrganizationSourceFactory()
+        source_v1 = OrganizationSourceFactory(
+            mnemonic=source_head.mnemonic, organization=source_head.organization, version='v1', released=True)
+        source_v2 = OrganizationSourceFactory(
+            mnemonic=source_head.mnemonic, organization=source_head.organization, version='v2', released=True)
+        expansion.explicit_source_versions.add(source_v1)
+
+        diff = expansion.get_resolved_repo_version_diff_with_latest_updates()
+
+        # source_v1 has a newer released version (source_v2), so it should appear in the diff
+        self.assertEqual(len(diff), 1)
+        self.assertIn(source_v1.url, diff)
+        self.assertEqual(diff[source_v1.url], source_v2.url)
+
+        # A source version that is already the latest released should not appear in the diff
+        source2_head = OrganizationSourceFactory()
+        source2_v1 = OrganizationSourceFactory(
+            mnemonic=source2_head.mnemonic, organization=source2_head.organization, version='v1', released=True)
+        expansion.explicit_source_versions.add(source2_v1)
+
+        diff = expansion.get_resolved_repo_version_diff_with_latest_updates()
+
+        self.assertEqual(len(diff), 1)
+        self.assertIn(source_v1.url, diff)
+        self.assertNotIn(source2_v1.url, diff)
+
 
 class ExpansionParametersTest(OCLTestCase):
     def test_apply_active_only(self):

core/collections/urls.py

@@ -146,6 +146,11 @@
         views.CollectionVersionExpansionView.as_view(),
         name='collection-version-expansion-detail'
     ),
+    path(
+        '<str:collection>/<str:version>/expansions/<str:expansion>/resolved-repo-updates/',
+        views.CollectionVersionExpansionResolvedRepoUpdatesView.as_view(),
+        name='collection-version-expansion-resolved-repo-updates'
+    ),
     path(
         '<str:collection>/<str:version>/expansions/<str:expansion>/re-evaluate/',
         views.CollectionVersionExpansionReEvaluateView.as_view(),

core/collections/views.py

@@ -905,6 +905,16 @@ def destroy(self, request, *args, **kwargs):  # pylint: disable=unused-argument
         return Response(status=status.HTTP_204_NO_CONTENT)
 
 
+class CollectionVersionExpansionResolvedRepoUpdatesView(
+    CollectionVersionExpansionBaseView, RetrieveAPIView, DestroyAPIView):
+    serializer_class = ExpansionDetailSerializer
+    permission_classes = (HasAccessToVersionedObject, )
+
+    def get(self, request, *args, **kwargs):  # pylint: disable=unused-argument
+        obj = self.get_object()
+        return Response(obj.get_resolved_repo_version_diff_with_latest_updates(), status=status.HTTP_200_OK)
+
+
 class CollectionVersionExpansionReEvaluateView(CollectionVersionExpansionBaseView, TaskMixin):
     serializer_class = TaskSerializer
     permission_classes = (CanViewConceptDictionary, )

core/integration_tests/tests_collections.py

@@ -18,6 +18,7 @@
 from core.mappings.serializers import MappingDetailSerializer, MappingListSerializer
 from core.mappings.tests.factories import MappingFactory
 from core.orgs.tests.factories import OrganizationFactory
+from core.common.constants import ACCESS_TYPE_NONE
 from core.sources.models import Source
 from core.sources.tests.factories import OrganizationSourceFactory
 from core.tasks.models import Task
@@ -3965,3 +3966,70 @@ def test_get(self):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(len(response.data), 3)
         self.assertEqual([expansion['mnemonic'] for expansion in response.data], ['e2-head', 'e1-v1', 'e1-head'])
+
+
+class CollectionVersionExpansionResolvedRepoUpdatesViewTest(OCLAPITestCase):
+    def setUp(self):
+        super().setUp()
+        self.collection = OrganizationCollectionFactory()
+        self.expansion = ExpansionFactory(collection_version=self.collection)
+        self.collection.expansion_uri = self.expansion.uri
+        self.collection.save()
+        self.token = self.collection.created_by.get_token()
+
+    def test_get_200_empty(self):
+        # No explicit repo versions linked — response should be an empty dict
+        response = self.client.get(
+            self.expansion.url + 'resolved-repo-updates/',
+            HTTP_AUTHORIZATION='Token ' + self.token,
+            format='json'
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data, {})
+
+    def test_get_200_with_updates(self):
+        # HEAD source must exist for resolve_reference_expression to find the latest released version
+        source_head = OrganizationSourceFactory()
+        source_v1 = OrganizationSourceFactory(
+            mnemonic=source_head.mnemonic, organization=source_head.organization, version='v1', released=True)
+        source_v2 = OrganizationSourceFactory(
+            mnemonic=source_head.mnemonic, organization=source_head.organization, version='v2', released=True)
+        self.expansion.explicit_source_versions.add(source_v1)
+
+        response = self.client.get(
+            self.expansion.url + 'resolved-repo-updates/',
+            HTTP_AUTHORIZATION='Token ' + self.token,
+            format='json'
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data), 1)
+        self.assertIn(source_v1.url, response.data)
+        self.assertEqual(response.data[source_v1.url], source_v2.url)
+
+    def test_get_401_unauthenticated_private_collection(self):
+        # Private collection — unauthenticated request should be denied (403 via custom permission)
+        private_collection = OrganizationCollectionFactory(public_access=ACCESS_TYPE_NONE)
+        expansion = ExpansionFactory(collection_version=private_collection)
+
+        response = self.client.get(
+            expansion.url + 'resolved-repo-updates/',
+            format='json'
+        )
+
+        self.assertEqual(response.status_code, 401)
+
+    def test_get_403_unauthorized_private_collection(self):
+        # Private collection — user not in the owning org should be denied
+        private_collection = OrganizationCollectionFactory(public_access=ACCESS_TYPE_NONE)
+        expansion = ExpansionFactory(collection_version=private_collection)
+        other_user = UserProfileFactory()
+
+        response = self.client.get(
+            expansion.url + 'resolved-repo-updates/',
+            HTTP_AUTHORIZATION='Token ' + other_user.get_token(),
+            format='json'
+        )
+
+        self.assertEqual(response.status_code, 403)