diff --git a/core/collections/constants.py b/core/collections/constants.py index a6f5e4206..d4dd9748e 100644 --- a/core/collections/constants.py +++ b/core/collections/constants.py @@ -37,3 +37,7 @@ UNRESOLVED_RESOURCE_VERSION_TYPE = 'reference.resource_version' INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE = 'INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE' EXPANSION_LOCKED_REPO_VERSION_TOGGLE = 'EXPANSION_LOCKED_REPO_VERSION_TOGGLE' +VERSION_EXPANSION_LOCK = 'lock' +VERSION_EXPANSION_LATEST = 'latest' +VERSION_EXPANSION_NONE = 'none' +VERSION_EXPANSION_CHOICES = (VERSION_EXPANSION_LOCK, VERSION_EXPANSION_LATEST, VERSION_EXPANSION_NONE) diff --git a/core/collections/models.py b/core/collections/models.py index f5723cd77..4d5590841 100644 --- a/core/collections/models.py +++ b/core/collections/models.py @@ -18,7 +18,8 @@ REFERENCE_TYPE_CHOICES, CONCEPT_REFERENCE_TYPE, MAPPING_REFERENCE_TYPE, SOURCE_MAPPINGS, SOURCE_TO_CONCEPTS, TRANSFORM_TO_RESOURCE_VERSIONS, COLLECTION_REFERENCE_TYPE, TRANSFORM_TO_EXTENSIONAL, RESOURCE_VERSION_REPO_RESOLUTION_TOGGLE, UNRESOLVED_RESOURCE_VERSION_TYPE, - INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE, EXPANSION_LOCKED_REPO_VERSION_TOGGLE) + INTENSIONAL_REFERENCE_RESOURCE_VERSIONS_TOGGLE, EXPANSION_LOCKED_REPO_VERSION_TOGGLE, VERSION_EXPANSION_LOCK, + VERSION_EXPANSION_LATEST) from core.collections.parsers import CollectionReferenceParser from core.collections.translators import CollectionReferenceTranslator from core.collections.utils import is_concept, is_mapping @@ -402,19 +403,21 @@ def get_locked_repo_versions(self, relation): expansion = self.expansion if CollectionReference.is_expansion_locked_repo_version_on() else None return list(getattr(expansion, relation).filter(is_active=True).order_by('-id')) if expansion else [] - def can_seed_from_head_expansion(self): - return bool( - not self.is_head and self.should_auto_expand and CollectionReference.is_expansion_locked_repo_version_on() - and get(self.head, 'expansion') - ) + def get_expansion_choice(self, expansion=None): + """How a new version's expansion is built with the lock toggle on: lock or latest, else None (as before)""" + if self.is_head or not self.should_auto_expand or not CollectionReference.is_expansion_locked_repo_version_on(): + return None + if expansion in (VERSION_EXPANSION_LOCK, VERSION_EXPANSION_LATEST): + return expansion + return VERSION_EXPANSION_LOCK if get(self.head, 'expansion') else VERSION_EXPANSION_LATEST - def seed_from_head_expansion(self, index=True): - """False when HEAD has no expansion to copy, so the version seeds as without one""" + def seed_from_head_expansion(self, lock=True, index=True): + """References from HEAD, evaluated again with HEAD's expansion parameters (and its locks, with lock)""" is_outermost = not transaction.get_connection().in_atomic_block attempts = SERIALIZATION_RETRY_ATTEMPTS if is_outermost else 1 for attempt in range(1, attempts + 1): try: - expansion = self.seed_from_head_expansion_in_snapshot(is_outermost) + expansion = self.seed_from_head_expansion_in_snapshot(is_outermost, lock) break except OperationalError as ex: if attempt == attempts or not self.is_serialization_failure(ex): @@ -422,16 +425,14 @@ def seed_from_head_expansion(self, index=True): self.refresh_from_db() self.__dict__.pop('locked_source_versions', None) self.__dict__.pop('locked_collection_versions', None) - if not expansion: - return False if index: expansion.index_resources( list(expansion.concepts.values_list('id', flat=True)), list(expansion.mappings.values_list('id', flat=True))) return True - def seed_from_head_expansion_in_snapshot(self, is_outermost): - """References and expansion copied from HEAD in one transaction, so both come from the same HEAD state""" + def seed_from_head_expansion_in_snapshot(self, is_outermost, lock=True): + """References, parameters and locks read from HEAD in one transaction, so all come from the same HEAD state""" with transaction.atomic(): if is_outermost: with transaction.get_connection().cursor() as cursor: @@ -442,30 +443,28 @@ def seed_from_head_expansion_in_snapshot(self, is_outermost): return expansion head = self.head head_expansion = Expansion.objects.filter(uri=head.expansion_uri).first() if head.expansion_uri else None - if not head_expansion: - return None self.seed_references() + parameters = dict(get(head_expansion, 'parameters') or {}) + system_versions = head_expansion.get_locked_system_version_parameter() if lock and head_expansion else None + if system_versions: + parameters[ExpansionParameters.INCLUDE_SYSTEM] = system_versions expansion = Expansion( - mnemonic=mnemonic, collection_version=self, parameters=head_expansion.parameters, + mnemonic=mnemonic, collection_version=self, parameters=parameters, created_by_id=self.created_by_id, updated_by_id=self.created_by_id) expansion.clean() expansion.full_clean() expansion.save() - for relation in Expansion.REPO_VERSION_RELATIONS: - getattr(expansion, relation).set(getattr(head_expansion, relation).all()) + if lock and head_expansion: + # system-version doesn't reach value sets, so they start from HEAD's value set lock + expansion.evaluated_collection_versions.set( + head_expansion.evaluated_collection_versions.filter(is_active=True)) if not self.expansion_uri: self.expansion_uri = expansion.uri self.save() - if head_expansion.is_processing or head_expansion.has_pending_readd: - # HEAD's expansion is mid-build, so evaluate this version's references against HEAD's locks - references = list(self.references.all()) - for reference in references: - reference._async = True # pylint: disable=protected-access - expansion.add_references(references, index=False, force_reevaluate=True) - else: - expansion.copy_members_from(head_expansion) - expansion.unresolved_repo_versions = head_expansion.unresolved_repo_versions - expansion.save(update_fields=['unresolved_repo_versions']) + references = list(self.references.all()) + for reference in references: + reference._async = True # pylint: disable=protected-access + expansion.add_references(references, index=False, force_reevaluate=True) self.replace_head_resources(expansion) return expansion @@ -1797,11 +1796,11 @@ def resolve_system_version_entries(value, namespace=None): ] @classmethod - def get_system_version_conflicts(cls, value, namespace=None): - """The system-version entries naming different versions of one repository, by the repository's URL. - An entry that doesn't resolve still counts, under the repository its URL names and the version it asks for.""" + def get_system_version_entry_repos(cls, value, namespace=None): + """Each system-version entry with the repository it names and the version it asks for. + An entry that doesn't resolve still names a repository, by its URL without the version.""" from core.sources.models import Source - entries_by_repo = {} + repos = [] for entry, version in cls.resolve_system_version_entries(value, namespace): url, requested = canonical_url_to_url_and_version(entry) if version.id: @@ -1813,6 +1812,14 @@ def get_system_version_conflicts(cls, value, namespace=None): path = url if url.endswith('/') else url + '/' repo_url = drop_version(path) requested = requested or path[len(repo_url):].strip('/') or None + repos.append((entry, repo_url, requested)) + return repos + + @classmethod + def get_system_version_conflicts(cls, value, namespace=None): + """The system-version entries naming different versions of one repository, by the repository's URL.""" + entries_by_repo = {} + for entry, repo_url, requested in cls.get_system_version_entry_repos(value, namespace): entries_by_repo.setdefault(repo_url, {}).setdefault(requested, entry) return { repo_url: list(entries.values()) for repo_url, entries in entries_by_repo.items() if len(entries) > 1 @@ -2081,20 +2088,28 @@ def index_resources(self, concepts, mappings): if mappings: self.index_mappings(mappings) - @property - def has_pending_readd(self): - """A reference removal leaves this expansion short until its re-add succeeds""" - return Task.unfinished_readd_references_tasks().filter(args__0=self.id).exists() - - def copy_members_from(self, expansion): - for relation in ('concepts', 'mappings'): - through = getattr(Expansion, relation).through - field = f'{relation[:-1]}_id' - with transaction.get_connection().cursor() as cursor: - cursor.execute( - f'INSERT INTO {through._meta.db_table} (expansion_id, {field}) ' # pylint: disable=protected-access - f'SELECT %s, {field} FROM {through._meta.db_table} WHERE expansion_id = %s', # pylint: disable=protected-access - [self.id, expansion.id]) + def get_locked_system_version_parameter(self): + """This expansion's system-version, plus each source it locked that the parameter doesn't name yet. + Entries resolve in the owner's namespace, as the expansion built from them will.""" + namespace = self.get_system_version_namespace(self.collection_version.parent_url) + value = str(get(self.parameters, ExpansionParameters.INCLUDE_SYSTEM) or '') + repos = self.get_system_version_entry_repos(value, namespace) + entries = [entry for entry, _, _ in repos] + named = {repo_url for _, repo_url, _ in repos} + for version in self.evaluated_source_versions.filter(is_active=True).order_by('-id'): + if version.versioned_object_url in named: + continue + named.add(version.versioned_object_url) + entry = f"{version.canonical_url}|{version.version}" if version.canonical_url else None + if not entry or self.resolve_system_version_entries(entry, namespace)[0][1].id != version.id: + entry = f"{version.versioned_object_url}|{version.version}" + entries.append(entry) + locked = ','.join(entries) + # as on expansion create, one version per source: never add a conflict HEAD's own parameter doesn't have + if set(self.get_system_version_conflicts(locked, namespace)) - set( + self.get_system_version_conflicts(value, namespace)): + return value + return locked @staticmethod def replace_head_rows(rows, owner, field, klass): @@ -2111,12 +2126,32 @@ def replace_head_rows(rows, owner, field, klass): rows.model.objects.filter(id__in=head_rows.values('id')).update(**{f'{field}_id': Subquery(latest)}) def seed_children(self, index=True, force_reevaluate=False): + version = self.collection_version + # a version's default expansion holds resource versions, so a rebuild swaps HEAD rows as its creation did + swap_head = force_reevaluate and not version.is_head and self.is_default and \ + CollectionReference.is_expansion_locked_repo_version_on() + previous_concept_ids = set(self.concepts.values_list('id', flat=True)) if swap_head and index else set() + previous_mapping_ids = set(self.mappings.values_list('id', flat=True)) if swap_head and index else set() + references = version.references + if swap_head: + # cascades run without a result cap, as when the version was created + references = list(references.all()) + for reference in references: + reference._async = True # pylint: disable=protected-access self.add_references( - references=self.collection_version.references, - index=index, + references=references, + index=index and not swap_head, is_adding_all=True, force_reevaluate=force_reevaluate ) + if swap_head: + version.replace_head_resources(self) + if index: + concept_ids = set(self.concepts.values_list('id', flat=True)) + mapping_ids = set(self.mappings.values_list('id', flat=True)) + self.index_resources(list(concept_ids), list(mapping_ids)) + self.full_index_resources( + list(previous_concept_ids - concept_ids), list(previous_mapping_ids - mapping_ids)) def wait_until_processed(self): # pragma: no cover processing = self.is_processing diff --git a/core/collections/serializers.py b/core/collections/serializers.py index ad4da25aa..f246034a7 100644 --- a/core/collections/serializers.py +++ b/core/collections/serializers.py @@ -4,11 +4,12 @@ from django.core.validators import RegexValidator from pydash import get from rest_framework.fields import CharField, ListField, IntegerField, DateTimeField, JSONField, \ - BooleanField, SerializerMethodField + BooleanField, SerializerMethodField, ChoiceField from rest_framework.relations import PrimaryKeyRelatedField from rest_framework.serializers import ModelSerializer, Serializer, ValidationError from core.client_configs.serializers import ClientConfigSerializer +from core.collections.constants import VERSION_EXPANSION_CHOICES, VERSION_EXPANSION_NONE from core.collections.models import Collection, CollectionReference, Expansion from core.common.constants import HEAD, DEFAULT_ACCESS_TYPE, NAMESPACE_REGEX, INCLUDE_SUMMARY, \ INCLUDE_CLIENT_CONFIGS, INVALID_EXPANSION_URL, INCLUDE_STATES, INCLUDE_TASKS, INCLUDE_RESOLVED_REPO_VERSIONS, \ @@ -264,6 +265,7 @@ class CollectionCreateSerializer(CollectionCreateOrUpdateSerializer): locked_date = DateTimeField(required=False, allow_null=True) autoexpand_head = BooleanField(required=False, default=True) autoexpand = BooleanField(required=False, default=True) + expansion = ChoiceField(choices=VERSION_EXPANSION_CHOICES, required=False, write_only=True) def validate(self, attrs): # create/create_version consume the raw payload (not validated_data), so DRF's boolean @@ -287,7 +289,12 @@ def create(self, validated_data): def create_version(self, validated_data): collection = self.prepare_object(validated_data) user = self.context['request'].user - errors = Collection.persist_new_version(collection, user) + expansion = validated_data.get('expansion') + if expansion == VERSION_EXPANSION_NONE: + collection.autoexpand = False + elif expansion: + collection.autoexpand = True + errors = Collection.persist_new_version(collection, user, expansion=expansion) self._errors.update(errors) return collection diff --git a/core/collections/tests/tests.py b/core/collections/tests/tests.py index e8ac420dd..fae024bd0 100644 --- a/core/collections/tests/tests.py +++ b/core/collections/tests/tests.py @@ -8,6 +8,7 @@ from django.http import QueryDict from django.test import override_settings, tag from django.db import OperationalError +from django.db.models import F from mock import patch, Mock from mock.mock import ANY from rest_framework import status @@ -15,7 +16,7 @@ from core.collections.constants import SOURCE_TO_CONCEPTS, TRANSFORM_TO_RESOURCE_VERSIONS from core.collections.documents import CollectionDocument -from core.common.constants import SERIALIZATION_RETRY_ATTEMPTS +from core.common.constants import SERIALIZATION_RETRY_ATTEMPTS, HEAD from core.collections.models import CollectionReference, Collection, Expansion from core.collections.models import ExpansionParameters, ExpansionSystemParameter from core.collections.parsers import CollectionReferenceExpressionStringParser, \ @@ -35,7 +36,7 @@ from core.common.tests import OCLTestCase, OCLAPITestCase from core.common.utils import to_owner_uri, get_falsy_values from core.concepts.documents import ConceptDocument -from core.concepts.models import Concept +from core.concepts.models import Concept, HierarchicalConcepts from core.concepts.tests.factories import ConceptFactory, ConceptNameFactory from core.mappings.documents import MappingDocument from core.mappings.models import Mapping @@ -2187,28 +2188,102 @@ def test_seed_children_task(self, export_collection_task): self.assertEqual(expansion.mappings.count(), 1) export_collection_task.apply_async.assert_not_called() + @staticmethod + def head_locked_to_v1(): + """HEAD's expansion locks v1, whose concept version a later release v2 replaced""" + source, old, v1, v2 = CollectionReferenceTest.source_with_replaced_concept_version() + head = ExpansionTest.locked_auto_expansion(v1) + Expansion.objects.filter(uri=head.expansion_uri).update(parameters={'activeOnly': True}) + head.add_expressions({'expressions': [old.versioned_object.uri]}, head.created_by) + version = OrganizationCollectionFactory( + organization=head.organization, version='v1', mnemonic=head.mnemonic) + return source, old, v1, v2, head, version + @patch('core.common.tasks.export_collection') - def test_seed_children_task_copies_head_expansion(self, _): - collection = OrganizationCollectionFactory() - expansion = ExpansionFactory(collection_version=collection, parameters={'activeOnly': True}) - collection.expansion_uri = expansion.uri - collection.save() - concept = ConceptFactory() - mapping_version = MappingFactory().get_latest_version() - expansion.concepts.add(concept) - expansion.mappings.add(mapping_version) - expansion.evaluated_source_versions.add(concept.parent) - collection_v1 = OrganizationCollectionFactory( - organization=collection.organization, version='v1', mnemonic=collection.mnemonic) + def test_seed_children_task_lock_keeps_heads_versions_through_a_rebuild(self, _): + source, old, v1, _, head, version = self.head_locked_to_v1() + self.assertEqual(list(head.expansion.concepts.all()), [old]) - seed_children_to_new_version('collection', collection_v1.id, False) # pylint: disable=no-value-for-parameter + seed_children_to_new_version('collection', version.id, False, expansion='lock') # pylint: disable=no-value-for-parameter - collection_v1.refresh_from_db() - copy = collection_v1.expansion - self.assertEqual(list(copy.concepts.all()), [concept.get_latest_version()]) - self.assertEqual(list(copy.mappings.all()), [mapping_version]) - self.assertEqual(list(copy.evaluated_source_versions.all()), [concept.parent]) - self.assertEqual(copy.parameters, expansion.parameters) + version.refresh_from_db() + expansion = version.expansion + self.assertEqual(list(expansion.concepts.all()), [old]) + self.assertEqual(list(expansion.evaluated_source_versions.all()), [v1]) + self.assertEqual(expansion.parameters, {'activeOnly': True, 'system-version': f'{source.uri}|v1'}) + self.assertEqual(list(version.references.first().concepts.all()), [old]) + + expansion.seed_children(index=False, force_reevaluate=True) + + self.assertEqual(list(expansion.concepts.all()), [old]) + self.assertEqual(list(expansion.evaluated_source_versions.all()), [v1]) + + @patch('core.common.tasks.export_collection') + def test_seed_children_task_latest_evaluates_against_the_latest_release(self, _): + _, old, _, v2, _, version = self.head_locked_to_v1() + new = old.versioned_object.get_latest_version() + + seed_children_to_new_version('collection', version.id, False, expansion='latest') # pylint: disable=no-value-for-parameter + + version.refresh_from_db() + expansion = version.expansion + self.assertEqual(list(expansion.concepts.all()), [new]) + self.assertEqual(list(expansion.evaluated_source_versions.all()), [v2]) + self.assertEqual(expansion.parameters, {'activeOnly': True}) + self.assertEqual(list(version.references.first().concepts.all()), [new]) + + @patch('core.common.tasks.export_collection') + def test_seed_children_task_defaults_to_lock_with_a_head_expansion(self, _): + _, old, v1, _, _, version = self.head_locked_to_v1() + + seed_children_to_new_version('collection', version.id, False) # pylint: disable=no-value-for-parameter + + version.refresh_from_db() + self.assertEqual(list(version.expansion.concepts.all()), [old]) + self.assertEqual(list(version.expansion.evaluated_source_versions.all()), [v1]) + + def test_locked_system_version_parameter_keeps_sources_the_parameters_name(self): + source, _, v1, _ = CollectionReferenceTest.source_with_replaced_concept_version() + other = OrganizationSourceFactory() + expansion = ExpansionFactory( + collection_version=OrganizationCollectionFactory(), + parameters={'system-version': f'{source.uri}|v2'}) + expansion.evaluated_source_versions.add(v1, other) + + self.assertEqual(expansion.get_locked_system_version_parameter(), f'{source.uri}|v2,{other.uri}|HEAD') + + def test_locked_system_version_parameter_keeps_a_source_named_with_a_version_that_doesnt_resolve(self): + source, _, v1, _ = CollectionReferenceTest.source_with_replaced_concept_version() + for entry in (f'{source.uri}|v9', f'{source.uri}v9/'): + with self.subTest(entry=entry): + expansion = ExpansionFactory( + collection_version=OrganizationCollectionFactory(), parameters={'system-version': entry}) + expansion.evaluated_source_versions.add(v1) + + self.assertEqual(expansion.get_locked_system_version_parameter(), entry) + + def test_locked_system_version_parameter_never_adds_a_conflict(self): + source, _, v1, _ = CollectionReferenceTest.source_with_replaced_concept_version() + expansion = ExpansionFactory( + collection_version=OrganizationCollectionFactory(), parameters={'system-version': 'https://a.org|v1'}) + expansion.evaluated_source_versions.add(v1) + + def conflicts(value, namespace=None): # pylint: disable=unused-argument + return {source.uri: []} if source.uri in value else {} + + with patch.object(Expansion, 'get_system_version_conflicts', side_effect=conflicts): + self.assertEqual(expansion.get_locked_system_version_parameter(), 'https://a.org|v1') + + def test_locked_system_version_parameter_names_the_ocl_url_when_the_canonical_url_resolves_elsewhere(self): + source = OrganizationSourceFactory(canonical_url='https://example.com/cs') + expansion = ExpansionFactory(collection_version=OrganizationCollectionFactory()) + expansion.evaluated_source_versions.add(source) + + def unresolved(value, namespace=None): # pylint: disable=unused-argument + return [(entry.strip(), Mock(id=None)) for entry in value.split(',') if entry] + + with patch.object(Expansion, 'resolve_system_version_entries', side_effect=unresolved): + self.assertEqual(expansion.get_locked_system_version_parameter(), f'{source.uri}|HEAD') @staticmethod def head_with_head_concept_member(): @@ -2242,6 +2317,96 @@ def test_seed_children_task_swaps_head_resources_in_references(self, _, batch_in batch_index_resources_mock.apply_async.assert_called_once_with( ('concept', {'id__in': [latest.id]}), queue='indexing', permanent=False) + @staticmethod + def head_cascading_past_the_cap(): + """HEAD references a root with 1,000 children, one with a child: a capped cascade stops before it""" + source = OrganizationSourceFactory() + root = ConceptFactory(parent=source, mnemonic='root') + user = root.created_by + children = Concept.objects.bulk_create([ + Concept( + mnemonic=f'child{index}', version=HEAD, parent=source, concept_class='Diagnosis', datatype='None', + uri=f'{source.uri}concepts/child{index}/', created_by=user, updated_by=user, is_latest_version=True) + for index in range(1000) + ]) + Concept.objects.filter(id__in=[child.id for child in children]).update(versioned_object_id=F('id')) + Concept.sources.through.objects.bulk_create( + [Concept.sources.through(concept_id=child.id, source_id=source.id) for child in children]) + HierarchicalConcepts.objects.bulk_create( + [HierarchicalConcepts(child_id=child.id, parent_id=root.id) for child in children]) + grandchild = ConceptFactory(parent=source, mnemonic='grandchild') + HierarchicalConcepts.objects.create(child_id=grandchild.id, parent_id=children[0].id) + head = OrganizationCollectionFactory() + expansion = ExpansionFactory(collection_version=head, mnemonic='autoexpand-HEAD') + head.expansion_uri = expansion.uri + head.save() + head.add_references([CollectionReference( + system=source.uri, code=root.mnemonic, reference_type='concepts', created_by=head.created_by, + cascade={'method': SOURCE_TO_CONCEPTS})], head.created_by) + version = OrganizationCollectionFactory(organization=head.organization, version='v1', mnemonic=head.mnemonic) + return grandchild, version + + @patch('core.common.tasks.export_collection') + def test_rebuild_keeps_members_past_the_cascade_cap_as_creation_does(self, _): + grandchild, version = self.head_cascading_past_the_cap() + + seed_children_to_new_version('collection', version.id, False, expansion='lock') # pylint: disable=no-value-for-parameter + version.refresh_from_db() + expansion = version.expansion + self.assertGreater(expansion.concepts.count(), 1000) + self.assertTrue(expansion.concepts.filter(versioned_object_id=grandchild.versioned_object_id).exists()) + + expansion.seed_children(index=False, force_reevaluate=True) + + self.assertGreater(expansion.concepts.count(), 1000) + self.assertTrue(expansion.concepts.filter(versioned_object_id=grandchild.versioned_object_id).exists()) + + @patch('core.common.tasks.export_collection') + def test_rebuild_cascades_without_a_result_cap_as_creation_does(self, _): + source = OrganizationSourceFactory() + concept = ConceptFactory(parent=source, mnemonic='root') + head = OrganizationCollectionFactory() + expansion = ExpansionFactory(collection_version=head, mnemonic='autoexpand-HEAD') + head.expansion_uri = expansion.uri + head.save() + head.add_references([CollectionReference( + system=source.uri, code=concept.mnemonic, reference_type='concepts', created_by=head.created_by, + cascade={'method': SOURCE_TO_CONCEPTS})], head.created_by) + version = OrganizationCollectionFactory(organization=head.organization, version='v1', mnemonic=head.mnemonic) + cascade = Concept.cascade + + with patch.object(Concept, 'cascade', autospec=True, side_effect=cascade) as cascade_mock: + seed_children_to_new_version('collection', version.id, False, expansion='lock') # pylint: disable=no-value-for-parameter + version.refresh_from_db() + created = [call.kwargs.get('max_results', 1000) for call in cascade_mock.call_args_list] + cascade_mock.reset_mock() + version.expansion.seed_children(index=False, force_reevaluate=True) + rebuilt = [call.kwargs.get('max_results', 1000) for call in cascade_mock.call_args_list] + + self.assertTrue(created) + self.assertEqual(set(created), {None}) + self.assertEqual(set(rebuilt), {None}) + + @patch('core.common.tasks.export_collection') + def test_rebuild_keeps_the_resource_versions_a_head_resolved_version_was_created_with(self, _): + for choice in ('lock', 'latest'): + with self.subTest(choice=choice): + _, concept, collection_v1 = self.head_with_head_concept_member() + seed_children_to_new_version('collection', collection_v1.id, False, expansion=choice) # pylint: disable=no-value-for-parameter + collection_v1.refresh_from_db() + latest = concept.get_latest_version() + expansion = collection_v1.expansion + self.assertEqual(list(expansion.concepts.all()), [latest]) + + expansion.seed_children(index=False, force_reevaluate=True) + + self.assertEqual(list(expansion.concepts.all()), [latest]) + self.assertEqual(list(collection_v1.references.first().concepts.all()), [latest]) + + with patch.object(CollectionReference, 'is_expansion_locked_repo_version_on', return_value=False): + expansion.seed_children(index=False, force_reevaluate=True) + self.assertEqual(list(expansion.concepts.all()), [concept]) + @patch('core.common.tasks.export_collection') def test_seed_children_task_evaluates_references_while_head_expansion_is_processing(self, _): expansion, concept, collection_v1 = self.head_with_head_concept_member() @@ -2253,7 +2418,7 @@ def test_seed_children_task_evaluates_references_while_head_expansion_is_process latest = concept.get_latest_version() copy = collection_v1.expansion self.assertFalse(copy.is_processing) - self.assertEqual(copy.parameters, {'activeOnly': True}) + self.assertEqual(copy.parameters, {'activeOnly': True, 'system-version': f'{concept.parent.uri}|HEAD'}) self.assertEqual(list(copy.concepts.all()), [latest]) self.assertEqual(list(collection_v1.references.first().concepts.all()), [latest]) @@ -2275,7 +2440,7 @@ def test_seed_children_task_evaluates_references_while_a_removal_awaits_its_read with override_settings(TEST_MODE=False), self.captureOnCommitCallbacks(execute=True): head.delete_references([shared.uri]) self.assertTrue(readd_mock.apply_async.called) - self.assertTrue(expansion.has_pending_readd) + self.assertTrue(Task.unfinished_readd_references_tasks().filter(args__0=expansion.id).exists()) self.assertEqual(list(expansion.concepts.values_list('mnemonic', flat=True)), ['other']) collection_v1 = OrganizationCollectionFactory( @@ -2287,18 +2452,18 @@ def test_seed_children_task_evaluates_references_while_a_removal_awaits_its_read sorted(collection_v1.expansion.concepts.values_list('id', flat=True)), sorted([shared.get_latest_version().id, other.get_latest_version().id])) - def test_has_pending_readd_until_the_readd_succeeds(self): + def test_readd_stays_unfinished_until_it_succeeds(self): expansion = ExpansionFactory(collection_version=OrganizationCollectionFactory()) task = Task.new( user=expansion.collection_version.created_by, name='readd_references_to_expansion_on_references_removal', args=[expansion.id, [1]]) - self.assertTrue(expansion.has_pending_readd) + self.assertTrue(Task.unfinished_readd_references_tasks().filter(args__0=expansion.id).exists()) Task.objects.filter(id=task.id).update(state='FAILURE') - self.assertTrue(expansion.has_pending_readd) + self.assertTrue(Task.unfinished_readd_references_tasks().filter(args__0=expansion.id).exists()) Task.objects.filter(id=task.id).update(state='SUCCESS') - self.assertFalse(expansion.has_pending_readd) + self.assertFalse(Task.unfinished_readd_references_tasks().filter(args__0=expansion.id).exists()) def test_successful_readd_clears_only_older_unfinished_readds_of_that_expansion(self): expansion = ExpansionFactory(collection_version=OrganizationCollectionFactory()) @@ -2562,9 +2727,68 @@ def test_persist_new_version_sync_seeds_inline(self, seed_mock): with self.captureOnCommitCallbacks(execute=False) as callbacks: Collection.persist_new_version(collection_v1, collection.created_by, sync=True) - seed_mock.assert_called_once_with('collection', collection_v1.id, True, True) + seed_mock.assert_called_once_with('collection', collection_v1.id, True, True, expansion='latest') self.assertEqual(callbacks, []) + @override_settings(TEST_MODE=False) + @patch('core.common.models.handle_m2m_changed', Mock()) + @patch('core.common.models.handle_save', Mock()) + @patch('core.common.models.seed_children_to_new_version.apply_async') + def test_persist_new_version_records_the_choice_with_the_task_not_in_its_message(self, apply_async): + collection = OrganizationCollectionFactory() + collection_v1 = Collection(version='v1', mnemonic=collection.mnemonic, organization=collection.organization) + + with self.captureOnCommitCallbacks(execute=True): + Collection.persist_new_version(collection_v1, collection.created_by, expansion='latest') + + task = Task.objects.get(name='seed_children_to_new_version') + self.assertEqual(task.args, ['collection', collection_v1.id, True, False]) + self.assertEqual(task.kwargs, {'expansion': 'latest'}) + apply_async.assert_called_once_with( + ('collection', collection_v1.id, True, False), task_id=task.id, queue='default', persist_args=True) + + @patch('core.common.models.handle_m2m_changed', Mock()) + @patch('core.common.models.handle_save', Mock()) + @patch('core.common.models.seed_children_to_new_version.apply_async') + def test_persist_new_version_settles_the_default_before_the_task_runs(self, _): + _, _, _, _, head, _ = self.head_locked_to_v1() + collection_v2 = Collection(version='v2', mnemonic=head.mnemonic, organization=head.organization) + + # only the queuing runs as outside tests, and its index tasks are mocked, so nothing reaches Redis + with override_settings(TEST_MODE=False), self.captureOnCommitCallbacks(execute=True): + Collection.persist_new_version(collection_v2, head.created_by) + + task = Task.objects.get(name='seed_children_to_new_version', args__1=collection_v2.id) + self.assertEqual(task.kwargs, {'expansion': 'lock'}) + + Collection.objects.filter(id=head.id).update(expansion_uri=None) + with patch.object(Collection, 'seed_from_head_expansion', autospec=True, return_value=True) as seed_mock, \ + patch('core.common.tasks.export_collection'): + seed_children_to_new_version.push_request(id=task.id) + try: + seed_children_to_new_version.run('collection', collection_v2.id, False) + finally: + seed_children_to_new_version.pop_request() + + seed_mock.assert_called_once_with(ANY, lock=True) + + @patch('core.common.tasks.export_collection') + def test_seed_children_task_reads_the_choice_from_its_task(self, _): + _, _, _, v2, _, version = self.head_locked_to_v1() + task = Task.new(user=version.created_by, name='seed_children_to_new_version', kwargs={'expansion': 'latest'}) + Task.before_start(task.id, ['collection', version.id, False], {}, name='seed_children_to_new_version') + + seed_children_to_new_version.push_request(id=task.id) + try: + seed_children_to_new_version.run('collection', version.id, False) + finally: + seed_children_to_new_version.pop_request() + + version.refresh_from_db() + self.assertEqual(list(version.expansion.evaluated_source_versions.all()), [v2]) + task.refresh_from_db() + self.assertEqual(task.kwargs, {'expansion': 'latest'}) + @override_settings(TEST_MODE=False) def test_persist_new_version_retries_serialization_failures_for_overlapping_versions(self): collection = OrganizationCollectionFactory() @@ -2579,7 +2803,7 @@ def conflict_until_last_attempt(*_, **__): if len(calls) < SERIALIZATION_RETRY_ATTEMPTS: raise failure - with patch.object(Collection, 'can_seed_from_head_expansion', return_value=True), \ + with patch.object(Collection, 'get_expansion_choice', return_value='lock'), \ patch('core.common.models.transaction.get_connection', return_value=Mock(in_atomic_block=False)), \ patch.object(Collection, 'persist_new_version_in_transaction', side_effect=conflict_until_last_attempt): errors = Collection.persist_new_version(collection_v1, collection.created_by, sync=True) @@ -2595,7 +2819,7 @@ def test_persist_new_version_gives_up_after_the_serialization_retry_attempts(sel failure = OperationalError('could not serialize access') failure.__cause__ = serialization_cause() - with patch.object(Collection, 'can_seed_from_head_expansion', return_value=True), \ + with patch.object(Collection, 'get_expansion_choice', return_value='lock'), \ patch('core.common.models.transaction.get_connection', return_value=Mock(in_atomic_block=False)), \ patch.object( Collection, 'persist_new_version_in_transaction', side_effect=failure) as in_transaction: @@ -3245,7 +3469,7 @@ def test_delete_references_not_test_mode_queues_readd_task(self, batch_index_moc name='core.common.tasks.readd_references_to_expansion_on_references_removal', args__0=expansion.id) readd_task_mock.apply_async.assert_called_once_with( (expansion.id, [reference.id]), queue='default', task_id=task.id, persist_args=True) - self.assertTrue(expansion.has_pending_readd) + self.assertTrue(Task.unfinished_readd_references_tasks().filter(args__0=expansion.id).exists()) @override_settings(TEST_MODE=False) @patch('core.collections.models.readd_references_to_expansion_on_references_removal') @@ -6204,6 +6428,34 @@ def test_cascade_keeps_an_explicit_global_namespace_on_generated_references(self self.assertTrue(all(expression.startswith(self.other_source.uri) for expression in expressions), expressions) self.assertEqual(set(self.collection.references.values_list('namespace', flat=True)), {'/'}) + @patch('core.common.tasks.export_collection') + def test_lock_keeps_heads_version_when_the_owner_registry_maps_its_canonical_url_elsewhere(self, _): + self.set_registry_toggle(True) + self.add_owner_registry_entry() + GlobalURLRegistryFactory(url=self.URL, namespace=self.other.uri) + concept = ConceptFactory(parent=self.other_source, mnemonic='c1') + old = concept.get_latest_version() + new = ConceptFactory(parent=self.other_source, mnemonic='c1', versioned_object=concept, version='new') + old.sources.remove(self.other_source) + old.sources.add(self.other_source_v1) + other_v2 = OrganizationSourceFactory( + organization=self.other, mnemonic='cs', version='v2', canonical_url=self.URL, released=True) + new.sources.add(other_v2) + Source.objects.filter(id=self.other_source_v1.id).update(released=True) + self.collection.expansion_uri = self.expansion.uri + self.collection.save() + self.expansion.evaluated_source_versions.add(self.other_source_v1) + self.collection.add_expressions({'expressions': [concept.uri]}, self.collection.created_by) + self.assertEqual(list(self.collection.expansion.concepts.all()), [old]) + version = OrganizationCollectionFactory(organization=self.owner, mnemonic='vs', version='v1') + + seed_children_to_new_version('collection', version.id, False, expansion='lock') # pylint: disable=no-value-for-parameter + + version.refresh_from_db() + self.assertEqual(version.expansion.parameters['system-version'], f'{self.other_source.uri}|v1') + self.assertEqual(list(version.expansion.concepts.all()), [old]) + self.assertEqual(list(version.expansion.evaluated_source_versions.all()), [self.other_source_v1]) + def test_get_system_version_conflicts(self): self.set_registry_toggle(True) self.add_owner_registry_entry() diff --git a/core/common/models.py b/core/common/models.py index de6aa7dd8..041a7ac0d 100644 --- a/core/common/models.py +++ b/core/common/models.py @@ -1251,6 +1251,7 @@ def persist_new_version(cls, obj, user=None, **kwargs): obj.is_active = True sync = kwargs.pop('sync', False) + expansion = kwargs.pop('expansion', None) if user: obj.created_by = user obj.updated_by = user @@ -1263,11 +1264,12 @@ def persist_new_version(cls, obj, user=None, **kwargs): is_test_mode = get(settings, 'TEST_MODE', False) # a synchronous seed from HEAD runs in this transaction, so it needs this transaction's snapshot is_repeatable_read = sync and not is_test_mode and not transaction.get_connection().in_atomic_block and \ - obj.can_seed_from_head_expansion() + bool(obj.get_expansion_choice(expansion)) attempts = SERIALIZATION_RETRY_ATTEMPTS if is_repeatable_read else 1 for attempt in range(1, attempts + 1): try: - cls.persist_new_version_in_transaction(obj, head, user, sync, is_repeatable_read, **kwargs) + cls.persist_new_version_in_transaction( + obj, head, user, sync, is_repeatable_read, expansion, **kwargs) break except OperationalError as ex: if attempt == attempts or not cls.is_serialization_failure(ex): @@ -1279,7 +1281,8 @@ def persist_new_version(cls, obj, user=None, **kwargs): return errors @classmethod - def persist_new_version_in_transaction(cls, obj, head, user, sync, is_repeatable_read, **kwargs): # pylint: disable=too-many-arguments + def persist_new_version_in_transaction( # pylint: disable=too-many-arguments + cls, obj, head, user, sync, is_repeatable_read, expansion=None, **kwargs): is_test_mode = get(settings, 'TEST_MODE', False) with transaction.atomic(): if is_repeatable_read: @@ -1289,20 +1292,24 @@ def persist_new_version_in_transaction(cls, obj, head, user, sync, is_repeatable head = cls.objects.select_for_update().get(id=head.id) obj.update_version_data(head) obj.save(**kwargs) + # the default is settled while HEAD is locked, so a queued task can't see another HEAD + expansion = obj.get_expansion_choice(expansion) or expansion if obj.id: obj.sibling_versions.update(is_latest_version=False) task_args = (obj.resource_type.lower(), obj.id, not is_test_mode, sync) if is_test_mode or sync: - seed_children_to_new_version(*task_args) + seed_children_to_new_version(*task_args, expansion=expansion) else: from core.tasks.models import Task + # the choice rides on the task record, so a worker on older code still takes the message task = Task.new( queue='default', user=user, name=seed_children_to_new_version.__name__, args=task_args, + kwargs={'expansion': expansion} if expansion else None, ) # Mark the version as processing before the response goes out @@ -1326,8 +1333,8 @@ def is_serialization_failure(ex): def index_resources_for_self_as_latest_released(self, only_update=False): # pylint: disable=unused-argument pass - def can_seed_from_head_expansion(self): - return False + def get_expansion_choice(self, expansion=None): # pylint: disable=unused-argument + return None @classmethod def persist_changes(cls, obj, updated_by, original_schema, **kwargs): # pylint: disable=too-many-locals,too-many-branches diff --git a/core/common/tasks.py b/core/common/tasks.py index d3a02976c..41d3be50f 100644 --- a/core/common/tasks.py +++ b/core/common/tasks.py @@ -24,6 +24,7 @@ from core.celery import app from core.common import ERRBIT_LOGGER +from core.collections.constants import VERSION_EXPANSION_LOCK from core.common.constants import CONFIRM_EMAIL_ADDRESS_MAIL_SUBJECT, PASSWORD_RESET_MAIL_SUBJECT from core.common.exceptions import BatchIndexingError from core.common.utils import write_export_file, web_url, get_resource_class_from_resource_name, get_export_service, \ @@ -384,7 +385,7 @@ def send_user_reset_password_email(user_id): @app.task(bind=True) -def seed_children_to_new_version(self, resource, obj_id, export=True, sync=False): # pylint: disable=too-many-arguments,too-many-locals,too-many-branches,too-many-statements +def seed_children_to_new_version(self, resource, obj_id, export=True, sync=False, expansion=None): # pylint: disable=too-many-arguments,too-many-locals,too-many-branches,too-many-statements export_task = None changelog_task = None autoexpand = True @@ -418,8 +419,10 @@ def seed_children_to_new_version(self, resource, obj_id, export=True, sync=False instance.snapshot = CollectionDetailSerializer(head).data instance.save(update_fields=['snapshot']) instance.get_checksums(recalculate=True) - seed_from_head = is_collection and autoexpand and instance.can_seed_from_head_expansion() and \ - instance.seed_from_head_expansion() + if is_collection and not expansion and task_id: + expansion = get(Task.objects.filter(id=task_id).values_list('kwargs', flat=True).first(), 'expansion') + choice = instance.get_expansion_choice(expansion) if is_collection else None + seed_from_head = bool(choice) and instance.seed_from_head_expansion(lock=choice == VERSION_EXPANSION_LOCK) if not seed_from_head: instance.seed_references() if is_source: diff --git a/core/integration_tests/tests_collections.py b/core/integration_tests/tests_collections.py index 868afa368..3c213875e 100644 --- a/core/integration_tests/tests_collections.py +++ b/core/integration_tests/tests_collections.py @@ -3311,6 +3311,65 @@ def test_post_201(self): self.assertEqual(expansion.concepts.count(), 1) self.assertEqual(expansion.mappings.count(), 0) + def post_version(self, version, **payload): + return self.client.post( + self.collection.uri + 'versions/', {'id': version, 'description': 'version', **payload}, + HTTP_AUTHORIZATION='Token ' + self.token, format='json') + + def test_post_201_expansion_choice(self): + seed_from_head = Collection.seed_from_head_expansion + cases = [ + (True, None, None, 'lock', True), (True, None, True, 'lock', True), (True, None, False, None, False), + (True, 'lock', None, 'lock', True), (True, 'latest', None, 'latest', True), + (True, 'none', None, None, False), (False, None, None, None, True), + (False, None, True, None, True), (False, None, False, None, False), + (False, 'lock', None, None, True), (False, 'latest', None, None, True), + (False, 'none', None, None, False), + ] + for index, (is_locked, choice, autoexpand, built, expands) in enumerate(cases): + with self.subTest(is_locked=is_locked, choice=choice, autoexpand=autoexpand), \ + patch.object(CollectionReference, 'is_expansion_locked_repo_version_on', return_value=is_locked), \ + patch.object( + Collection, 'seed_from_head_expansion', autospec=True, side_effect=seed_from_head + ) as seed_mock: + payload = {'expansion': choice} if choice else {} + if autoexpand is not None: + payload['autoexpand'] = autoexpand + response = self.post_version(f'v{index}', **payload) + + self.assertEqual(response.status_code, 201) + version = self.collection.versions.get(version=f'v{index}') + if built: + seed_mock.assert_called_once_with(version, lock=built == 'lock') + else: + seed_mock.assert_not_called() + self.assertEqual(version.autoexpand, expands) + self.assertEqual(version.references.count(), 1) + if expands: + self.assertEqual(version.expansions.count(), 1) + self.assertEqual(version.expansion.concepts.count(), 1) + else: + self.assertEqual(version.expansions.count(), 0) + self.assertIsNone(version.expansion_uri) + + def test_post_201_defaults_to_latest_without_head_expansion(self): + Collection.objects.filter(id=self.collection.id).update(expansion_uri=None) + + with patch.object(Collection, 'seed_from_head_expansion', autospec=True, return_value=True) as seed_mock: + response = self.post_version('v1') + + self.assertEqual(response.status_code, 201) + seed_mock.assert_called_once_with(self.collection.versions.get(version='v1'), lock=False) + + def test_post_400_expansion_choice_invalid(self): + response = self.client.post( + self.collection.uri + 'versions/', {'id': 'v1', 'description': 'version1', 'expansion': 'rebuild'}, + HTTP_AUTHORIZATION='Token ' + self.token, format='json') + + self.assertEqual(response.status_code, 400) + self.assertIn('expansion', response.data) + self.assertEqual(self.collection.versions.count(), 1) + def test_post_201_uses_head_name_when_older_version_exists(self): # Regression for the same class of bug fixed on SourceVersionListView.create(): resolving # "head_object" via get_queryset().first() (ordered by -created_at, no version filter) would diff --git a/core/tasks/models.py b/core/tasks/models.py index c3ecf021b..fb072fd66 100644 --- a/core/tasks/models.py +++ b/core/tasks/models.py @@ -121,7 +121,7 @@ def record_exception(self, exception): return self def save_common(self, args, kwargs, einfo=None): # pylint: disable=unused-argument - self.kwargs = kwargs + self.kwargs = kwargs or self.kwargs if einfo: self.record_exception(einfo) @@ -159,7 +159,8 @@ def before_start(cls, task_id, args, kwargs, name=None): # pylint: disable=unus task.name = name or task.name task.state = STARTED task.queue = kwargs.get('queue', None) or task.queue or 'default' - task.kwargs = kwargs + # a message without kwargs keeps the ones recorded with the task + task.kwargs = kwargs or task.kwargs task.started_at = timezone.now() task.save() return task