Improve bulk import flow for hierarchy content regardless of concept import order (#852)

* fix(core/importers): build hierarchy after inline bulk import

* Track parent_concept_urls for non-delete Concept imports to record child-to-parent relationships before finalization
* Invert child-to-parent map and call make_hierarchy to attach newly imported concepts under their parents

* fix(importers): rebuild concept hierarchy after bulk import

* Capture child-to-parent relationships before clearing input list to preserve hierarchy data during import lifecycle
* Invert saved map after processing and call make_hierarchy to restore parent links for imported concepts

* fix(importers): normalize bulk import hierarchy URIs and guard reconciliation by edit access

* Encode concept_id when building child URIs to match stored concept URI format
* Skip hierarchy reconciliation for concepts without edit permissions to prevent unintended updates

* fix(importers): reconcile bulk import concept hierarchies after import

* Collect concept parent links during parallel bulk import execution for deferred reconciliation
* Apply hierarchy mapping after all import tasks complete to ensure correct parent-child ordering
* Remove inline hierarchy reconciliation from BulkImportInline to avoid partial updates
* Add regression tests for child-before-parent ordering, inaccessible concepts, non-concept records, and URI encoding

Author: filiperochalopes

core/common/utils.py

@@ -439,6 +439,11 @@ def queue_bulk_import(  # pylint: disable=too-many-arguments
             task_func = bulk_import_parallel_inline
             args = (to_import, username, update_if_exists, threads)
         else:
+            # TODO: bulk_import_inline is unreachable from the current view layer —
+            # import_response always passes `parallel_threads = request.data.get('parallel') or 5`
+            # which is always truthy, so the `elif threads` branch above is always taken.
+            # Consider removing bulk_import_inline and this branch once confirmed no external
+            # callers rely on it (check API consumers and the test suite first).
             task_func = bulk_import_inline
     else:
         task_func = bulk_import

core/importers/models.py

@@ -15,7 +15,7 @@
 from core.collections.models import Collection
 from core.common.constants import HEAD
 from core.common.tasks import bulk_import_parts_inline, delete_organization, batch_index_resources, \
-    post_import_update_resource_counts
+    post_import_update_resource_counts, make_hierarchy
 from core.common.utils import drop_version, is_url_encoded_string, encode_string, to_parent_uri, chunks
 from core.concepts.models import Concept
 from core.mappings.models import Mapping
@@ -968,11 +968,13 @@ def __init__(
         self.parts = deque([])
         self.result = None
         self._json_result = None
+        self.concept_hierarchy_map = {}  # child_uri -> [parent_uris], built before input_list is cleared
         if self.content:
             self.input_list = self.content if isinstance(self.content, list) else self.content.splitlines()
             self.total = len(self.input_list)
         self.make_resource_distribution()
         self.make_parts()
+        self.collect_concept_hierarchy_map()
         self.content = None  # memory optimization
         self.input_list = []  # memory optimization
 
@@ -1022,6 +1024,25 @@ def make_parts(self):
                     self.parts[-1].append(line)
                 prev_line = line
 
+    def collect_concept_hierarchy_map(self):
+        for data in self.input_list:
+            line = data if isinstance(data, dict) else json.loads(data)
+            if line.get('type', '').lower() != 'concept':
+                continue
+            parent_urls = line.get('parent_concept_urls') or []
+            concept_id = line.get('id')
+            owner = line.get('owner')
+            source = line.get('source')
+            if parent_urls and concept_id and owner and source:
+                owner_type = line.get('owner_type', '').lower()
+                owner_prefix = 'users' if owner_type in ['user', 'users'] else 'orgs'
+                # P2: normalize concept_id the same way ConceptImporter.parse() does,
+                # so the URI matches what was actually persisted in the database.
+                if not is_url_encoded_string(concept_id):
+                    concept_id = encode_string(concept_id)
+                child_uri = f'/{owner_prefix}/{owner}/sources/{source}/concepts/{concept_id}/'
+                self.concept_hierarchy_map[child_uri] = parent_urls
+
     @staticmethod
     def chunker_list(seq, size, is_child):  # pylint: disable=too-many-locals
         """
@@ -1127,6 +1148,30 @@ def run(self):
                             self.resource_wise_time[part_type] = 0
                         self.resource_wise_time[part_type] += round(time.time() - start_time, 4)
 
+        if self.concept_hierarchy_map:
+            # P1: restrict reconciliation to concepts the importing user can actually edit,
+            # mirroring the has_edit_access guard in ConceptImporter.process(). This excludes
+            # PERMISSION_DENIED rows and prevents hierarchy changes on sources the user does
+            # not own, even when those concepts already exist in the database.
+            user = UserProfile.objects.filter(username=self.username).first()
+            accessible_uris = set(
+                concept.uri
+                for concept in Concept.objects.filter(
+                    uri__in=self.concept_hierarchy_map.keys(), id=F('versioned_object_id')
+                ).select_related('parent')
+                if concept.parent.has_edit_access(user)
+            )
+            inverted = {}
+            for child_uri, parent_uris in self.concept_hierarchy_map.items():
+                if child_uri not in accessible_uris:
+                    continue
+                for parent_uri in parent_uris:
+                    if parent_uri not in inverted:
+                        inverted[parent_uri] = []
+                    inverted[parent_uri].append(child_uri)
+            if inverted:
+                make_hierarchy(inverted)
+
         post_import_update_resource_counts.apply_async(queue='default', permanent=False)
 
         self.update_elapsed_seconds()

core/importers/tests.py

@@ -1264,6 +1264,239 @@ def test_import_subtask_multiple_resource_per_file(self):
                                              'OCL', 'Concept', [{'start_index': 5, 'end_index': 10}]).run()
             self.assertEqual(concept_result, [1, 1, 1, 1, 1])
 
+    # ── collect_concept_hierarchy_map ────────────────────────────────────────
+
+    def test_collect_concept_hierarchy_map_basic(self):
+        """Concepts with parent_concept_urls are collected; those without are ignored."""
+        content = '\n'.join([
+            json.dumps({
+                "type": "Concept", "id": "ChildConcept",
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+                "parent_concept_urls": ["/orgs/TestOrg/sources/TestSource/concepts/ParentConcept/"]
+            }),
+            json.dumps({
+                "type": "Concept", "id": "ParentConcept",
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+            }),
+        ])
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+
+        self.assertEqual(
+            importer.concept_hierarchy_map,
+            {
+                '/orgs/TestOrg/sources/TestSource/concepts/ChildConcept/':
+                    ['/orgs/TestOrg/sources/TestSource/concepts/ParentConcept/']
+            }
+        )
+
+    def test_collect_concept_hierarchy_map_encodes_special_chars(self):
+        """Concept IDs with special characters (e.g. &) are URL-encoded to match persisted URIs (P2)."""
+        content = json.dumps({
+            "type": "Concept", "id": "1A40.0&XA8UM1",
+            "owner": "OpenMRS-OCL-Squad", "owner_type": "Organization", "source": "ICD-11-CIEL-Bridge",
+            "parent_concept_urls": ["/orgs/OpenMRS-OCL-Squad/sources/ICD-11-CIEL-Bridge/concepts/BlockL1-1A0/"]
+        })
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+
+        encoded_uri = '/orgs/OpenMRS-OCL-Squad/sources/ICD-11-CIEL-Bridge/concepts/1A40.0%26XA8UM1/'
+        raw_uri     = '/orgs/OpenMRS-OCL-Squad/sources/ICD-11-CIEL-Bridge/concepts/1A40.0&XA8UM1/'
+
+        self.assertIn(encoded_uri, importer.concept_hierarchy_map)
+        self.assertNotIn(raw_uri, importer.concept_hierarchy_map)
+
+    def test_collect_concept_hierarchy_map_user_owner_type(self):
+        """Concepts owned by a User (not an Org) use /users/ prefix in their URI."""
+        content = json.dumps({
+            "type": "Concept", "id": "MyConcept",
+            "owner": "johndoe", "owner_type": "User", "source": "MySource",
+            "parent_concept_urls": ["/users/johndoe/sources/MySource/concepts/RootConcept/"]
+        })
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+
+        self.assertIn('/users/johndoe/sources/MySource/concepts/MyConcept/', importer.concept_hierarchy_map)
+
+    def test_collect_concept_hierarchy_map_ignores_non_concepts(self):
+        """Only Concept lines are indexed; Source, Mapping, and Reference lines are ignored."""
+        content = '\n'.join([
+            json.dumps({"type": "Source", "id": "S1", "owner": "O", "owner_type": "Organization",
+                        "parent_concept_urls": ["/orgs/O/sources/S1/concepts/Root/"]}),
+            json.dumps({"type": "Mapping", "id": "M1", "owner": "O", "owner_type": "Organization",
+                        "source": "S1",
+                        "parent_concept_urls": ["/orgs/O/sources/S1/concepts/Root/"]}),
+            json.dumps({"type": "Concept", "id": "C1", "owner": "O", "owner_type": "Organization",
+                        "source": "S1"}),
+        ])
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+
+        self.assertEqual(importer.concept_hierarchy_map, {})
+
+    # ── run() hierarchy reconciliation ───────────────────────────────────────
+
+    @patch('core.importers.models.make_hierarchy')
+    @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive')
+    @patch('core.importers.models.BulkImportParallelRunner.queue_tasks')
+    def test_run_calls_make_hierarchy_with_inverted_map(self, queue_tasks_mock, wait_mock, make_hierarchy_mock):
+        """After all chunks complete, make_hierarchy receives the inverted {parent_uri: [child_uris]} map."""
+        org    = OrganizationFactory(mnemonic='TestOrg')
+        source = OrganizationSourceFactory(organization=org, mnemonic='TestSource', version='HEAD')
+        parent = ConceptFactory(parent=source, mnemonic='ParentConcept')
+        child  = ConceptFactory(parent=source, mnemonic='ChildConcept')
+
+        # File order: child before parent (the exact scenario the fix targets)
+        content = '\n'.join([
+            json.dumps({
+                "type": "Concept", "id": child.mnemonic,
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+                "parent_concept_urls": [parent.uri]
+            }),
+            json.dumps({
+                "type": "Concept", "id": parent.mnemonic,
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+            }),
+        ])
+
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+        importer.run()
+
+        make_hierarchy_mock.assert_called_once()
+        inverted = make_hierarchy_mock.call_args[0][0]
+        self.assertIn(parent.uri, inverted)
+        self.assertIn(child.uri, inverted[parent.uri])
+
+    @patch('core.importers.models.make_hierarchy')
+    @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive')
+    @patch('core.importers.models.BulkImportParallelRunner.queue_tasks')
+    def test_run_skips_make_hierarchy_when_no_hierarchy(self, queue_tasks_mock, wait_mock, make_hierarchy_mock):
+        """make_hierarchy is not called when no concept in the import has parent_concept_urls."""
+        content = '\n'.join([
+            json.dumps({
+                "type": "Concept", "id": "StandaloneConcept",
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+            }),
+        ])
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+        importer.run()
+
+        make_hierarchy_mock.assert_not_called()
+
+    @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive')
+    @patch('core.importers.models.BulkImportParallelRunner.queue_tasks')
+    def test_run_hierarchy_child_before_parent(self, queue_tasks_mock, wait_mock):
+        """
+        End-to-end: when child appears before parent in the import file, the reconciliation
+        step must correctly establish the parent_concepts M2M link in the database.
+        This is the primary bug scenario fixed by this PR.
+        """
+        org    = OrganizationFactory(mnemonic='TestOrg')
+        source = OrganizationSourceFactory(organization=org, mnemonic='TestSource', version='HEAD')
+        parent = ConceptFactory(parent=source, mnemonic='ParentConcept')
+        child  = ConceptFactory(parent=source, mnemonic='ChildConcept')
+
+        # child listed before parent — the exact ordering that broke hierarchy before the fix
+        content = '\n'.join([
+            json.dumps({
+                "type": "Concept", "id": child.mnemonic,
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+                "parent_concept_urls": [parent.uri]
+            }),
+            json.dumps({
+                "type": "Concept", "id": parent.mnemonic,
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+            }),
+        ])
+
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+        importer.run()
+
+        child.refresh_from_db()
+        self.assertIn(parent.get_latest_version(), child.parent_concepts.all())
+
+    @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive')
+    @patch('core.importers.models.BulkImportParallelRunner.queue_tasks')
+    def test_run_hierarchy_parent_before_child(self, queue_tasks_mock, wait_mock):
+        """
+        End-to-end: when parent appears before child (natural order), the reconciliation
+        must also establish the link correctly — confirming the fix is order-agnostic.
+        """
+        org    = OrganizationFactory(mnemonic='TestOrg')
+        source = OrganizationSourceFactory(organization=org, mnemonic='TestSource', version='HEAD')
+        parent = ConceptFactory(parent=source, mnemonic='ParentConcept')
+        child  = ConceptFactory(parent=source, mnemonic='ChildConcept')
+
+        # parent listed before child — normal/expected order
+        content = '\n'.join([
+            json.dumps({
+                "type": "Concept", "id": parent.mnemonic,
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+            }),
+            json.dumps({
+                "type": "Concept", "id": child.mnemonic,
+                "owner": "TestOrg", "owner_type": "Organization", "source": "TestSource",
+                "parent_concept_urls": [parent.uri]
+            }),
+        ])
+
+        importer = BulkImportParallelRunner(content, 'ocladmin', True)
+        importer.run()
+
+        child.refresh_from_db()
+        self.assertIn(parent.get_latest_version(), child.parent_concepts.all())
+
+    @patch('core.importers.models.make_hierarchy')
+    @patch('core.importers.models.BulkImportParallelRunner.wait_till_tasks_alive')
+    @patch('core.importers.models.BulkImportParallelRunner.queue_tasks')
+    def test_run_excludes_inaccessible_concepts(self, queue_tasks_mock, wait_mock, make_hierarchy_mock):
+        """
+        Concepts from sources the importing user cannot edit must be excluded from make_hierarchy,
+        even when they exist in the database. This prevents hierarchy changes on foreign sources
+        that were denied during import (P1 security fix).
+        """
+        # OrgA — importing user is a member → has edit access
+        org_a    = OrganizationFactory(mnemonic='OrgA')
+        source_a = OrganizationSourceFactory(organization=org_a, mnemonic='SourceA', version='HEAD')
+        parent_a = ConceptFactory(parent=source_a, mnemonic='ParentA')
+        child_a  = ConceptFactory(parent=source_a, mnemonic='ChildA')
+
+        # OrgB — importing user is NOT a member and source is not publicly editable → no edit access
+        org_b    = OrganizationFactory(mnemonic='OrgB')
+        source_b = OrganizationSourceFactory(organization=org_b, mnemonic='SourceB', version='HEAD', public_access='None')
+        parent_b = ConceptFactory(parent=source_b, mnemonic='ParentB')
+        child_b  = ConceptFactory(parent=source_b, mnemonic='ChildB')
+
+        importing_user = UserProfileFactory(username='importer-user')
+        org_a.members.add(importing_user)
+        self.assertFalse(org_b.is_member(importing_user))
+
+        content = '\n'.join([
+            # accessible concept (OrgA)
+            json.dumps({
+                "type": "Concept", "id": child_a.mnemonic,
+                "owner": "OrgA", "owner_type": "Organization", "source": "SourceA",
+                "parent_concept_urls": [parent_a.uri]
+            }),
+            # inaccessible concept (OrgB — user has no permission)
+            json.dumps({
+                "type": "Concept", "id": child_b.mnemonic,
+                "owner": "OrgB", "owner_type": "Organization", "source": "SourceB",
+                "parent_concept_urls": [parent_b.uri]
+            }),
+        ])
+
+        importer = BulkImportParallelRunner(content, importing_user.username, True)
+        importer.run()
+
+        make_hierarchy_mock.assert_called_once()
+        inverted = make_hierarchy_mock.call_args[0][0]
+
+        # OrgA child must be linked to its parent
+        self.assertIn(parent_a.uri, inverted)
+        self.assertIn(child_a.uri, inverted[parent_a.uri])
+
+        # OrgB child must NOT appear — user has no access to SourceB
+        self.assertNotIn(parent_b.uri, inverted)
+        all_children = [uri for uris in inverted.values() for uri in uris]
+        self.assertNotIn(child_b.uri, all_children)
+
 
 class BulkImportViewTest(OCLAPITestCase):
     def setUp(self):