diff --git a/core/collections/models.py b/core/collections/models.py index c2f2e35b..aa47e59d 100644 --- a/core/collections/models.py +++ b/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' diff --git a/core/collections/tests/tests.py b/core/collections/tests/tests.py index 72d9eef5..904cdaae 100644 --- a/core/collections/tests/tests.py +++ b/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): diff --git a/core/collections/urls.py b/core/collections/urls.py index 0c152913..3a87b6c6 100644 --- a/core/collections/urls.py +++ b/core/collections/urls.py @@ -146,6 +146,11 @@ views.CollectionVersionExpansionView.as_view(), name='collection-version-expansion-detail' ), + path( + '//expansions//resolved-repo-updates/', + views.CollectionVersionExpansionResolvedRepoUpdatesView.as_view(), + name='collection-version-expansion-resolved-repo-updates' + ), path( '//expansions//re-evaluate/', views.CollectionVersionExpansionReEvaluateView.as_view(), diff --git a/core/collections/views.py b/core/collections/views.py index 170257e1..371c75f0 100644 --- a/core/collections/views.py +++ b/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, ) diff --git a/core/integration_tests/tests_collections.py b/core/integration_tests/tests_collections.py index bee2282d..abccf2e9 100644 --- a/core/integration_tests/tests_collections.py +++ b/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)