Skip to content

OpenConceptLab/ocl_issues#2571 | Repo HEAD can be exported by repo authors #880

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
snyaggarwal merged 1 commit into from
Jun 16, 2026
Merged

OpenConceptLab/ocl_issues#2571 | Repo HEAD can be exported by repo authors #880

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions core/common/mixins.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -969,6 +969,11 @@ def save_as_new_version(self, user, **kwargs): # pylint: disable=too-many-branc
class ConceptContainerExportMixin:
permission_classes = (CanViewConceptDictionaryVersion, IsAuthenticated)

@staticmethod
def is_head_export_permitted(user, version):
"""Allow HEAD exports for repo admins, not just staff users."""
return user.is_staff or user.is_superuser or user.is_admin_for(version)

def get_object(self):
queryset = self.get_queryset()
if 'version' not in self.kwargs:
Expand All @@ -988,7 +993,7 @@ def get(self, request, *args, **kwargs): # pylint: disable=unused-argument
logger.debug(
'Export requested for %s version %s', self.entity.lower(), version.version
)
if version.is_head and not request.user.is_staff:
if version.is_head and not self.is_head_export_permitted(request.user, version):
return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)

if version.is_exporting:
Expand All @@ -1014,7 +1019,7 @@ def get(self, request, *args, **kwargs): # pylint: disable=unused-argument
def post(self, request, *args, **kwargs): # pylint: disable=unused-argument
version = self.get_object()

if version.is_head and not request.user.is_staff:
if version.is_head and not self.is_head_export_permitted(request.user, version):
return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)

logger.debug('%s Export requested for version %s (post)', self.entity, version.version)
Expand All @@ -1040,7 +1045,7 @@ def delete(self, request, *args, **kwargs): # pylint: disable=unused-argument
user = request.user
version = self.get_object()

if version.is_head and not user.is_staff:
if version.is_head and not self.is_head_export_permitted(user, version):
return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED)

permitted = user.is_staff or user.is_superuser or user.is_admin_for(version)
Expand Down
44 changes: 36 additions & 8 deletions core/integration_tests/tests_collections.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

from celery_once import AlreadyQueued
from mock import patch, Mock, ANY
from mock.mock import PropertyMock
from rest_framework.exceptions import ErrorDetail

from core.collections.models import CollectionReference, Collection
Expand Down Expand Up @@ -2614,7 +2615,7 @@ def test_get_204_head(self, s3_exists_mock):

response = self.client.get(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -2662,7 +2663,7 @@ def test_get_302_head(self, s3_exists_mock, s3_url_for_mock):

response = self.client.get(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand All @@ -2679,7 +2680,7 @@ def test_get_500_head_when_signed_url_generation_fails(self, s3_exists_mock, s3_

response = self.client.get(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand All @@ -2689,18 +2690,20 @@ def test_get_500_head_when_signed_url_generation_fails(self, s3_exists_mock, s3_
s3_url_for_mock.assert_called_once_with(f"users/username/username_coll_vHEAD.{self.HEAD_updated_at}.zip")

def test_get_405(self):
random_user = UserProfileFactory()
response = self.client.get(
f'/users/{self.collection.parent.mnemonic}/collections/{self.collection.mnemonic}/{"HEAD"}/export/',
HTTP_AUTHORIZATION='Token ' + self.token,
HTTP_AUTHORIZATION='Token ' + random_user.get_token(),
format='json'
)

self.assertEqual(response.status_code, 405)

def test_post_405(self):
random_user = UserProfileFactory()
response = self.client.post(
f'/users/{self.collection.parent.mnemonic}/collections/{self.collection.mnemonic}/{"HEAD"}/export/',
HTTP_AUTHORIZATION='Token ' + self.token,
HTTP_AUTHORIZATION='Token ' + random_user.get_token(),
format='json'
)

Expand All @@ -2711,7 +2714,7 @@ def test_post_303_head(self, s3_exists_mock):
s3_exists_mock.return_value = True
response = self.client.post(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -2739,7 +2742,7 @@ def test_post_202_head(self, s3_exists_mock, export_collection_mock):
export_collection_mock.__name__ = 'export_collection'
response = self.client.post(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -2773,7 +2776,7 @@ def test_post_409_head(self, s3_exists_mock, export_collection_mock):
export_collection_mock.apply_async.side_effect = AlreadyQueued('already-queued')
response = self.client.post(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -2802,6 +2805,31 @@ def test_post_409_version(self, s3_has_path_mock, export_collection_mock):
self.assertEqual(
Task.objects.filter(created_by=self.user, state='PENDING', name='export_collection').count(), 0)

def test_delete_405_head(self):
random_user = UserProfileFactory()
response = self.client.delete(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + random_user.get_token(),
format='json'
)

self.assertEqual(response.status_code, 405)

@patch('core.collections.models.Collection.version_export_path', new_callable=PropertyMock)
@patch('core.collections.models.Collection.has_export')
@patch('core.services.storages.cloud.aws.S3.remove')
def test_delete_204_head(self, s3_remove_mock, has_export_mock, export_path_mock):
has_export_mock.return_value = True
export_path_mock.return_value = 'head/export/path'
response = self.client.delete(
self.collection.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

self.assertEqual(response.status_code, 204)
s3_remove_mock.assert_called_once_with('head/export/path')


class CollectionVersionListViewTest(OCLAPITestCase):
def setUp(self):
Expand Down
35 changes: 26 additions & 9 deletions core/integration_tests/tests_sources.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -983,7 +983,7 @@ def test_get_204_head(self, s3_exists_mock):

response = self.client.get(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -1011,7 +1011,7 @@ def test_get_302_head(self, s3_exists_mock, s3_url_for_mock):

response = self.client.get(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -1048,7 +1048,7 @@ def test_get_500_head_when_signed_url_generation_fails(self, s3_exists_mock, s3_

response = self.client.get(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand All @@ -1064,7 +1064,7 @@ def test_get_208_HEAD(self, s3_exists_mock, is_exporting_mock):

response = self.client.get(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand All @@ -1086,18 +1086,20 @@ def test_get_208_version(self, s3_has_path_mock, is_exporting_mock):
s3_has_path_mock.assert_not_called()

def test_get_405(self):
random_user = UserProfileFactory()
response = self.client.get(
f'/users/{self.source.parent.mnemonic}/sources/{self.source.mnemonic}/{"HEAD"}/export/',
HTTP_AUTHORIZATION='Token ' + self.token,
HTTP_AUTHORIZATION='Token ' + random_user.get_token(),
format='json'
)

self.assertEqual(response.status_code, 405)

def test_post_405(self):
random_user = UserProfileFactory()
response = self.client.post(
f'/users/{self.source.parent.mnemonic}/sources/{self.source.mnemonic}/{"HEAD"}/export/',
HTTP_AUTHORIZATION='Token ' + self.token,
HTTP_AUTHORIZATION='Token ' + random_user.get_token(),
format='json'
)

Expand All @@ -1108,7 +1110,7 @@ def test_post_303_head(self, s3_exists_mock):
s3_exists_mock.return_value = True
response = self.client.post(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -1136,7 +1138,7 @@ def test_post_202_head(self, s3_exists_mock, export_source_mock):
s3_exists_mock.return_value = False
response = self.client.post(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -1181,7 +1183,7 @@ def test_post_409_head(self, s3_exists_mock, export_source_mock):
s3_exists_mock.return_value = False
response = self.client.post(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.admin_token,
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

Expand Down Expand Up @@ -1215,6 +1217,21 @@ def test_delete_405(self):

self.assertEqual(response.status_code, 405)

@patch('core.sources.models.Source.version_export_path', new_callable=PropertyMock)
@patch('core.sources.models.Source.has_export')
@patch('core.services.storages.cloud.aws.S3.remove')
def test_delete_204_head(self, s3_remove_mock, has_export_mock, export_path_mock):
has_export_mock.return_value = True
export_path_mock.return_value = 'head/export/path'
response = self.client.delete(
self.source.uri + 'HEAD/export/',
HTTP_AUTHORIZATION='Token ' + self.token,
format='json'
)

self.assertEqual(response.status_code, 204)
s3_remove_mock.assert_called_once_with('head/export/path')

def test_delete_403(self):
random_user = UserProfileFactory()
response = self.client.delete(
Expand Down
7 changes: 5 additions & 2 deletions core/users/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,8 +168,11 @@ def set_token(self, token):
Token.objects.create(user=self, key=token)

def is_admin_for(self, concept_container): # pragma: no cover
parent_id = concept_container.parent_id
return parent_id == self.id or self.organizations.filter(id=parent_id).exists()
user_id = concept_container.user_id
if user_id and user_id == self.id:
return True
organization_id = concept_container.organization_id
return self.organizations.filter(id=organization_id).exists()

def __create_token(self):
return Token.objects.create(user=self)
Expand Down