feat(import): domain reconciliation + ontos* round-trip on ODCS/ODPS import (#851)

src/backend/src/common/authorization.py

@@ -649,6 +649,30 @@ async def enforce_feature_permission(
         )
 
 
+def user_has_feature_level(auth_manager, user, feature_id: str, required_level: FeatureAccessLevel) -> bool:
+    """Imperative, group-based permission check for conditional in-handler gates.
+
+    Mirrors the group-based path of :class:`PermissionChecker` (without the team/applied
+    role overrides) for secondary checks such as the import "create missing domains"
+    toggle, which must require ``data-domains`` write on top of the entity's own
+    write permission.
+
+    Caveats (callers rely on both):
+    - **Group-based only.** It does NOT consult team role overrides or an explicit
+      applied-role override. Use it for cross-feature secondary gates where the
+      user's group memberships are the source of truth.
+    - **Fails closed.** Any exception during resolution is caught and ``False``
+      is returned (with a warning log), so a transient error denies the extra
+      capability rather than silently granting it.
+    """
+    try:
+        effective = auth_manager.get_user_effective_permissions(getattr(user, "groups", None) or [], None)
+        return auth_manager.has_permission(effective, feature_id, required_level)
+    except Exception:
+        logger.warning("user_has_feature_level check failed for feature '%s'; denying.", feature_id, exc_info=True)
+        return False
+
+
 class PermissionChecker:
     """FastAPI Dependency to check user permissions for a feature."""
     def __init__(self, feature_id: str, required_level: FeatureAccessLevel):

src/backend/src/controller/data_contracts_manager.py

@@ -61,7 +61,7 @@ def _is_valid_uuid(value: str) -> bool:
 from src.repositories.data_contracts_repository import data_contract_repo
 from src.repositories.teams_repository import team_repo
 from src.repositories.entity_domain_association_repository import entity_domain_repo
-from src.controller.domain_export_adapter import domain_export_adapter
+from src.controller.domain_export_adapter import domain_export_adapter, ONTOS_ORIGINAL_DOMAIN_PROPERTY
 
 from src.common.logging import get_logger
 from src.common.delivery_mixin import DeliveryMixin
@@ -963,23 +963,20 @@ def build_odcs_from_db(self, db_obj: DataContractDb, db_session=None) -> Dict[st
                             prop_dict['transformLogic'] = prop.transform_logic
                         if prop.transform_source_objects:
                             try:
-                                import json
                                 prop_dict['transformSourceObjects'] = json.loads(prop.transform_source_objects)
                             except (json.JSONDecodeError, TypeError):
                                 prop_dict['transformSourceObjects'] = prop.transform_source_objects
                         if prop.transform_description:
                             prop_dict['description'] = prop.transform_description
                         if prop.examples:
                             try:
-                                import json
                                 prop_dict['examples'] = json.loads(prop.examples)
                             except (json.JSONDecodeError, TypeError):
                                 prop_dict['examples'] = prop.examples
                         if prop.critical_data_element is not None:
                             prop_dict['criticalDataElement'] = prop.critical_data_element
                         if prop.logical_type_options_json:
                             try:
-                                import json
                                 logical_type_options = json.loads(prop.logical_type_options_json)
                                 prop_dict.update(logical_type_options)  # Merge constraints into property
                             except (json.JSONDecodeError, TypeError):
@@ -1068,7 +1065,6 @@ def build_odcs_from_db(self, db_obj: DataContractDb, db_session=None) -> Dict[st
                                 prop_value = custom_prop.value
                                 try:
                                     # Try to parse JSON if it's a serialized object
-                                    import json
                                     prop_value = json.loads(custom_prop.value)
                                 except (json.JSONDecodeError, TypeError):
                                     pass  # Keep as string
@@ -1540,14 +1536,22 @@ def build_odcs_from_db(self, db_obj: DataContractDb, db_session=None) -> Dict[st
             odcs['servers'] = servers
 
         # Inject semantic assignments from EntitySemanticLinks
+        from contextlib import nullcontext
         from src.controller.semantic_links_manager import SemanticLinksManager
         from src.common.database import get_db_session
         from src.utils.semantic_helpers import get_semantic_assignment_type
 
         SEMANTIC_ASSIGNMENT_TYPE = get_semantic_assignment_type()
 
+        # Reuse the caller's session when one was passed, rather than opening a
+        # second session via get_db_session(). The read-only semantic lookups don't
+        # need their own session, and get_db_session() commits on exit — under the
+        # test harness's shared single SQLite connection that commit lands on the
+        # outer per-test transaction, defeating its rollback and leaking rows across
+        # tests; in production it was simply a redundant session + commit per export.
         try:
-            with get_db_session() as db:
+            session_cm = nullcontext(db_session) if db_session is not None else get_db_session()
+            with session_cm as db:
                 semantic_manager = SemanticLinksManager(db)
 
                 # Inject contract-level semantic assignments
@@ -1653,20 +1657,11 @@ def _resolve_domain(self, db, domain_id: Optional[str] = None, domain_name: Opti
                 domain_obj = data_domain_repo.get_by_name(db, name=domain_name)
                 if domain_obj:
                     return domain_obj.id
-                # Auto-create missing domain
-                try:
-                    from src.controller.data_domains_manager import DataDomainManager
-                    from src.models.data_domains import DataDomainCreate
-                    manager = DataDomainManager(repository=data_domain_repo)
-                    created_read = manager.create_domain(
-                        db,
-                        domain_in=DataDomainCreate(name=domain_name, description=None, owner=['system'], tags=[], parent_id=None),
-                        current_user_id='system'
-                    )
-                    return str(created_read.id)
-                except Exception as ce:
-                    logger.warning(f"Auto-create domain failed for '{domain_name}': {ce}")
-                    return None
+                # No match → leave unassigned (#851). The previous silent auto-create is
+                # gone; auto-create is now an explicit, opt-in import toggle handled in the
+                # reconciliation path (domain_export_adapter.parse_odcs(create_missing=...)).
+                logger.info("Domain name '%s' not found; leaving unassigned.", domain_name)
+                return None
         except ValueError:
             raise
         except Exception as e:
@@ -2890,19 +2885,22 @@ def create_from_upload(
         self,
         db,
         parsed_odcs: dict,
-        current_user: Optional[str] = None
+        current_user: Optional[str] = None,
+        create_missing_domains: bool = False,
     ) -> DataContractDb:
         """
         Create a contract from uploaded ODCS file. Manages transaction.
-        
+
         Args:
             db: Database session
             parsed_odcs: Parsed ODCS dictionary
             current_user: Username of current user
-
+            create_missing_domains: When True, auto-create by name any domain that
+                doesn't resolve (the opt-in "Create missing domains" import toggle, #851).
+
         Returns:
             Created DataContractDb instance
-            
+
         Raises:
             ValueError: If validation fails
             Exception: If creation fails
@@ -2928,9 +2926,26 @@ def create_from_upload(
             elif not isinstance(description, dict):
                 description = {}
 
-            # Resolve the domain assignment from the parsed ODCS payload (primary name +
-            # customProperties.additionalDomains, or app round-trip domainIds).
-            import_domain_ids, import_primary_domain_id = domain_export_adapter.parse_odcs(parsed_odcs, db)
+            # Reconcile the domain assignment from the parsed payload: best-effort ID→name
+            # (authoritative ontosDomainId key first, then legacy keys, then the standard
+            # `domain` name). No match → unassigned unless create_missing_domains is set (#851).
+            import_domain_ids, import_primary_domain_id = domain_export_adapter.parse_odcs(
+                parsed_odcs, db, create_missing=create_missing_domains, created_by=current_user,
+            )
+            # Always preserve the raw incoming domain string(s) as provenance so the original
+            # is never lost (and a later export can surface it via ontosOriginalDomain).
+            original_domains = domain_export_adapter.extract_original_domain_strings(parsed_odcs)
+            if original_domains:
+                cprops = parsed_odcs.get('customProperties')
+                if not isinstance(cprops, list):
+                    cprops = [] if cprops is None else cprops
+                if isinstance(cprops, list):
+                    cprops = [
+                        c for c in cprops
+                        if not (isinstance(c, dict) and c.get('property') == ONTOS_ORIGINAL_DOMAIN_PROPERTY)
+                    ]
+                    cprops.append({"property": ONTOS_ORIGINAL_DOMAIN_PROPERTY, "value": original_domains})
+                    parsed_odcs['customProperties'] = cprops
 
             # Try to resolve owner as team name
             owner_team_id = self._resolve_team_name_to_id(db, owner_val)
@@ -3124,6 +3139,7 @@ def create_contracts_from_files(
         db,
         files: List[tuple],
         current_user: Optional[str] = None,
+        create_missing_domains: bool = False,
     ) -> BatchImportResult:
         """Import ODCS contracts from one or more uploaded files.
 
@@ -3132,6 +3148,8 @@ def create_contracts_from_files(
             files: List of ``(filename, content_text, content_type)`` tuples. Each
                 file may contain a single ODCS object or an array of them.
             current_user: Username of the uploader.
+            create_missing_domains: opt-in "Create missing domains" import toggle (#851),
+                applied per-upload to every entity in the batch.
 
         Returns:
             A :class:`BatchImportResult`. Per-entity failures are captured as failed
@@ -3193,6 +3211,7 @@ def create_contracts_from_files(
                 try:
                     created = self.create_from_upload(
                         db=db, parsed_odcs=entity, current_user=current_user,
+                        create_missing_domains=create_missing_domains,
                     )
                     result.add(ImportItemResult(
                         index=index, source_file=filename, source_id=source_id,

src/backend/src/controller/data_products_manager.py

@@ -59,7 +59,7 @@ def _is_valid_uuid(value: str) -> bool:
 from src.repositories.data_products_repository import data_product_repo, subscription_repo
 from src.repositories.teams_repository import team_repo
 from src.repositories.entity_domain_association_repository import entity_domain_repo
-from src.controller.domain_export_adapter import domain_export_adapter
+from src.controller.domain_export_adapter import domain_export_adapter, ONTOS_ORIGINAL_DOMAIN_PROPERTY
 from src.repositories.genie_spaces_repository import genie_space_repo
 from src.models.genie_spaces import GenieSpaceCreate
 from src.common.search_interfaces import SearchableAsset, SearchIndexItem
@@ -144,13 +144,20 @@ def _attach_domains(self, api_obj, db: Optional[Session] = None, preloaded_domai
             logger.warning(f"Failed to attach domains for product {getattr(api_obj, 'id', '?')}: {e}")
         return api_obj
 
-    def _resolve_product_domain_assignment(self, data: dict, db: Optional[Session] = None) -> tuple:
+    def _resolve_product_domain_assignment(
+        self,
+        data: dict,
+        db: Optional[Session] = None,
+        create_missing: bool = False,
+        created_by: Optional[str] = None,
+    ) -> tuple:
         """Resolve a product payload to (domain_ids, primary_domain_id).
 
         Prefers domain_ids + primary_domain_id; falls back to the legacy single ``domain``
         (which may be a domain ID or name). Names are resolved via the data domain repo.
         Uses the caller's ``db`` session when provided so resolution sees rows written in
-        the same request transaction.
+        the same request transaction. When ``create_missing`` is set, unresolved names in
+        the round-trip/ODCS-standard path are auto-created (the opt-in import toggle, #851).
         """
         from src.repositories.data_domain_repository import data_domain_repo
         session = db if db is not None else self._db
@@ -185,7 +192,9 @@ def _resolve_one(value):
         #    ODCS-standard primary ``domain`` name + ``customProperties.additionalDomains``
         #    are all honoured. Reading only the single ``domain`` name here previously
         #    dropped additional domains on re-import (round-trip loss).
-        return domain_export_adapter.parse_odcs(data, session)
+        return domain_export_adapter.parse_odcs(
+            data, session, create_missing=create_missing, created_by=created_by,
+        )
 
     def get_statuses(self) -> List[str]:
         """Get all ODPS v1.0.0 status values."""
@@ -198,6 +207,7 @@ def create_product(
         user: Optional[str] = None,
         background_tasks: Optional[Any] = None,
         preserve_source_id: bool = False,
+        create_missing_domains: bool = False,
     ) -> DataProductApi:
         """Creates a new ODPS v1.0.0 data product via the repository.
 
@@ -208,6 +218,8 @@ def create_product(
             background_tasks: Optional FastAPI BackgroundTasks for async delivery
             preserve_source_id: If True, preserve the original non-UUID id as a
                 sourceId custom property (used during batch upload/import)
+            create_missing_domains: When True, auto-create by name any domain that doesn't
+                resolve (the opt-in "Create missing domains" import toggle, #851).
         """
         from src.controller.delivery_service import DeliveryChangeType
 
@@ -228,6 +240,23 @@ def create_product(
                     product_data['customProperties'] = custom_props
                     logger.info(f"Preserved original ID '{original_id}' as {SOURCE_ID_PROPERTY} custom property")
 
+            # Always preserve the raw incoming domain string(s) as provenance (#851), so the
+            # original is never lost and a later export can surface it via ontosOriginalDomain.
+            original_domains = domain_export_adapter.extract_original_domain_strings(product_data)
+            if original_domains:
+                cprops = product_data.get('customProperties')
+                if isinstance(cprops, list):
+                    cprops = [
+                        c for c in cprops
+                        if not (isinstance(c, dict) and c.get('property') == ONTOS_ORIGINAL_DOMAIN_PROPERTY)
+                    ]
+                    cprops.append({"property": ONTOS_ORIGINAL_DOMAIN_PROPERTY, "value": original_domains})
+                    product_data['customProperties'] = cprops
+                elif not cprops:
+                    product_data['customProperties'] = [
+                        {"property": ONTOS_ORIGINAL_DOMAIN_PROPERTY, "value": original_domains}
+                    ]
+
             # Always generate a UUID for the product ID
             product_data['id'] = str(uuid.uuid4())
             logger.info(f"Generated ID {product_data['id']} for new product.")
@@ -267,7 +296,9 @@ def create_product(
 
             # Assign domains via the junction table (accepts domain_ids/primary_domain_id
             # or legacy single `domain` id/name).
-            prod_domain_ids, prod_primary = self._resolve_product_domain_assignment(product_data, db=db_session)
+            prod_domain_ids, prod_primary = self._resolve_product_domain_assignment(
+                product_data, db=db_session, create_missing=create_missing_domains, created_by=user,
+            )
             if prod_domain_ids:
                 entity_domain_repo.set_domains_for_entity(
                     db_session, entity_type="data_product", entity_id=created_db_obj.id,
@@ -1650,13 +1681,16 @@ def create_products_from_files(
         self,
         files: List[tuple],
         user: Optional[str] = None,
+        create_missing_domains: bool = False,
     ) -> BatchImportResult:
         """Import ODPS products from one or more uploaded files.
 
         Args:
             files: List of ``(filename, content_bytes)`` tuples. Each file may
                 contain a single product object or an array of them.
             user: Username of the uploader (stamped as personal-draft owner).
+            create_missing_domains: opt-in "Create missing domains" import toggle (#851),
+                applied per-upload to every entity in the batch.
 
         Returns:
             A :class:`BatchImportResult`. A file that cannot be parsed at all is
@@ -1719,6 +1753,7 @@ def create_products_from_files(
                 try:
                     created = self.create_product(
                         product_data, user=user, preserve_source_id=True,
+                        create_missing_domains=create_missing_domains,
                     )
                     result.add(ImportItemResult(
                         index=index, source_file=filename, source_id=source_id,