From d619bcbaf1181b4a76941e7f31ed7c784a748c64 Mon Sep 17 00:00:00 2001 From: Sunny Aggarwal Date: Thu, 8 Oct 2026 13:02:54 +0530 Subject: [PATCH] OpenConceptLab/ocl_online#397 | changing source public_access re-index resources via task --- core/common/mixins.py | 2 + core/common/models.py | 1 - core/common/tasks.py | 29 ++++++++ core/common/tests.py | 29 +++++++- core/sources/models.py | 16 ++++- core/sources/signals.py | 9 +-- core/sources/tests/tests.py | 136 ++++++++++++++++++++++++++++++++++++ 7 files changed, 211 insertions(+), 11 deletions(-) diff --git a/core/common/mixins.py b/core/common/mixins.py index 3feb5291..fae329cd 100644 --- a/core/common/mixins.py +++ b/core/common/mixins.py @@ -535,6 +535,8 @@ def update(self, request, *args, **kwargs): # pylint: disable=unused-argument if serializer.is_valid(): self.object.get_checksums(recalculate=True) serializer = self.get_detail_serializer(self.object) + if get(self.object, '_public_access_task'): + success_status_code = status.HTTP_202_ACCEPTED return Response(serializer.data, status=success_status_code) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) diff --git a/core/common/models.py b/core/common/models.py index bdbd6a45..3458d8f3 100644 --- a/core/common/models.py +++ b/core/common/models.py @@ -543,7 +543,6 @@ def get_actions(ids): '_index': index_name, '_id': object_id, 'doc': partial_doc, - 'doc_as_upsert': True, } return BaseModel.batch_index_partial_by_ids(queryset, document, get_actions, single_batch, parallel) diff --git a/core/common/tasks.py b/core/common/tasks.py index 94f0d220..19814d59 100644 --- a/core/common/tasks.py +++ b/core/common/tasks.py @@ -903,6 +903,35 @@ def index_source_mappings( source.clear_mappings_cache() +@app.task( + bind=True, autoretry_for=(Exception, WorkerLostError, ), retry_backoff=True, retry_backoff_max=600, + max_retries=2, acks_late=True, reject_on_worker_lost=True +) +def index_source_public_access(self, source_id): + """ + Sets public_can_view on a source's concept and mapping docs from its current public_access, with retries; the + source shows as processing meanwhile. Not queued once, so each public_access change gets its own run. + """ + from core.sources.models import Source + from core.concepts.documents import ConceptDocument + from core.mappings.documents import MappingDocument + source = Source.objects.filter(id=source_id).first() + if not source: + return + task_id = self.request.id + source.add_processing(task_id) + try: + partial_doc = {'public_can_view': source.public_can_view} + batch_index_with_summary(source.batch_index, source.concepts_set, ConceptDocument, partial_doc=partial_doc) + batch_index_with_summary(source.batch_index, source.mappings_set, MappingDocument, partial_doc=partial_doc) + except Exception: + if self.request.retries < self.max_retries: + raise + source.remove_processing(task_id) + raise + source.remove_processing(task_id) + + @app.task(base=QueueOnceCustomTask) def update_source_active_concepts_count(source_id): from core.sources.models import Source diff --git a/core/common/tests.py b/core/common/tests.py index f5bfb0a3..0be40302 100644 --- a/core/common/tests.py +++ b/core/common/tests.py @@ -1719,18 +1719,43 @@ def get_batch(batch_slice): '_index': doc_instance._index._name, # pylint: disable=protected-access '_id': 10, 'doc': {'flag': True}, - 'doc_as_upsert': True }, { '_op_type': 'update', '_index': doc_instance._index._name, # pylint: disable=protected-access '_id': 11, 'doc': {'flag': True}, - 'doc_as_upsert': True } ]) self.assertEqual([call.kwargs for call in bulk_calls], [{'raise_on_error': False}]) + @override_settings(TEST_MODE=False) + @patch('core.common.models.parallel_bulk', return_value=[(False, {'update': {'_id': 11, 'status': 404}})]) + def test_batch_index_partial_full_indexes_missing_docs_instead_of_upserting(self, _): + ids_queryset = MagicMock() + ids_queryset.__getitem__.side_effect = lambda s: [10, 11] if s == slice(0, 500, None) else [] + + queryset = Mock() + ordered_queryset = Mock() + ordered_queryset.values_list.return_value = ids_queryset + queryset.order_by.return_value = ordered_queryset + missing_queryset = Mock() + queryset.filter.return_value = missing_queryset + + doc_instance = Mock() + doc_instance.django.auto_refresh = False + doc_instance.django.queryset_pagination = None + document = Mock(return_value=doc_instance) + + with patch.object(BaseModel, 'batch_index_full') as batch_index_full_mock: + BaseModel.batch_index_partial(queryset, document, False, {'public_can_view': False}) + + queryset.filter.assert_called_once_with(id__in={11}) + batch_index_full_mock.assert_called_once_with( + single_batch=False, queryset=missing_queryset, document=document, prefetch=[], select_related=[], + refresh=False + ) + @override_settings(TEST_MODE=False) @patch('core.common.models.parallel_bulk', return_value=[]) def test_batch_index_source_version_append_streams_batches_without_parallel_bulk(self, parallel_bulk_mock): diff --git a/core/sources/models.py b/core/sources/models.py index cf465d65..c039198e 100644 --- a/core/sources/models.py +++ b/core/sources/models.py @@ -15,7 +15,7 @@ from core.common.constants import HEAD from core.common.models import ConceptContainerModel from core.common.tasks import update_mappings_source, index_source_concepts, index_source_mappings, \ - sync_source_concept_vectors, resolve_url_registry_entries + sync_source_concept_vectors, resolve_url_registry_entries, index_source_public_access from core.common.utils import to_camel_case from core.common.validators import validate_non_negative from core.concepts.models import ConceptName, Concept @@ -544,6 +544,20 @@ def index_concepts_async(self, user, partial_doc=None, locales=None, exclude_loc except AlreadyQueued: pass + def index_public_access_async(self, user=None): + """ + Queues the public_can_view re-index of this source's docs once the change is committed, marking the source + processing before the response goes out, and returns its Task. In TEST_MODE it runs inline: returns None. + """ + if get(settings, 'TEST_MODE', False): + index_source_public_access(self.id) # pylint: disable=no-value-for-parameter + return None + task = Task.new(queue='indexing', user=user or self.updated_by, name=index_source_public_access.__name__) + self.add_processing(task.id) + transaction.on_commit(lambda: index_source_public_access.apply_async( + (self.id, ), queue='indexing', persist_args=True, task_id=task.id)) + return task + def sync_concept_vectors_async(self, user=None, recheck=True, countdown=None): """ Queues this version's vector sync (sync_source_concept_vectors) on the indexing queue, with its arguments diff --git a/core/sources/signals.py b/core/sources/signals.py index 1d581f32..5e0051f9 100644 --- a/core/sources/signals.py +++ b/core/sources/signals.py @@ -20,10 +20,5 @@ def propagate_parent_attributes(sender, instance=None, created=False, **kwargs): updated_mappings = instance.mappings_set.exclude( public_access=instance.public_access).update(public_access=instance.public_access) - partial_doc = {'public_can_view': instance.public_can_view} - if updated_concepts: - from core.concepts.documents import ConceptDocument - instance.batch_index(instance.concepts_set, ConceptDocument, partial_doc=partial_doc) - if updated_mappings: - from core.mappings.documents import MappingDocument - instance.batch_index(instance.mappings_set, MappingDocument, partial_doc=partial_doc) + if updated_concepts or updated_mappings: + instance._public_access_task = instance.index_public_access_async() # pylint: disable=protected-access diff --git a/core/sources/tests/tests.py b/core/sources/tests/tests.py index 879c0f82..e6611b96 100644 --- a/core/sources/tests/tests.py +++ b/core/sources/tests/tests.py @@ -24,6 +24,7 @@ from core.common.tasks import update_source_active_concepts_count from core.common.tasks import update_source_active_mappings_count from core.common.tasks import update_validation_schema +from core.common.tasks import index_source_public_access from core.common.tests import OCLTestCase, OCLAPITestCase from core.concepts.documents import ConceptDocument from core.concepts.models import Concept @@ -130,6 +131,34 @@ def test_version_update_sets_external_id_from_version_external_id(self): version.refresh_from_db() self.assertEqual(version.external_id, 'ext-1') + @patch('core.sources.models.Source.index_public_access_async') + def test_put_public_access_change_returns_202_with_reindex_queued(self, index_public_access_async_mock): + source = OrganizationSourceFactory( + created_by=self.admin, updated_by=self.admin, public_access=ACCESS_TYPE_VIEW) + concept = ConceptFactory(parent=source, public_access=ACCESS_TYPE_VIEW) + + response = self.client.put( + source.uri, {'public_access': ACCESS_TYPE_NONE}, format='json', + HTTP_AUTHORIZATION=f"Token {self.admin_token}" + ) + + self.assertEqual(response.status_code, 202) + self.assertEqual(response.data['public_access'], ACCESS_TYPE_NONE) + index_public_access_async_mock.assert_called_once_with() + concept.refresh_from_db() + self.assertEqual(concept.public_access, ACCESS_TYPE_NONE) + + @patch('core.sources.models.Source.index_public_access_async') + def test_put_without_public_access_change_returns_200(self, index_public_access_async_mock): + source = OrganizationSourceFactory(created_by=self.admin, updated_by=self.admin) + ConceptFactory(parent=source) + + response = self.client.put( + source.uri, {'name': 'renamed'}, format='json', HTTP_AUTHORIZATION=f"Token {self.admin_token}") + + self.assertEqual(response.status_code, 200) + index_public_access_async_mock.assert_not_called() + def test_version_delete_validation_error(self): source = OrganizationSourceFactory(created_by=self.admin, updated_by=self.admin) version = OrganizationSourceFactory( @@ -2078,6 +2107,35 @@ def test_index_mappings_async_swallows_already_queued(self, mock_task): source = OrganizationSourceFactory() source.index_mappings_async(source.created_by) + @override_settings(TEST_MODE=False) + @patch('core.sources.models.index_source_public_access') + def test_index_public_access_async_marks_processing_and_queues_on_commit(self, mock_task): + mock_task.__name__ = 'index_source_public_access' + source = OrganizationSourceFactory() + + with self.captureOnCommitCallbacks(execute=False) as callbacks: + task = source.index_public_access_async(source.created_by) + + self.assertEqual(task.name, 'index_source_public_access') + self.assertEqual(task.queue, 'indexing') + source.refresh_from_db() + self.assertEqual(source._background_process_ids, [task.id]) # pylint: disable=protected-access + mock_task.apply_async.assert_not_called() + + for callback in callbacks: + callback() + mock_task.apply_async.assert_called_once_with( + (source.id, ), queue='indexing', persist_args=True, task_id=task.id) + + @patch('core.sources.models.index_source_public_access') + def test_index_public_access_async_runs_inline_in_test_mode(self, mock_task): + source = OrganizationSourceFactory() + + self.assertIsNone(source.index_public_access_async()) + + mock_task.assert_called_once_with(source.id) + mock_task.apply_async.assert_not_called() + @patch('core.sources.models.index_source_concepts') def test_index_concepts_async_swallows_already_queued(self, mock_task): mock_task.__name__ = 'index_source_concepts' @@ -2310,6 +2368,31 @@ def test_propagate_parent_attributes_updates_mapping_public_access(self): mapping.refresh_from_db() self.assertEqual(mapping.public_access, ACCESS_TYPE_VIEW) + @patch('core.sources.models.Source.index_public_access_async') + def test_propagate_parent_attributes_queues_public_access_reindex(self, index_public_access_async_mock): + source = OrganizationSourceFactory(public_access=ACCESS_TYPE_VIEW) + ConceptFactory(parent=source, public_access=ACCESS_TYPE_VIEW) + + source.public_access = ACCESS_TYPE_NONE + source._should_update_public_access = True # pylint: disable=protected-access + source.save() + + index_public_access_async_mock.assert_called_once_with() + self.assertEqual( + source._public_access_task, index_public_access_async_mock.return_value) # pylint: disable=protected-access + + @patch('core.sources.models.Source.index_public_access_async') + def test_propagate_parent_attributes_skips_reindex_without_children_to_update( + self, index_public_access_async_mock): + source = OrganizationSourceFactory(public_access=ACCESS_TYPE_VIEW) + + source.public_access = ACCESS_TYPE_NONE + source._should_update_public_access = True # pylint: disable=protected-access + source.save() + + index_public_access_async_mock.assert_not_called() + self.assertFalse(hasattr(source, '_public_access_task')) + class SourceCloneAPITest(OCLAPITestCase): def test_clone_api_returns_structured_errors_and_rolls_back(self): @@ -2425,6 +2508,59 @@ def test_clean_filters_extra_keys(self): source.clean_filters() +class IndexSourcePublicAccessTaskTest(OCLTestCase): + @staticmethod + def run_task(source_id, task_id='public-access-task', retries=0): + """Runs the task body as a worker would, with this task id and retry count.""" + index_source_public_access.push_request(id=task_id, retries=retries) + try: + return index_source_public_access.run(source_id) + finally: + index_source_public_access.pop_request() + + @patch('core.sources.models.Source.batch_index') + def test_sets_public_can_view_on_concepts_and_mappings_while_processing(self, batch_index_mock): + source = OrganizationSourceFactory(public_access=ACCESS_TYPE_NONE) + processing_ids = [] + batch_index_mock.side_effect = lambda *args, **kwargs: processing_ids.append( + Source.objects.get(id=source.id)._background_process_ids) # pylint: disable=protected-access + + self.run_task(source.id) + + self.assertEqual(batch_index_mock.call_args_list, [ + call(ANY, ConceptDocument, partial_doc={'public_can_view': False}), + call(ANY, MappingDocument, partial_doc={'public_can_view': False}), + ]) + self.assertEqual(processing_ids, [['public-access-task'], ['public-access-task']]) + source.refresh_from_db() + self.assertEqual(source._background_process_ids, []) # pylint: disable=protected-access + + @patch('core.sources.models.Source.batch_index', side_effect=BatchIndexingError('ES down', {})) + def test_stays_processing_when_it_will_be_retried(self, _): + source = OrganizationSourceFactory() + + with self.assertRaises(BatchIndexingError): + self.run_task(source.id, retries=0) + + source.refresh_from_db() + self.assertEqual(source._background_process_ids, ['public-access-task']) # pylint: disable=protected-access + + @patch('core.sources.models.Source.batch_index', side_effect=BatchIndexingError('ES down', {})) + def test_stops_processing_when_out_of_retries(self, _): + source = OrganizationSourceFactory() + + with self.assertRaises(BatchIndexingError): + self.run_task(source.id, retries=index_source_public_access.max_retries) + + source.refresh_from_db() + self.assertEqual(source._background_process_ids, []) # pylint: disable=protected-access + + @patch('core.sources.models.Source.batch_index') + def test_does_nothing_for_missing_source(self, batch_index_mock): + self.assertIsNone(self.run_task(0)) + batch_index_mock.assert_not_called() + + class TasksTest(OCLTestCase): @patch('core.sources.models.Source.index_children') @patch('core.common.tasks.export_source')