diff --git a/cognite_toolkit/_cdf_tk/apps/_migrate_app.py b/cognite_toolkit/_cdf_tk/apps/_migrate_app.py index 744bb636a3..647cd4ea77 100644 --- a/cognite_toolkit/_cdf_tk/apps/_migrate_app.py +++ b/cognite_toolkit/_cdf_tk/apps/_migrate_app.py @@ -50,8 +50,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: self.command("timeseries")(self.timeseries) self.command("files")(self.files) self.command("canvas")(self.canvas) - # Uncomment when infield v2 config migration is ready - # self.command("infield-configs")(self.infield_configs) + self.command("infield-configs")(self.infield_configs) def main(self, ctx: typer.Context) -> None: """Migrate resources from Asset-Centric to data modeling in CDF.""" diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/command.py b/cognite_toolkit/_cdf_tk/commands/_migrate/command.py index cc3244ad26..346268dd79 100644 --- a/cognite_toolkit/_cdf_tk/commands/_migrate/command.py +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/command.py @@ -259,15 +259,65 @@ def create( verbose: bool = False, ) -> DeployResults: """This method is used to create migration resource in CDF.""" - self.validate_migration_model_available(client) + # Only validate migration model if the creator uses lineage/mapping + # InfieldV2ConfigCreator doesn't use the migration model, so skip validation + if creator.HAS_LINEAGE: + self.validate_migration_model_available(client) deploy_cmd = DeployCommand(self.print_warning, silent=self.silent) deploy_cmd.tracker = self.tracker + results = DeployResults([], "deploy", dry_run=dry_run) + + # Special handling for InfieldV2ConfigCreator which creates two different resource types + from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator + from cognite_toolkit._cdf_tk.cruds import LocationFilterCRUD + + if isinstance(creator, InfieldV2ConfigCreator): + # Deploy LocationFilters using Location Filters API + location_filters = creator.create_location_filters() + if location_filters: + location_filter_crud = LocationFilterCRUD.create_loader(client) + location_filter_worker = ResourceWorker(location_filter_crud, "deploy") + location_filter_by_id = { + location_filter_crud.get_id(item): (item.dump(), item) for item in location_filters + } + location_filter_worker.validate_access(location_filter_by_id, is_dry_run=dry_run) + cdf_location_filters = location_filter_crud.retrieve(list(location_filter_by_id.keys())) + location_filter_resources = location_filter_worker.categorize_resources( + location_filter_by_id, cdf_location_filters, False, verbose + ) + + if dry_run: + location_filter_result = deploy_cmd.dry_run_deploy( + location_filter_resources, location_filter_crud, False, False + ) + else: + location_filter_result = deploy_cmd.actual_deploy(location_filter_resources, location_filter_crud) + + verb = "Would deploy" if dry_run else "Deploying" + self.console(f"{verb} {len(location_filters)} location filters to CDF.") + + location_filter_configs = creator.location_filter_configs(location_filters) + for config in location_filter_configs: + filepath = ( + output_dir + / location_filter_crud.folder_name + / f"{sanitize_filename(config.filestem)}.{location_filter_crud.kind}.yaml" + ) + filepath.parent.mkdir(parents=True, exist_ok=True) + safe_write(filepath, yaml_safe_dump(config.data)) + self.console( + f"{len(location_filter_configs)} {location_filter_crud.kind} resource configurations written to {(output_dir / location_filter_crud.folder_name).as_posix()!r}" + ) + + if location_filter_result: + results[location_filter_result.name] = location_filter_result + + # Deploy InFieldLocationConfig nodes using Data Modeling Instance API crud_cls = creator.CRUD resource_list = creator.create_resources() - results = DeployResults([], "deploy", dry_run=dry_run) crud = crud_cls.create_loader(client) worker = ResourceWorker(crud, "deploy") local_by_id = {crud.get_id(item): (item.dump(), item) for item in resource_list} diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/creators.py b/cognite_toolkit/_cdf_tk/commands/_migrate/creators.py index 918e5e7ffa..794e5463d8 100644 --- a/cognite_toolkit/_cdf_tk/commands/_migrate/creators.py +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/creators.py @@ -21,11 +21,13 @@ from cognite_toolkit._cdf_tk.client import ToolkitClient from cognite_toolkit._cdf_tk.client.data_classes.apm_config_v1 import APMConfig, APMConfigList -from cognite_toolkit._cdf_tk.cruds import NodeCRUD, ResourceCRUD, SpaceCRUD +from cognite_toolkit._cdf_tk.client.data_classes.location_filters import LocationFilterWriteList +from cognite_toolkit._cdf_tk.cruds import LocationFilterCRUD, NodeCRUD, ResourceCRUD, SpaceCRUD from cognite_toolkit._cdf_tk.exceptions import ToolkitRequiredValueError from cognite_toolkit._cdf_tk.utils import humanize_collection from .data_model import CREATED_SOURCE_SYSTEM_VIEW_ID, SPACE, SPACE_SOURCE_VIEW_ID +from .infield_config import create_infield_v2_config @dataclass @@ -215,17 +217,81 @@ class InfieldV2ConfigCreator(MigrationCreator[NodeApplyList]): HAS_LINEAGE = False def create_resources(self) -> NodeApplyList: + """Create InFieldLocationConfig nodes (LocationFilters are handled separately).""" apm_config_nodes = self.client.data_modeling.instances.list(instance_type="node", sources=APMConfig.view_id) apm_config = APMConfigList.from_nodes(apm_config_nodes) - new_config_nodes = NodeApplyList([]) + # Filter configs: prefer APP_CONFIG_V2, fallback to default-config, otherwise skip + config_to_migrate = None for config in apm_config: - new_config = self._create_infield_v2_config(config) - new_config_nodes.append(new_config) - return new_config_nodes + if config.external_id == "APP_CONFIG_V2": + config_to_migrate = config + break + if config_to_migrate is None: + for config in apm_config: + if config.external_id == "default-config": + config_to_migrate = config + break + + if config_to_migrate is None: + return NodeApplyList([]) + + if not config_to_migrate.feature_configuration or not config_to_migrate.feature_configuration.root_location_configurations: + return NodeApplyList([]) + + feature_config = config_to_migrate.feature_configuration + root_location_configs = feature_config.root_location_configurations or [] + # Convert feature_config to dict for disciplines and dataExplorationConfig migration + feature_config_dict = feature_config.dump(camel_case=True) if hasattr(feature_config, "dump") else None + migration_result = create_infield_v2_config( + root_location_configs, + feature_configuration=feature_config_dict, + config_external_id=config_to_migrate.external_id, + client=self.client, + ) + return migration_result.all_nodes() + + def create_location_filters(self) -> LocationFilterWriteList: + """Create LocationFilter resources (to be deployed via Location Filters API).""" + apm_config_nodes = self.client.data_modeling.instances.list(instance_type="node", sources=APMConfig.view_id) + apm_config = APMConfigList.from_nodes(apm_config_nodes) + + # Filter configs: prefer APP_CONFIG_V2, fallback to default-config, otherwise skip + config_to_migrate = None + for config in apm_config: + if config.external_id == "APP_CONFIG_V2": + config_to_migrate = config + break + if config_to_migrate is None: + for config in apm_config: + if config.external_id == "default-config": + config_to_migrate = config + break + + if config_to_migrate is None: + return LocationFilterWriteList([]) + + if not config_to_migrate.feature_configuration or not config_to_migrate.feature_configuration.root_location_configurations: + return LocationFilterWriteList([]) + + feature_config = config_to_migrate.feature_configuration + root_location_configs = feature_config.root_location_configurations or [] + # Convert feature_config to dict (not needed for location filters, but keep consistent) + feature_config_dict = feature_config.dump(camel_case=True) if hasattr(feature_config, "dump") else None + migration_result = create_infield_v2_config( + root_location_configs, + feature_configuration=feature_config_dict, + config_external_id=config_to_migrate.external_id, + client=self.client, + ) + return migration_result.all_location_filters() def resource_configs(self, resources: NodeApplyList) -> list[ResourceConfig]: return [ResourceConfig(filestem=node.external_id, data=node.dump()) for node in resources] - def _create_infield_v2_config(self, config: APMConfig) -> NodeApply: - raise NotImplementedError("To be implemented") + def location_filter_configs(self, resources: LocationFilterWriteList) -> list[ResourceConfig]: + return [ResourceConfig(filestem=location_filter.external_id, data=location_filter.dump()) for location_filter in resources] + + def store_lineage(self, resources: NodeApplyList) -> int: + # No lineage to store for Infield V2 configs + return 0 diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/__init__.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/__init__.py new file mode 100644 index 0000000000..888bfd429d --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/__init__.py @@ -0,0 +1,10 @@ +"""InField V2 configuration migration module. + +This module handles migration from old APM Config format to InField V2 configuration format. +The migration is split into separate modules for better organization and maintainability. +""" + +from .migration import create_infield_v2_config + +__all__ = ["create_infield_v2_config"] + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/constants.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/constants.py new file mode 100644 index 0000000000..a03fc5d255 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/constants.py @@ -0,0 +1,19 @@ +"""Constants for InField V2 config migration.""" + +from cognite.client.data_classes.data_modeling import ViewId +from cognite.client.data_classes.data_modeling.ids import DataModelId + +# View IDs for the new format +LOCATION_CONFIG_VIEW_ID = ViewId("infield_cdm_source_desc_sche_asset_file_ts", "InFieldLocationConfig", "v1") +DATA_EXPLORATION_CONFIG_VIEW_ID = ViewId("infield_cdm_source_desc_sche_asset_file_ts", "DataExplorationConfig", "v1") + +# Target space for InFieldLocationConfig nodes +TARGET_SPACE = "APM_Config" + +# Default data model for LocationFilter +DEFAULT_LOCATION_FILTER_DATA_MODEL = DataModelId( + space="infield_cdm_source_desc_sche_asset_file_ts", + external_id="InFieldOnCDM", + version="v1", +) + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/data_exploration_config.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/data_exploration_config.py new file mode 100644 index 0000000000..0875ea0ada --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/data_exploration_config.py @@ -0,0 +1,86 @@ +"""Migration of DataExplorationConfig for InField V2 config migration. + +This module handles the creation of a single DataExplorationConfig node per APM Config, +which is shared across all InFieldLocationConfig nodes via a direct relation. +""" + +from typing import Any + +from cognite.client.data_classes.data_modeling import NodeApply, NodeOrEdgeData + +from .constants import DATA_EXPLORATION_CONFIG_VIEW_ID, TARGET_SPACE +from .types_new import DataExplorationConfigProperties + + +def create_data_exploration_config_node( + feature_configuration: dict[str, Any] | None, + config_external_id: str, +) -> NodeApply | None: + """Create a DataExplorationConfig node from FeatureConfiguration. + + Only one DataExplorationConfig is created per APM Config, shared across all locations. + + Args: + feature_configuration: FeatureConfiguration dict from old APM Config + config_external_id: External ID of the APM Config (used to generate DataExplorationConfig external ID) + + Returns: + NodeApply for DataExplorationConfig, or None if feature_configuration is missing or incomplete + """ + if not feature_configuration: + return None + + # Extract properties from FeatureConfiguration + props: DataExplorationConfigProperties = {} + + # Migrate observations + if observations := feature_configuration.get("observations"): + props["observations"] = observations + + # Migrate activities + if activities := feature_configuration.get("activities"): + props["activities"] = activities + + # Migrate documents + if documents := feature_configuration.get("documents"): + # Remove metadata. prefix from type and description if present + migrated_documents = documents.copy() + if "type" in migrated_documents and isinstance(migrated_documents["type"], str): + migrated_documents["type"] = migrated_documents["type"].removeprefix("metadata.") + if "description" in migrated_documents and isinstance(migrated_documents["description"], str): + migrated_documents["description"] = migrated_documents["description"].removeprefix("metadata.") + props["documents"] = migrated_documents + + # Migrate notifications + if notifications := feature_configuration.get("notifications"): + props["notifications"] = notifications + + # Migrate assets (from assetPageConfiguration) + if assets := feature_configuration.get("assetPageConfiguration"): + props["assets"] = assets + + # Only create node if at least one property is present + if not props: + return None + + # Generate external ID for DataExplorationConfig + data_exploration_external_id = f"data_exploration_{config_external_id}" + + return NodeApply( + space=TARGET_SPACE, + external_id=data_exploration_external_id, + sources=[NodeOrEdgeData(source=DATA_EXPLORATION_CONFIG_VIEW_ID, properties=props)], + ) + + +def get_data_exploration_config_external_id(config_external_id: str) -> str: + """Generate external ID for DataExplorationConfig node. + + Args: + config_external_id: External ID of the APM Config + + Returns: + External ID for the DataExplorationConfig node + """ + return f"data_exploration_{config_external_id}" + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/__init__.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/__init__.py new file mode 100644 index 0000000000..d7f8d0ebf5 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/__init__.py @@ -0,0 +1,6 @@ +"""InFieldLocationConfig field migration module.""" + +from .fields import apply_location_config_fields + +__all__ = ["apply_location_config_fields"] + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/access_management.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/access_management.py new file mode 100644 index 0000000000..41f8f0a469 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/access_management.py @@ -0,0 +1,34 @@ +"""Migration of accessManagement field for InFieldLocationConfig.""" + +from typing import Any + +from ..types_new import AccessManagement + + +def migrate_access_management(location_dict: dict[str, Any]) -> AccessManagement | None: + """Migrate accessManagement from old configuration. + + Extracts templateAdmins and checklistAdmins from the old location configuration + and creates an AccessManagement dict. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + AccessManagement dict, or None if neither templateAdmins nor checklistAdmins are present + """ + template_admins = location_dict.get("templateAdmins") + checklist_admins = location_dict.get("checklistAdmins") + + # Only create accessManagement if at least one of the fields is present + if not template_admins and not checklist_admins: + return None + + access_management: AccessManagement = {} + if template_admins: + access_management["templateAdmins"] = template_admins + if checklist_admins: + access_management["checklistAdmins"] = checklist_admins + + return access_management + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/app_instance_space.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/app_instance_space.py new file mode 100644 index 0000000000..7522e181d4 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/app_instance_space.py @@ -0,0 +1,24 @@ +"""Migration of appInstanceSpace field for InFieldLocationConfig.""" + +from typing import Any + + +def migrate_app_instance_space(location_dict: dict[str, Any]) -> str | None: + """Migrate appInstanceSpace from appDataInstanceSpace. + + Extracts the appDataInstanceSpace from the old configuration and returns it + as appInstanceSpace. If appDataInstanceSpace is not present or is an empty string, + returns None. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + App instance space string, or None if not present or empty + """ + app_instance_space = location_dict.get("appDataInstanceSpace") + if not app_instance_space: # None or empty string + return None + + return app_instance_space + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/data_filters.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/data_filters.py new file mode 100644 index 0000000000..dffebbf8d2 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/data_filters.py @@ -0,0 +1,27 @@ +"""Migration of dataFilters field for InFieldLocationConfig.""" + +from typing import Any + +from ..types_new import RootLocationDataFilters + + +def migrate_data_filters(location_dict: dict[str, Any]) -> RootLocationDataFilters | None: + """Migrate dataFilters from old configuration. + + Extracts the dataFilters dictionary from the old location configuration + and returns it as a RootLocationDataFilters dict. The structure matches + between old and new format, so it can be returned as-is. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + RootLocationDataFilters dict, or None if dataFilters is not present + """ + data_filters = location_dict.get("dataFilters") + if not data_filters: + return None + + # Return the data filters as-is (already in the correct format) + return data_filters + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/disciplines.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/disciplines.py new file mode 100644 index 0000000000..19311541fe --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/disciplines.py @@ -0,0 +1,29 @@ +"""Migration of disciplines field for InFieldLocationConfig.""" + +from typing import Any + +from ..types_new import Discipline + + +def migrate_disciplines(feature_configuration: dict[str, Any] | None) -> list[Discipline] | None: + """Migrate disciplines from old FeatureConfiguration. + + Extracts the disciplines list from the FeatureConfiguration level + (disciplines are shared across all locations in a config). + + Args: + feature_configuration: FeatureConfiguration dict from old APM Config + + Returns: + List of Discipline dicts, or None if disciplines is not present or empty + """ + if not feature_configuration: + return None + + disciplines = feature_configuration.get("disciplines") + if not disciplines: + return None + + # Return disciplines as-is (already in the correct format) + return disciplines + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/feature_toggles.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/feature_toggles.py new file mode 100644 index 0000000000..4b09e3d370 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/feature_toggles.py @@ -0,0 +1,26 @@ +"""Migration of featureToggles field for InFieldLocationConfig.""" + +from typing import Any + +from ..types_new import FeatureToggles + + +def migrate_feature_toggles(location_dict: dict[str, Any]) -> FeatureToggles | None: + """Migrate featureToggles from old configuration. + + Extracts the featureToggles from the old configuration and returns it + as a FeatureToggles dict. If featureToggles is not present, returns None. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + FeatureToggles dict, or None if not present in old config + """ + feature_toggles = location_dict.get("featureToggles") + if not feature_toggles: + return None + + # Return the feature toggles as-is (already in the correct format) + return feature_toggles + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/fields.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/fields.py new file mode 100644 index 0000000000..dcf047d024 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/fields.py @@ -0,0 +1,125 @@ +"""Migration of InFieldLocationConfig fields from old APM Config format. + +This module orchestrates the migration of all InFieldLocationConfig fields. +Individual field migrations are handled in separate modules for better organization. +""" + +from typing import Any + +from cognite.client.data_classes.data_modeling import DirectRelationReference +from cognite.client.exceptions import CogniteAPIError + +from ..data_exploration_config import get_data_exploration_config_external_id +from ..constants import TARGET_SPACE +from .access_management import migrate_access_management +from .app_instance_space import migrate_app_instance_space +from .data_filters import migrate_data_filters +from .disciplines import migrate_disciplines +from .feature_toggles import migrate_feature_toggles +from .root_asset import migrate_root_asset + + +def apply_location_config_fields( + location_dict: dict[str, Any], + feature_configuration: dict[str, Any] | None = None, + client: Any = None, + config_external_id: str | None = None, +) -> dict[str, Any]: + """Apply migrated fields to InFieldLocationConfig properties. + + This function applies all migrated fields from the old configuration + to the InFieldLocationConfig properties. Fields are migrated incrementally + as features are implemented. + + Currently migrated fields: + - featureToggles: From featureToggles in old configuration + - rootAsset: Direct relation from sourceDataInstanceSpace and assetExternalId (only if asset exists) + - appInstanceSpace: From appDataInstanceSpace in old configuration + - accessManagement: From templateAdmins and checklistAdmins in old configuration + - disciplines: From disciplines in FeatureConfiguration (shared across all locations) + - dataFilters: From dataFilters in old configuration + - dataExplorationConfig: Direct relation to DataExplorationConfig (shared across all locations) + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + feature_configuration: Optional FeatureConfiguration dict (for disciplines and dataExplorationConfig) + client: Optional client to verify asset existence before including rootAsset + config_external_id: Optional external ID of the APM Config (for dataExplorationConfig reference) + + Returns: + Dictionary of properties to add to InFieldLocationConfig + """ + props: dict[str, Any] = {} + + # Migrate featureToggles + feature_toggles = migrate_feature_toggles(location_dict) + if feature_toggles is not None: + props["featureToggles"] = feature_toggles + + # Migrate rootAsset - only include if asset exists (to avoid auto-create errors) + # TODO: Re-enable when asset data is ready in the project + # root_asset = migrate_root_asset(location_dict) + # if root_asset is not None and _asset_exists(root_asset, client): + # props["rootAsset"] = root_asset + + # Migrate appInstanceSpace + app_instance_space = migrate_app_instance_space(location_dict) + if app_instance_space is not None: + props["appInstanceSpace"] = app_instance_space + + # Migrate accessManagement + access_management = migrate_access_management(location_dict) + if access_management is not None: + props["accessManagement"] = access_management + + # Migrate disciplines (from FeatureConfiguration level) + disciplines = migrate_disciplines(feature_configuration) + if disciplines is not None: + props["disciplines"] = disciplines + + # Migrate dataFilters + data_filters = migrate_data_filters(location_dict) + if data_filters is not None: + props["dataFilters"] = data_filters + + # Migrate dataExplorationConfig (direct relation to shared DataExplorationConfig node) + if config_external_id: + data_exploration_external_id = get_data_exploration_config_external_id(config_external_id) + props["dataExplorationConfig"] = DirectRelationReference( + space=TARGET_SPACE, + external_id=data_exploration_external_id, + ) + + # TODO: Add more field migrations here as they are implemented + # - threeDConfiguration + # etc. + + return props + + +def _asset_exists(root_asset: DirectRelationReference, client: Any | None) -> bool: + """Check if the asset node exists in CDF. + + Args: + root_asset: DirectRelationReference to the asset + client: ToolkitClient or None + + Returns: + True if asset exists, False otherwise (including if client is None) + """ + if client is None: + # If no client provided, we can't verify - return False to skip it + # (to avoid auto-create errors) + return False + + try: + # Try to retrieve the node to verify it exists + from cognite.client.data_classes.data_modeling import NodeId + node_id = NodeId(space=root_asset.space, external_id=root_asset.external_id) + result = client.data_modeling.instances.retrieve(nodes=[node_id]) + # Check if we actually got a node back (retrieve may return empty list if node doesn't exist) + return len(result.nodes) > 0 + except CogniteAPIError: + # Asset doesn't exist or we can't access it + return False + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/root_asset.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/root_asset.py new file mode 100644 index 0000000000..97bb7d39d6 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_config/root_asset.py @@ -0,0 +1,29 @@ +"""Migration of rootAsset field for InFieldLocationConfig.""" + +from typing import Any + +from cognite.client.data_classes.data_modeling import DirectRelationReference + + +def migrate_root_asset(location_dict: dict[str, Any]) -> DirectRelationReference | None: + """Migrate rootAsset from sourceDataInstanceSpace and assetExternalId. + + Creates a DirectRelationReference using: + - space: sourceDataInstanceSpace from old config + - externalId: assetExternalId from old config + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + DirectRelationReference, or None if required fields are missing + """ + source_space = location_dict.get("sourceDataInstanceSpace") + asset_external_id = location_dict.get("assetExternalId") + + # Both fields are required to create a DirectRelationReference + if not source_space or not asset_external_id: + return None + + return DirectRelationReference(space=source_space, external_id=asset_external_id) + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/__init__.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/__init__.py new file mode 100644 index 0000000000..0e9ba2fafe --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/__init__.py @@ -0,0 +1,6 @@ +"""LocationFilter field migration module.""" + +from .fields import apply_location_filter_fields + +__all__ = ["apply_location_filter_fields"] + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/data_models.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/data_models.py new file mode 100644 index 0000000000..43bbf173f2 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/data_models.py @@ -0,0 +1,18 @@ +"""Migration of dataModels field for LocationFilter.""" + +from cognite.client.data_classes.data_modeling.ids import DataModelId + +from ..constants import DEFAULT_LOCATION_FILTER_DATA_MODEL + + +def migrate_data_models() -> list[DataModelId]: + """Migrate dataModels field. + + This function returns a hardcoded list containing the default data model + for LocationFilter. All LocationFilters will have this data model. + + Returns: + List containing the default LocationFilter data model + """ + return [DEFAULT_LOCATION_FILTER_DATA_MODEL] + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/fields.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/fields.py new file mode 100644 index 0000000000..f9f52a1812 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/fields.py @@ -0,0 +1,51 @@ +"""Migration of LocationFilter fields from old APM Config format. + +This module orchestrates the migration of all LocationFilter fields. +Individual field migrations are handled in separate modules for better organization. +""" + +from typing import Any + +from cognite_toolkit._cdf_tk.client.data_classes.location_filters import LocationFilterWrite + +from .data_models import migrate_data_models +from .instance_spaces import migrate_instance_spaces + + +def apply_location_filter_fields( + location_filter: LocationFilterWrite, location_dict: dict[str, Any] +) -> LocationFilterWrite: + """Apply migrated fields to a LocationFilter. + + This function applies all migrated fields from the old configuration + to the LocationFilter. Fields are migrated incrementally as features + are implemented. + + Currently migrated fields: + - instanceSpaces: From sourceDataInstanceSpace and appDataInstanceSpace + - dataModels: Hardcoded default data model + + Args: + location_filter: The LocationFilter to update + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + Updated LocationFilter with migrated fields applied + """ + # Migrate instanceSpaces + instance_spaces = migrate_instance_spaces(location_dict) + if instance_spaces is not None: + location_filter.instance_spaces = instance_spaces + + # Migrate dataModels + data_models = migrate_data_models() + location_filter.data_models = data_models + + # TODO: Add more field migrations here as they are implemented + # - views + # - assetCentric + # - scene + # etc. + + return location_filter + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/instance_spaces.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/instance_spaces.py new file mode 100644 index 0000000000..b3130d68aa --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/location_filter/instance_spaces.py @@ -0,0 +1,27 @@ +"""Migration of instanceSpaces field for LocationFilter.""" + +from typing import Any + + +def migrate_instance_spaces(location_dict: dict[str, Any]) -> list[str] | None: + """Migrate instanceSpaces from sourceDataInstanceSpace and appDataInstanceSpace. + + Collects both sourceDataInstanceSpace and appDataInstanceSpace from the old + configuration and returns them as a list. Empty strings are excluded. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + List of instance spaces, or None if no valid spaces found + """ + instance_spaces = [] + if source_space := location_dict.get("sourceDataInstanceSpace"): + if source_space: # Check that it's not empty string + instance_spaces.append(source_space) + if app_space := location_dict.get("appDataInstanceSpace"): + if app_space: # Check that it's not empty string + instance_spaces.append(app_space) + + return instance_spaces if instance_spaces else None + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/migration.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/migration.py new file mode 100644 index 0000000000..f2e4101934 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/migration.py @@ -0,0 +1,196 @@ +"""Main migration functions for InField V2 config migration.""" + +from dataclasses import dataclass +from typing import Any + +from cognite.client.data_classes.data_modeling import NodeApply, NodeApplyList, NodeOrEdgeData + +from cognite_toolkit._cdf_tk.client.data_classes.location_filters import ( + LocationFilterWriteList, +) + +from .constants import LOCATION_CONFIG_VIEW_ID, TARGET_SPACE +from .data_exploration_config import create_data_exploration_config_node +from .location_config.fields import apply_location_config_fields +from .location_filter.fields import apply_location_filter_fields +from .types_new import InFieldLocationConfigProperties +from .types_old import RootLocationConfiguration +from .utils import ( + get_location_config_external_id, + get_location_filter_external_id, +) + + +@dataclass +class InfieldV2MigrationResult: + """Result of migrating an APMConfig to InField V2 format.""" + + location_filters: LocationFilterWriteList + infield_location_config_nodes: NodeApplyList + data_explorer_config_nodes: NodeApplyList + + def all_nodes(self) -> NodeApplyList: + """Return all InFieldLocationConfig nodes (LocationFilters are handled separately).""" + nodes = NodeApplyList([]) + nodes.extend(self.infield_location_config_nodes) + nodes.extend(self.data_explorer_config_nodes) + return nodes + + def all_location_filters(self) -> LocationFilterWriteList: + """Return all LocationFilter resources.""" + return self.location_filters + + +def create_infield_v2_config( + root_location_configs: list[RootLocationConfiguration | Any], + feature_configuration: dict[str, Any] | None = None, + config_external_id: str | None = None, + client: Any | None = None, +) -> InfieldV2MigrationResult: + """Migrate root location configurations to the new InField V2 format. + + For now, only migrates basic fields: externalId, name, and description. + Returns structured output with separate lists for different node types. + + Args: + root_location_configs: List of root location configurations from the old format + + Returns: + InfieldV2MigrationResult containing separate lists of nodes + """ + from cognite_toolkit._cdf_tk.client.data_classes.location_filters import LocationFilterWriteList + + location_filter_nodes = NodeApplyList([]) + infield_location_config_nodes = NodeApplyList([]) + data_explorer_config_nodes = NodeApplyList([]) + + if not root_location_configs: + return InfieldV2MigrationResult( + location_filters=LocationFilterWriteList([]), + infield_location_config_nodes=infield_location_config_nodes, + data_explorer_config_nodes=data_explorer_config_nodes, + ) + + # Create location filters (using Location Filters API) + location_filters = create_location_filters(root_location_configs) + + # Create DataExplorationConfig node (one per config, shared across all locations) + if feature_configuration and config_external_id: + data_exploration_node = create_data_exploration_config_node( + feature_configuration, config_external_id + ) + if data_exploration_node: + data_explorer_config_nodes.append(data_exploration_node) + + # Create infield location config nodes (using Data Modeling Instance API) + infield_location_config_nodes = create_infield_location_config_nodes( + root_location_configs, + feature_configuration=feature_configuration, + config_external_id=config_external_id, + client=client, + ) + + return InfieldV2MigrationResult( + location_filters=location_filters, + infield_location_config_nodes=infield_location_config_nodes, + data_explorer_config_nodes=data_explorer_config_nodes, + ) + + +def create_location_filters(root_location_configs: list[RootLocationConfiguration | Any]) -> LocationFilterWriteList: + """Create LocationFilter resources for each root location configuration. + + These will be upserted using the Location Filters API (storage/config/locationfilters). + + Args: + root_location_configs: List of root location configurations from the old format + + Returns: + LocationFilterWriteList containing LocationFilter resources + """ + from cognite_toolkit._cdf_tk.client.data_classes.location_filters import ( + LocationFilterWrite, + LocationFilterWriteList, + ) + + location_filters = LocationFilterWriteList([]) + + for old_location in root_location_configs: + location_dict = old_location.dump(camel_case=True) if hasattr(old_location, "dump") else {} + + # Generate location filter external ID with prefix + location_filter_external_id = get_location_filter_external_id(location_dict) + + # Get name from displayName or assetExternalId + name = location_dict.get("displayName") or location_dict.get("assetExternalId") or location_filter_external_id + + # Create LocationFilterWrite with basic fields + location_filter = LocationFilterWrite._load( + { + "externalId": location_filter_external_id, + "name": name, + "description": "InField location, migrated from old location configuration", + } + ) + + # Apply additional migrated fields (instanceSpaces, etc.) + location_filter = apply_location_filter_fields(location_filter, location_dict) + + location_filters.append(location_filter) + + return location_filters + + +def create_infield_location_config_nodes( + root_location_configs: list[RootLocationConfiguration | Any], + feature_configuration: dict[str, Any] | None = None, + config_external_id: str | None = None, + client: Any | None = None, +) -> NodeApplyList: + """Create InFieldLocationConfig nodes for each root location configuration. + + If the old config does not have externalId but only assetExternalId, adds a postfix with the index + to ensure uniqueness. Each InFieldLocationConfig references its corresponding LocationFilterDTO + via rootLocationExternalId. + + Args: + root_location_configs: List of root location configurations from the old format + + Returns: + NodeApplyList containing InFieldLocationConfig nodes + """ + nodes = NodeApplyList([]) + + for index, old_location in enumerate(root_location_configs): + location_dict = old_location.dump(camel_case=True) if hasattr(old_location, "dump") else {} + + # Generate location filter external ID (must match the one created in create_location_filters) + location_filter_external_id = get_location_filter_external_id(location_dict) + + # Generate location config external ID + location_external_id = get_location_config_external_id(location_dict, index) + + # Create location config with rootLocationExternalId reference + location_props: InFieldLocationConfigProperties = { + "rootLocationExternalId": location_filter_external_id, + } + + # Apply additional migrated fields (featureToggles, etc.) + additional_props = apply_location_config_fields( + location_dict, + feature_configuration=feature_configuration, + config_external_id=config_external_id, + client=client, + ) + location_props.update(additional_props) + + nodes.append( + NodeApply( + space=TARGET_SPACE, + external_id=location_external_id, + sources=[NodeOrEdgeData(source=LOCATION_CONFIG_VIEW_ID, properties=location_props)], + ) + ) + + return nodes + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/types_new.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/types_new.py new file mode 100644 index 0000000000..26f6940462 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/types_new.py @@ -0,0 +1,117 @@ +"""Type definitions for new InField V2 configuration format. + +This module contains TypedDict definitions for the new InField V2 configuration +format that is the target of the migration from the old APM Config format. +""" + +from typing import Any, TypedDict + +from cognite.client.data_classes.data_modeling.ids import DataModelId +from cognite.client.data_classes.data_modeling import DirectRelationReference + + +class LocationFilterDTOProperties(TypedDict, total=False): + """Properties for LocationFilterDTO node. + + Currently migrated fields: + - name: The name of the location filter + - description: Description indicating this was migrated from old location + - instanceSpaces: List of instance spaces from sourceDataInstanceSpace and appDataInstanceSpace + - dataModels: List of DataModelId references to data models + """ + externalId: str + name: str + description: str + instanceSpaces: list[str] + dataModels: list[DataModelId] + + +class ObservationFeatureToggles(TypedDict, total=False): + """Feature toggles for observations.""" + isEnabled: bool + isWriteBackEnabled: bool + notificationsEndpointExternalId: str + attachmentsEndpointExternalId: str + + +class FeatureToggles(TypedDict, total=False): + """Feature toggles for InField location configuration.""" + threeD: bool + trends: bool + documents: bool + workorders: bool + notifications: bool + media: bool + templateChecklistFlow: bool + workorderChecklistFlow: bool + observations: ObservationFeatureToggles + + +class AccessManagement(TypedDict, total=False): + """Access management configuration.""" + templateAdmins: list[str] # list of CDF group external IDs + checklistAdmins: list[str] # list of CDF group external IDs + + +class Discipline(TypedDict, total=False): + """Discipline definition.""" + externalId: str + name: str + + +class ResourceFilters(TypedDict, total=False): + """Resource filters.""" + datasetIds: list[int] | None + assetSubtreeExternalIds: list[str] | None + rootAssetExternalIds: list[str] | None + externalIdPrefix: str | None + spaces: list[str] | None + + +class RootLocationDataFilters(TypedDict, total=False): + """Data filters for root location.""" + general: ResourceFilters | None + assets: ResourceFilters | None + files: ResourceFilters | None + timeseries: ResourceFilters | None + + +class DataExplorationConfigProperties(TypedDict, total=False): + """Properties for DataExplorationConfig node. + + Contains configuration for data exploration features: + - observations: Observations feature configuration + - activities: Activities configuration + - documents: Document configuration + - notifications: Notifications configuration + - assets: Asset page configuration + """ + observations: dict[str, Any] # ObservationsConfigFeature + activities: dict[str, Any] # ActivitiesConfiguration + documents: dict[str, Any] # DocumentConfiguration + notifications: dict[str, Any] # NotificationsConfiguration + assets: dict[str, Any] # AssetPageConfiguration + + +class InFieldLocationConfigProperties(TypedDict, total=False): + """Properties for InFieldLocationConfig node. + + Currently migrated fields: + - rootLocationExternalId: Reference to the LocationFilterDTO external ID + - featureToggles: Feature toggles migrated from old configuration + - rootAsset: Direct relation to the root asset (space and externalId) + - appInstanceSpace: Application instance space from appDataInstanceSpace + - accessManagement: Template and checklist admin groups (from templateAdmins and checklistAdmins) + - disciplines: List of disciplines (from disciplines in FeatureConfiguration) + - dataFilters: Data filters for general, assets, files, and timeseries (from dataFilters in old configuration) + - dataExplorationConfig: Direct relation to the DataExplorationConfig node (shared across all locations) + """ + rootLocationExternalId: str + featureToggles: FeatureToggles + rootAsset: DirectRelationReference + appInstanceSpace: str + accessManagement: AccessManagement + disciplines: list[Discipline] + dataFilters: RootLocationDataFilters + dataExplorationConfig: DirectRelationReference + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/types_old.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/types_old.py new file mode 100644 index 0000000000..b7ec7b6e59 --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/types_old.py @@ -0,0 +1,209 @@ +"""Type definitions for old APM Config format. + +This module contains TypedDict definitions for the legacy APM Config format +that is being migrated to InField V2 configuration format. +""" + +from typing import Literal, TypedDict + + +class ViewReference(TypedDict, total=False): + """Reference to a view.""" + externalId: str + space: str + type: Literal["view"] + version: str + + +NamedView = Literal["activity", "asset", "operation", "notification"] + + +class ViewMappings(TypedDict, total=False): + """View mappings for different named views.""" + activity: ViewReference | None + asset: ViewReference | None + operation: ViewReference | None + notification: ViewReference | None + + +class ThreeDModelIdentifier(TypedDict, total=False): + """3D model identifier.""" + revisionId: int + modelId: int + name: str + + +class ThreeDConfiguration(TypedDict, total=False): + """3D configuration.""" + fullWeightModels: list[ThreeDModelIdentifier] + lightWeightModels: list[ThreeDModelIdentifier] + + +class ResourceFilters(TypedDict, total=False): + """Resource filters.""" + datasetIds: list[int] | None + assetSubtreeExternalIds: list[str] | None + rootAssetExternalIds: list[str] | None + externalIdPrefix: str | None + spaces: list[str] | None + + +class RootLocationDataFilters(TypedDict, total=False): + """Data filters for root location.""" + general: ResourceFilters | None + assets: ResourceFilters | None + files: ResourceFilters | None + timeseries: ResourceFilters | None + + +class ObservationFeatureToggles(TypedDict, total=False): + """Feature toggles for observations.""" + isEnabled: bool + isWriteBackEnabled: bool + notificationsEndpointExternalId: str + attachmentsEndpointExternalId: str + + +class RootLocationFeatureToggles(TypedDict, total=False): + """Feature toggles for root location.""" + threeD: bool + trends: bool + documents: bool + workorders: bool + notifications: bool + media: bool + templateChecklistFlow: bool + workorderChecklistFlow: bool + observations: ObservationFeatureToggles + + +class ObservationConfigFieldProperty(TypedDict, total=False): + """Observation config field property.""" + displayTitle: str + displayDescription: str + isRequired: bool + + +class ObservationConfigDropdownPropertyOption(TypedDict, total=False): + """Option for dropdown property.""" + id: str + value: str + label: str + + +class ObservationConfigDropdownProperty(ObservationConfigFieldProperty, total=False): + """Observation config dropdown property.""" + options: list[ObservationConfigDropdownPropertyOption] + + +class ObservationsConfig(TypedDict, total=False): + """Observations configuration.""" + files: ObservationConfigFieldProperty + description: ObservationConfigFieldProperty + asset: ObservationConfigFieldProperty + troubleshooting: ObservationConfigFieldProperty + type: ObservationConfigDropdownProperty + priority: ObservationConfigDropdownProperty + + +class RootLocationConfiguration(TypedDict, total=False): + """Root location configuration from old APM Config.""" + externalId: str + assetExternalId: str + displayName: str + threeDConfiguration: ThreeDConfiguration + dataSetId: int + templateAdmins: list[str] # list of CDF group names + checklistAdmins: list[str] # list of CDF group names + appDataInstanceSpace: str + sourceDataInstanceSpace: str + dataFilters: RootLocationDataFilters + featureToggles: RootLocationFeatureToggles + observations: ObservationsConfig + + +class Discipline(TypedDict, total=False): + """Discipline definition.""" + externalId: str + name: str + + +class PropertyConfiguration(TypedDict, total=False): + """Property configuration.""" + highlightedProperties: list[str] + linkableAssetKeys: list[str] + + +class AssetPageConfiguration(TypedDict, total=False): + """Asset page configuration.""" + propertyCard: PropertyConfiguration + + +class DocumentConfiguration(TypedDict, total=False): + """Document configuration.""" + title: str + description: str + type: str + + +class ActivityOverviewCardFilter(TypedDict, total=False): + """Activity overview card filter.""" + externalId: str + value: str + + +class ActivitiesConfiguration(TypedDict, total=False): + """Activities configuration.""" + overviewCard: dict[str, list[ActivityOverviewCardFilter]] + + +class NotificationOverviewCardFilter(TypedDict, total=False): + """Notification overview card filter.""" + externalId: str + value: str + + +class NotificationsConfiguration(TypedDict, total=False): + """Notifications configuration.""" + overviewCard: dict[str, list[NotificationOverviewCardFilter]] + + +class ObservationsConfigFeature(TypedDict, total=False): + """Observations feature config.""" + enabled: bool + sapWriteBack: dict[str, bool] + optionalMediaField: dict[str, bool] + + +class CopilotConfig(TypedDict, total=False): + """Copilot configuration.""" + enabled: bool + + +class FeatureConfiguration(TypedDict, total=False): + """Feature configuration from old APM Config.""" + # Shared + rootLocationConfigurations: list[RootLocationConfiguration] + + # InField + viewMappings: ViewMappings + assetPageConfiguration: AssetPageConfiguration + documents: DocumentConfiguration + activities: ActivitiesConfiguration + notifications: NotificationsConfiguration + disciplines: list[Discipline] + copilot: CopilotConfig + observations: ObservationsConfigFeature + + +class AppConfig(TypedDict, total=False): + """Main APM Config (old format).""" + name: str + externalId: str + appDataSpaceId: str + appDataSpaceVersion: str + customerDataSpaceId: str + customerDataSpaceVersion: str + viewMappings: ViewMappings + featureConfiguration: FeatureConfiguration + diff --git a/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/utils.py b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/utils.py new file mode 100644 index 0000000000..87a16a851d --- /dev/null +++ b/cognite_toolkit/_cdf_tk/commands/_migrate/infield_config/utils.py @@ -0,0 +1,56 @@ +"""Utility functions for InField V2 config migration.""" + +import uuid +from typing import Any + + +def get_original_external_id(location_dict: dict[str, Any]) -> str: + """Extract the original external ID from a location configuration dict. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + Original external ID (externalId, assetExternalId, or generated UUID) + """ + return ( + location_dict.get("externalId") + or location_dict.get("assetExternalId") + or str(uuid.uuid4()) + ) + + +def get_location_filter_external_id(location_dict: dict[str, Any]) -> str: + """Generate the LocationFilter external ID with prefix. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + + Returns: + Location filter external ID with "location_filter_" prefix + """ + original_external_id = get_original_external_id(location_dict) + return f"location_filter_{original_external_id}" + + +def get_location_config_external_id(location_dict: dict[str, Any], index: int) -> str: + """Generate the InFieldLocationConfig external ID. + + If externalId exists, use it directly. If only assetExternalId exists, + add index postfix to ensure uniqueness. Otherwise, generate a UUID. + + Args: + location_dict: Location configuration dict (from dump(camel_case=True)) + index: Index of the location in the list (for uniqueness when only assetExternalId exists) + + Returns: + Location config external ID + """ + if location_dict.get("externalId"): + return location_dict["externalId"] + elif location_dict.get("assetExternalId"): + # Add index postfix to ensure uniqueness when only assetExternalId is available + return f"{location_dict['assetExternalId']}_{index}" + else: + return f"infield_location_{str(uuid.uuid4())}" + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/__init__.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/__init__.py new file mode 100644 index 0000000000..3f12d87ac3 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/__init__.py @@ -0,0 +1,2 @@ +"""Tests for InField V2 config migration.""" + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/__init__.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/__init__.py new file mode 100644 index 0000000000..97fe520283 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/__init__.py @@ -0,0 +1,2 @@ +"""Tests for InFieldLocationConfig field migrations.""" + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_access_management.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_access_management.py new file mode 100644 index 0000000000..00ffc2d4b0 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_access_management.py @@ -0,0 +1,220 @@ +"""Tests for accessManagement migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestAccessManagementMigration: + def test_access_management_with_both_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that accessManagement is migrated when both templateAdmins and checklistAdmins exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "templateAdmins": ["template_admin_1", "template_admin_2"], + "checklistAdmins": ["checklist_admin_1"], + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that accessManagement is present + assert "accessManagement" in location_props + access_management = location_props["accessManagement"] + assert access_management["templateAdmins"] == ["template_admin_1", "template_admin_2"] + assert access_management["checklistAdmins"] == ["checklist_admin_1"] + + def test_access_management_with_only_template_admins( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that accessManagement is migrated when only templateAdmins exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "templateAdmins": ["template_admin_1"], + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that accessManagement is present with only templateAdmins + assert "accessManagement" in location_props + access_management = location_props["accessManagement"] + assert access_management["templateAdmins"] == ["template_admin_1"] + assert "checklistAdmins" not in access_management + + def test_access_management_with_only_checklist_admins( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that accessManagement is migrated when only checklistAdmins exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "checklistAdmins": ["checklist_admin_1", "checklist_admin_2"], + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that accessManagement is present with only checklistAdmins + assert "accessManagement" in location_props + access_management = location_props["accessManagement"] + assert access_management["checklistAdmins"] == ["checklist_admin_1", "checklist_admin_2"] + assert "templateAdmins" not in access_management + + def test_access_management_with_no_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that accessManagement is not included when neither templateAdmins nor checklistAdmins exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + # No templateAdmins or checklistAdmins + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that accessManagement is not present + assert "accessManagement" not in location_props + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_app_instance_space.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_app_instance_space.py new file mode 100644 index 0000000000..f09c06aa49 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_app_instance_space.py @@ -0,0 +1,217 @@ +"""Tests for appInstanceSpace migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestAppInstanceSpaceMigration: + def test_app_instance_space_present( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that appInstanceSpace is migrated when appDataInstanceSpace exists.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "appDataInstanceSpace": "app_space_123", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that appInstanceSpace is present + assert "appInstanceSpace" in location_props + assert location_props["appInstanceSpace"] == "app_space_123" + + def test_app_instance_space_missing( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that appInstanceSpace is not included when appDataInstanceSpace is missing.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + # No appDataInstanceSpace + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that appInstanceSpace is not present when appDataInstanceSpace is missing + assert "appInstanceSpace" not in location_props + + def test_app_instance_space_empty_string( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that appInstanceSpace is not included when appDataInstanceSpace is an empty string.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "appDataInstanceSpace": "", # Empty string + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that appInstanceSpace is not present when appDataInstanceSpace is empty string + assert "appInstanceSpace" not in location_props + + def test_app_instance_space_with_other_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that appInstanceSpace works together with other migrated fields.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "appDataInstanceSpace": "app_space_456", + "assetExternalId": "asset_456", + "sourceDataInstanceSpace": "source_space", + "featureToggles": { + "threeD": True, + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that appInstanceSpace and other fields are present + assert "appInstanceSpace" in location_props + assert location_props["appInstanceSpace"] == "app_space_456" + assert "featureToggles" in location_props + # Note: rootAsset migration is currently commented out + # When re-enabled, uncomment the following: + # assert "rootAsset" in location_props + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_data_filters.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_data_filters.py new file mode 100644 index 0000000000..11374bfa40 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_data_filters.py @@ -0,0 +1,326 @@ +"""Tests for dataFilters migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestDataFiltersMigration: + def test_data_filters_with_all_sections( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that dataFilters is migrated when all sections (general, assets, files, timeseries) are present.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "dataFilters": { + "general": { + "datasetIds": [1, 2, 3], + "spaces": ["space1", "space2"], + }, + "assets": { + "assetSubtreeExternalIds": ["asset1", "asset2"], + "rootAssetExternalIds": ["root1"], + }, + "files": { + "externalIdPrefix": "file_prefix_", + "datasetIds": [4, 5], + }, + "timeseries": { + "spaces": ["ts_space1"], + "externalIdPrefix": "ts_prefix_", + }, + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that dataFilters is present with all sections + assert "dataFilters" in location_props + data_filters = location_props["dataFilters"] + + assert "general" in data_filters + assert data_filters["general"]["datasetIds"] == [1, 2, 3] + assert data_filters["general"]["spaces"] == ["space1", "space2"] + + assert "assets" in data_filters + assert data_filters["assets"]["assetSubtreeExternalIds"] == ["asset1", "asset2"] + assert data_filters["assets"]["rootAssetExternalIds"] == ["root1"] + + assert "files" in data_filters + assert data_filters["files"]["externalIdPrefix"] == "file_prefix_" + assert data_filters["files"]["datasetIds"] == [4, 5] + + assert "timeseries" in data_filters + assert data_filters["timeseries"]["spaces"] == ["ts_space1"] + assert data_filters["timeseries"]["externalIdPrefix"] == "ts_prefix_" + + def test_data_filters_with_partial_sections( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that dataFilters is migrated when only some sections are present.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "dataFilters": { + "general": { + "datasetIds": [1], + }, + "assets": { + "assetSubtreeExternalIds": ["asset1"], + }, + # files and timeseries are missing + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that dataFilters is present with only the sections that were provided + assert "dataFilters" in location_props + data_filters = location_props["dataFilters"] + + assert "general" in data_filters + assert data_filters["general"]["datasetIds"] == [1] + + assert "assets" in data_filters + assert data_filters["assets"]["assetSubtreeExternalIds"] == ["asset1"] + + # files and timeseries should not be present + assert "files" not in data_filters + assert "timeseries" not in data_filters + + def test_data_filters_with_no_data_filters( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that dataFilters is not included when dataFilters is missing in old config.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + # No dataFilters + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that dataFilters is not present + assert "dataFilters" not in location_props + + def test_data_filters_with_empty_dict( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that dataFilters is not included when dataFilters is an empty dict.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "dataFilters": {}, # Empty dict + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Empty dict should be treated as falsy and not included + assert "dataFilters" not in location_props + + def test_data_filters_with_other_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that dataFilters is migrated correctly when other fields are also present.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "appDataInstanceSpace": "app_space", + "dataFilters": { + "general": { + "datasetIds": [1], + }, + }, + "featureToggles": { + "threeD": True, + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that both dataFilters and other fields are present + assert "dataFilters" in location_props + assert "appInstanceSpace" in location_props + assert "featureToggles" in location_props + + data_filters = location_props["dataFilters"] + assert "general" in data_filters + assert data_filters["general"]["datasetIds"] == [1] + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_disciplines.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_disciplines.py new file mode 100644 index 0000000000..cdfd1110d7 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_disciplines.py @@ -0,0 +1,229 @@ +"""Tests for disciplines migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestDisciplinesMigration: + def test_disciplines_with_disciplines( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that disciplines are migrated when present in FeatureConfiguration.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + } + ], + "disciplines": [ + {"externalId": "mechanical", "name": "Mechanical"}, + {"externalId": "electrical", "name": "Electrical"}, + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that disciplines are present + assert "disciplines" in location_props + disciplines = location_props["disciplines"] + assert len(disciplines) == 2 + assert disciplines[0]["externalId"] == "mechanical" + assert disciplines[0]["name"] == "Mechanical" + assert disciplines[1]["externalId"] == "electrical" + assert disciplines[1]["name"] == "Electrical" + + def test_disciplines_with_multiple_locations( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that disciplines are shared across all locations.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_1", + }, + { + "externalId": "loc2", + "assetExternalId": "asset_2", + }, + ], + "disciplines": [ + {"externalId": "mechanical", "name": "Mechanical"}, + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 2 + + # Both locations should have the same disciplines + for location_node in created_nodes: + location_props = location_node.sources[0].properties + assert "disciplines" in location_props + disciplines = location_props["disciplines"] + assert len(disciplines) == 1 + assert disciplines[0]["externalId"] == "mechanical" + assert disciplines[0]["name"] == "Mechanical" + + def test_disciplines_with_no_disciplines( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that disciplines are not included when not present in FeatureConfiguration.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + } + ], + # No disciplines + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that disciplines is not present + assert "disciplines" not in location_props + + def test_disciplines_with_empty_list( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that disciplines are not included when disciplines is an empty list.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + } + ], + "disciplines": [], # Empty list + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that disciplines is not present when empty + assert "disciplines" not in location_props + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_feature_toggles.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_feature_toggles.py new file mode 100644 index 0000000000..6777147514 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_feature_toggles.py @@ -0,0 +1,256 @@ +"""Tests for featureToggles migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestFeatureTogglesMigration: + def test_feature_toggles_with_all_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that featureToggles is migrated when all fields are present.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "featureToggles": { + "threeD": True, + "trends": False, + "documents": True, + "workorders": False, + "notifications": True, + "media": False, + "templateChecklistFlow": True, + "workorderChecklistFlow": False, + "observations": { + "isEnabled": True, + "isWriteBackEnabled": False, + "notificationsEndpointExternalId": "notif_endpoint", + "attachmentsEndpointExternalId": "attach_endpoint", + }, + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that featureToggles is present + assert "featureToggles" in location_props + feature_toggles = location_props["featureToggles"] + + # Check all boolean fields + assert feature_toggles["threeD"] is True + assert feature_toggles["trends"] is False + assert feature_toggles["documents"] is True + assert feature_toggles["workorders"] is False + assert feature_toggles["notifications"] is True + assert feature_toggles["media"] is False + assert feature_toggles["templateChecklistFlow"] is True + assert feature_toggles["workorderChecklistFlow"] is False + + # Check observations nested object + assert "observations" in feature_toggles + observations = feature_toggles["observations"] + assert observations["isEnabled"] is True + assert observations["isWriteBackEnabled"] is False + assert observations["notificationsEndpointExternalId"] == "notif_endpoint" + assert observations["attachmentsEndpointExternalId"] == "attach_endpoint" + + def test_feature_toggles_with_partial_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that featureToggles is migrated when only some fields are present.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "featureToggles": { + "threeD": True, + "documents": False, + # Other fields missing - should still work + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that featureToggles is present + assert "featureToggles" in location_props + feature_toggles = location_props["featureToggles"] + + # Check that only the present fields are migrated + assert feature_toggles["threeD"] is True + assert feature_toggles["documents"] is False + + def test_feature_toggles_with_no_feature_toggles( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that featureToggles is not included when missing from old config.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + # No featureToggles + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that featureToggles is not present when missing from old config + assert "featureToggles" not in location_props + + def test_feature_toggles_with_empty_observations( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that featureToggles works when observations is present but empty.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "featureToggles": { + "threeD": True, + "observations": {}, # Empty observations + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that featureToggles is present + assert "featureToggles" in location_props + feature_toggles = location_props["featureToggles"] + assert feature_toggles["threeD"] is True + assert "observations" in feature_toggles + assert feature_toggles["observations"] == {} + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_root_asset.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_root_asset.py new file mode 100644 index 0000000000..325586a0fa --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_config/test_root_asset.py @@ -0,0 +1,277 @@ +"""Tests for rootAsset migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestRootAssetMigration: + def test_root_asset_with_both_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that rootAsset is migrated when both sourceDataInstanceSpace and assetExternalId exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + "sourceDataInstanceSpace": "source_space", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Note: rootAsset migration is currently commented out + # When re-enabled, uncomment the following: + # assert "rootAsset" in location_props + # root_asset = location_props["rootAsset"] + # assert root_asset.space == "source_space" + # assert root_asset.external_id == "asset_123" + # For now, just verify the node was created + assert "rootLocationExternalId" in location_props + + def test_root_asset_with_missing_source_space( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that rootAsset is not included when sourceDataInstanceSpace is missing.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + # Missing sourceDataInstanceSpace + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that rootAsset is not present when sourceDataInstanceSpace is missing + assert "rootAsset" not in location_props + + def test_root_asset_with_missing_asset_external_id( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that rootAsset is not included when assetExternalId is missing.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "sourceDataInstanceSpace": "source_space", + # Missing assetExternalId + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that rootAsset is not present when assetExternalId is missing + assert "rootAsset" not in location_props + + def test_root_asset_with_empty_strings( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that rootAsset is not included when fields are empty strings.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "", # Empty string + "sourceDataInstanceSpace": "source_space", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that rootAsset is not present when assetExternalId is empty string + assert "rootAsset" not in location_props + + def test_root_asset_with_other_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that rootAsset works together with other migrated fields.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_456", + "sourceDataInstanceSpace": "source_space", + "featureToggles": { + "threeD": True, + }, + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + location_props = location_node.sources[0].properties + + # Check that featureToggles is present + assert "featureToggles" in location_props + # Note: rootAsset migration is currently commented out + # When re-enabled, uncomment the following: + # assert "rootAsset" in location_props + # root_asset = location_props["rootAsset"] + # assert root_asset.space == "source_space" + # assert root_asset.external_id == "asset_456" + + feature_toggles = location_props["featureToggles"] + assert feature_toggles["threeD"] is True + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/__init__.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/__init__.py new file mode 100644 index 0000000000..07c36271b0 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/__init__.py @@ -0,0 +1,2 @@ +"""Tests for LocationFilter field migrations.""" + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/test_data_models.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/test_data_models.py new file mode 100644 index 0000000000..1e5207bec3 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/test_data_models.py @@ -0,0 +1,175 @@ +"""Tests for dataModels migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from cognite_toolkit._cdf_tk.commands._migrate.infield_config.constants import DEFAULT_LOCATION_FILTER_DATA_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestDataModelsMigration: + def test_data_models_always_present( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that dataModels is always present with the default data model.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + + # Check that dataModels is always present + assert location_filter.data_models is not None + assert len(location_filter.data_models) == 1 + + # Check that it's the default data model + data_model = location_filter.data_models[0] + assert data_model.space == DEFAULT_LOCATION_FILTER_DATA_MODEL.space + assert data_model.external_id == DEFAULT_LOCATION_FILTER_DATA_MODEL.external_id + assert data_model.version == DEFAULT_LOCATION_FILTER_DATA_MODEL.version + + def test_data_models_with_other_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that dataModels is present even when other fields are also migrated.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "sourceDataInstanceSpace": "source_space", + "appDataInstanceSpace": "app_space", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + + # Check that both instanceSpaces and dataModels are present + assert location_filter.instance_spaces is not None + assert location_filter.data_models is not None + assert len(location_filter.data_models) == 1 + + # Verify it's the default data model + data_model = location_filter.data_models[0] + assert data_model.space == "infield_cdm_source_desc_sche_asset_file_ts" + assert data_model.external_id == "InFieldOnCDM" + assert data_model.version == "v1" + + def test_data_models_multiple_locations( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that all locations get the same default data model.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + {"externalId": "loc1"}, + {"externalId": "loc2"}, + {"externalId": "loc3"}, + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 3 + + # Check that all location filters have the same default data model + for location_filter in created_location_filters: + assert location_filter.data_models is not None + assert len(location_filter.data_models) == 1 + data_model = location_filter.data_models[0] + assert data_model.space == "infield_cdm_source_desc_sche_asset_file_ts" + assert data_model.external_id == "InFieldOnCDM" + assert data_model.version == "v1" + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/test_instance_spaces.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/test_instance_spaces.py new file mode 100644 index 0000000000..9d99234c64 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/location_filter/test_instance_spaces.py @@ -0,0 +1,256 @@ +"""Tests for instanceSpaces migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node, NodeApply, NodeList + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from cognite_toolkit._cdf_tk.data_classes import ResourceDeployResult +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestInstanceSpacesMigration: + def test_instance_spaces_with_both_spaces( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that instanceSpaces is populated when both sourceDataInstanceSpace and appDataInstanceSpace exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "sourceDataInstanceSpace": "source_space", + "appDataInstanceSpace": "app_space", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + assert location_filter.instance_spaces is not None + assert len(location_filter.instance_spaces) == 2 + assert "source_space" in location_filter.instance_spaces + assert "app_space" in location_filter.instance_spaces + + def test_instance_spaces_with_only_source_space( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that instanceSpaces only contains sourceDataInstanceSpace when appDataInstanceSpace is missing.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "sourceDataInstanceSpace": "source_space", + # No appDataInstanceSpace + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + assert location_filter.instance_spaces is not None + assert len(location_filter.instance_spaces) == 1 + assert "source_space" in location_filter.instance_spaces + + def test_instance_spaces_with_only_app_space( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that instanceSpaces only contains appDataInstanceSpace when sourceDataInstanceSpace is missing.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + # No sourceDataInstanceSpace + "appDataInstanceSpace": "app_space", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + assert location_filter.instance_spaces is not None + assert len(location_filter.instance_spaces) == 1 + assert "app_space" in location_filter.instance_spaces + + def test_instance_spaces_with_no_spaces( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that instanceSpaces is not included when neither sourceDataInstanceSpace nor appDataInstanceSpace exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + # No sourceDataInstanceSpace or appDataInstanceSpace + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + # instanceSpaces should be None (not included) when both spaces are missing + assert location_filter.instance_spaces is None + + def test_instance_spaces_with_empty_strings( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that empty strings for instance spaces are not included in instanceSpaces.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "sourceDataInstanceSpace": "", # Empty string + "appDataInstanceSpace": "app_space", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + # Empty string should be falsy and not included + assert location_filter.instance_spaces is not None + assert len(location_filter.instance_spaces) == 1 + assert "app_space" in location_filter.instance_spaces + assert "" not in location_filter.instance_spaces + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_data_exploration_config.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_data_exploration_config.py new file mode 100644 index 0000000000..87210d33bf --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_data_exploration_config.py @@ -0,0 +1,250 @@ +"""Tests for DataExplorationConfig migration in InField V2 config migration.""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestDataExplorationConfigMigration: + def test_data_exploration_config_created( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that DataExplorationConfig node is created when FeatureConfiguration has data.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + } + ], + "observations": {"enabled": True}, + "activities": {"overviewCard": {}}, + "documents": {"title": "Documents", "type": "documents", "description": "Document config"}, + "notifications": {"overviewCard": {}}, + "assetPageConfiguration": {"propertyCard": {}}, + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Check that DataExplorationConfig node was created + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + data_exploration_nodes = [ + n for n in created_nodes if n.external_id == "data_exploration_default-config" + ] + assert len(data_exploration_nodes) == 1 + + data_exploration_node = data_exploration_nodes[0] + props = data_exploration_node.sources[0].properties + + assert "observations" in props + assert "activities" in props + assert "documents" in props + assert "notifications" in props + assert "assets" in props + + def test_documents_metadata_prefix_removed( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that metadata. prefix is removed from documents type and description.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + } + ], + "documents": { + "title": "Documents", + "type": "metadata.documents", + "description": "metadata.document_description", + }, + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Check that DataExplorationConfig node has documents with metadata. prefix removed + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + data_exploration_nodes = [ + n for n in created_nodes if n.external_id == "data_exploration_default-config" + ] + assert len(data_exploration_nodes) == 1 + + data_exploration_node = data_exploration_nodes[0] + props = data_exploration_node.sources[0].properties + + assert "documents" in props + documents = props["documents"] + assert documents["type"] == "documents" # metadata. prefix removed + assert documents["description"] == "document_description" # metadata. prefix removed + assert documents["title"] == "Documents" # unchanged + + def test_documents_without_metadata_prefix( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that documents without metadata. prefix are unchanged.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset_123", + } + ], + "documents": { + "title": "Documents", + "type": "documents", + "description": "document_description", + }, + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Check that DataExplorationConfig node has documents unchanged + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + data_exploration_nodes = [ + n for n in created_nodes if n.external_id == "data_exploration_default-config" + ] + assert len(data_exploration_nodes) == 1 + + data_exploration_node = data_exploration_nodes[0] + props = data_exploration_node.sources[0].properties + + assert "documents" in props + documents = props["documents"] + assert documents["type"] == "documents" + assert documents["description"] == "document_description" + assert documents["title"] == "Documents" + + def test_data_exploration_config_linked_to_all_locations( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that DataExplorationConfig is linked to all InFieldLocationConfig nodes.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + {"externalId": "loc1", "assetExternalId": "asset_1"}, + {"externalId": "loc2", "assetExternalId": "asset_2"}, + ], + "documents": {"title": "Documents", "type": "documents"}, + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Check that all InFieldLocationConfig nodes have dataExplorationConfig reference + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + location_config_nodes = [ + n for n in created_nodes if n.external_id in ["loc1", "loc2"] + ] + assert len(location_config_nodes) == 2 + + for location_node in location_config_nodes: + props = location_node.sources[0].properties + assert "dataExplorationConfig" in props + data_exploration_ref = props["dataExplorationConfig"] + assert data_exploration_ref.space == "APM_Config" + assert data_exploration_ref.external_id == "data_exploration_default-config" + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_infield_v2_config_creator.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_infield_v2_config_creator.py new file mode 100644 index 0000000000..e938c4171c --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_infield_v2_config_creator.py @@ -0,0 +1,584 @@ +from pathlib import Path +from typing import Any + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node, NodeApply, NodeList, View + +from cognite_toolkit._cdf_tk.client.data_classes.apm_config_v1 import ( + APMConfig, + FeatureConfiguration, + RootLocationConfiguration, +) +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from cognite_toolkit._cdf_tk.data_classes import ResourceDeployResult +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestInfieldV2ConfigCreator: + @pytest.mark.parametrize("dry_run", [pytest.param(True, id="dry_run"), pytest.param(False, id="not_dry_run")]) + def test_create_infield_v2_configs_basic( + self, dry_run: bool, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test basic migration of APMConfig to InField V2 format.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + # Create a mock APMConfig node (using default-config as it's the fallback) + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "customerDataSpaceId": "APM_SourceData", + "customerDataSpaceVersion": "1", + "name": "Test Config", + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "location_1", + "assetExternalId": "asset_1", + "displayName": "Location 1", + "appDataInstanceSpace": "app_space_1", + "sourceDataInstanceSpace": "source_space_1", + "dataSetId": 123, + "featureToggles": { + "threeD": True, + "documents": True, + }, + "templateAdmins": ["admin1"], + "checklistAdmins": ["admin2"], + } + ], + }, + } + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=dry_run, + verbose=False, + output_dir=tmp_path, + ) + + # LocationFilters are now deployed separately via Location Filters API + assert "location filters" in results + location_filter_result = results["location filters"] + assert isinstance(location_filter_result, ResourceDeployResult) + assert location_filter_result.created == 1 + + # InFieldLocationConfig nodes are deployed via Data Modeling Instance API + assert "nodes" in results + node_result = results["nodes"] + assert isinstance(node_result, ResourceDeployResult) + assert node_result.created == 1 + + # Check that config files were created (both LocationFilter and Node files) + location_filter_configs = list(tmp_path.rglob("*LocationFilter.yaml")) + node_configs = list(tmp_path.rglob("*Node.yaml")) + assert len(location_filter_configs) == 1 + assert len(node_configs) == 1 + + if not dry_run: + # Check LocationFilter resource (deployed via Location Filters API) + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + from cognite_toolkit._cdf_tk.client.data_classes.location_filters import LocationFilterWrite + assert isinstance(location_filter, LocationFilterWrite) + assert location_filter.external_id == "location_filter_location_1" + assert location_filter.name == "Location 1" + assert location_filter.description == "InField location, migrated from old location configuration" + # Check that instanceSpaces is populated with both sourceDataInstanceSpace and appDataInstanceSpace + # Note: Detailed instanceSpaces tests are in test_infield_instance_spaces.py + assert location_filter.instance_spaces is not None + assert len(location_filter.instance_spaces) == 2 + assert "source_space_1" in location_filter.instance_spaces + assert "app_space_1" in location_filter.instance_spaces + + # Check InFieldLocationConfig node (deployed via Data Modeling Instance API) + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + assert isinstance(location_node, NodeApply) + assert location_node.external_id == "location_1" + assert location_node.space == "APM_Config" + assert len(location_node.sources) == 1 + assert location_node.sources[0].source.external_id == "InFieldLocationConfig" + + # Check properties - should have rootLocationExternalId referencing the location filter + location_props = location_node.sources[0].properties + assert location_props["rootLocationExternalId"] == "location_filter_location_1" + + def test_create_infield_v2_configs_multiple_locations( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test migration with multiple root locations.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "customerDataSpaceId": "APM_SourceData", + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset1", + }, + { + "externalId": "loc2", + "assetExternalId": "asset2", + }, + ], + }, + } + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # LocationFilters are now deployed separately via Location Filters API + assert "location filters" in results + location_filter_result = results["location filters"] + assert location_filter_result.created == 2 + + # InFieldLocationConfig nodes are deployed via Data Modeling Instance API + assert "nodes" in results + node_result = results["nodes"] + assert node_result.created == 2 + + # Check LocationFilters (they are now LocationFilter resources, not Node resources) + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 2 + location_filter_ids = [lf.external_id for lf in created_location_filters] + assert "location_filter_loc1" in location_filter_ids + assert "location_filter_loc2" in location_filter_ids + + # Check InFieldLocationConfig nodes + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 2 + + # Should have two location config nodes + location_nodes = [n for n in created_nodes if n.external_id in ["loc1", "loc2"]] + assert len(location_nodes) == 2 + + # Verify location configs reference their location filters + loc1_node = next((n for n in location_nodes if n.external_id == "loc1"), None) + loc2_node = next((n for n in location_nodes if n.external_id == "loc2"), None) + assert loc1_node is not None + assert loc2_node is not None + assert loc1_node.sources[0].properties["rootLocationExternalId"] == "location_filter_loc1" + assert loc2_node.sources[0].properties["rootLocationExternalId"] == "location_filter_loc2" + + def test_create_infield_v2_configs_empty(self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path) -> None: + """Test migration with no root locations.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "empty_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [], + }, + } + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + assert "nodes" in results + result = results["nodes"] + # No locations, so no nodes created + assert result.created == 0 + + def test_create_infield_v2_configs_no_feature_config( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test migration with missing feature configuration.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "no_feature_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": {}, + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + assert "nodes" in results + result = results["nodes"] + assert result.created == 0 + + def test_external_id_generation_with_external_id( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test external ID generation when externalId exists in old config.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "my_location", + "assetExternalId": "my_asset", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources["Node"] + location_node = next((n for n in created_nodes if n.external_id == "my_location"), None) + assert location_node is not None + # Should use externalId directly when it exists + assert location_node.external_id == "my_location" + + def test_external_id_generation_with_only_asset_external_id( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test external ID generation when only assetExternalId exists (should add index postfix).""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + # No externalId, only assetExternalId + "assetExternalId": "shared_asset", + }, + { + # No externalId, only assetExternalId (same asset) + "assetExternalId": "shared_asset", + }, + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources["Node"] + # Should have 2 location config nodes with index postfix for uniqueness + location_nodes = [n for n in created_nodes if n.external_id.startswith("shared_asset_")] + assert len(location_nodes) == 2 + # Should have index 0 and 1 + external_ids = [n.external_id for n in location_nodes] + assert "shared_asset_0" in external_ids + assert "shared_asset_1" in external_ids + + def test_external_id_generation_with_neither_external_id( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test external ID generation when neither externalId nor assetExternalId exists (should generate UUID).""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + # No externalId, no assetExternalId + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources["Node"] + # Should have location config with generated UUID + location_nodes = [n for n in created_nodes if n.external_id.startswith("infield_location_")] + assert len(location_nodes) == 1 + # Should start with infield_location_ prefix + assert location_nodes[0].external_id.startswith("infield_location_") + + def test_config_selection_prioritizes_app_config_v2( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that APP_CONFIG_V2 is selected when both APP_CONFIG_V2 and default-config exist.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + # Create two configs: one with APP_CONFIG_V2 and one with default-config + app_config_v2_node = Node._load( + { + "space": "APM_Config", + "externalId": "APP_CONFIG_V2", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "location_from_v2", + "assetExternalId": "asset_v2", + "displayName": "Location from V2", + "appDataInstanceSpace": "app_space_v2", + "sourceDataInstanceSpace": "source_space_v2", + } + ], + }, + } + } + }, + } + ) + + default_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "location_from_default", + "assetExternalId": "asset_default", + "displayName": "Location from Default", + "appDataInstanceSpace": "app_space_default", + "sourceDataInstanceSpace": "source_space_default", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, app_config_v2_node) + toolkit_client_approval.append(Node, default_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Should only migrate from APP_CONFIG_V2, not default-config + assert "nodes" in results + node_result = results["nodes"] + assert node_result.created == 1 # Only one location from APP_CONFIG_V2 + + # Verify that only the location from APP_CONFIG_V2 was migrated + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + assert location_node.external_id == "location_from_v2" + + # Verify location properties point to APP_CONFIG_V2 data + location_props = location_node.sources[0].properties + # Note: rootAsset migration is currently commented out, so we don't check for it here + # When re-enabled, uncomment the following: + # root_asset = location_props["rootAsset"] + # assert root_asset.external_id == "asset_v2" + # assert root_asset.space == "source_space_v2" + + # Verify LocationFilter was also created from APP_CONFIG_V2 + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + assert location_filter.external_id == "location_filter_location_from_v2" + assert location_filter.name == "Location from V2" + + def test_config_with_wrong_external_id_not_migrated( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test that configs with externalId other than APP_CONFIG_V2 or default-config are not migrated.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + # Create a config with wrong externalId + wrong_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "other-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "location_from_other", + "assetExternalId": "asset_other", + "displayName": "Location from Other Config", + "appDataInstanceSpace": "app_space_other", + "sourceDataInstanceSpace": "source_space_other", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, wrong_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Should not migrate anything + assert "nodes" in results + node_result = results["nodes"] + assert node_result.created == 0 + + # When there's nothing to migrate, "location filters" may not be in results + # (it's only added if location_filters list is non-empty) + if "location filters" in results: + location_filter_result = results["location filters"] + assert location_filter_result.created == 0 + + # Verify no resources were created + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 0 + + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 0 + + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_infield_v2_config_integration.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_infield_v2_config_integration.py new file mode 100644 index 0000000000..c06aa5f4d3 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/infield_config/test_infield_v2_config_integration.py @@ -0,0 +1,178 @@ +"""Integration tests for InField V2 config migration. + +This test suite verifies that all components of the migration work together correctly. +Individual field migrations are tested in their respective test files: +- test_infield_v2_config_creator.py: Core migration logic and external ID generation +- test_infield_instance_spaces.py: instanceSpaces field migration + +This file focuses on end-to-end integration scenarios. +""" + +from pathlib import Path + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node + +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from cognite_toolkit._cdf_tk.data_classes import ResourceDeployResult +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestInfieldV2ConfigIntegration: + """Integration tests for complete InField V2 config migration.""" + + def test_complete_migration_with_all_fields( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test complete migration with all currently supported fields.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "customerDataSpaceId": "APM_SourceData", + "customerDataSpaceVersion": "1", + "name": "Complete Config", + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "complete_location", + "assetExternalId": "complete_asset", + "displayName": "Complete Location", + "sourceDataInstanceSpace": "source_space", + "appDataInstanceSpace": "app_space", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Verify both resource types were created + assert "location filters" in results + assert "nodes" in results + location_filter_result = results["location filters"] + node_result = results["nodes"] + + assert location_filter_result.created == 1 + assert node_result.created == 1 + + # Verify LocationFilter was created with all migrated fields + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + assert location_filter.external_id == "location_filter_complete_location" + assert location_filter.name == "Complete Location" + assert location_filter.instance_spaces is not None + assert len(location_filter.instance_spaces) == 2 + + # Verify InFieldLocationConfig node was created + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + assert location_node.external_id == "complete_location" + assert location_node.sources[0].properties["rootLocationExternalId"] == "location_filter_complete_location" + + def test_migration_with_multiple_locations_complete( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test migration of multiple locations with all supported fields.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "default-config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "sourceDataInstanceSpace": "source1", + "appDataInstanceSpace": "app1", + }, + { + "externalId": "loc2", + "sourceDataInstanceSpace": "source2", + # Missing appDataInstanceSpace - should still work + }, + { + # Missing externalId - should use assetExternalId with index + "assetExternalId": "shared_asset", + "sourceDataInstanceSpace": "source3", + }, + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # Should create 3 LocationFilters and 3 InFieldLocationConfig nodes + assert results["location filters"].created == 3 + assert results["nodes"].created == 3 + + # Verify all LocationFilters have correct instanceSpaces + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 3 + + location_filter_by_id = {lf.external_id: lf for lf in created_location_filters} + + # loc1 should have both spaces + assert location_filter_by_id["location_filter_loc1"].instance_spaces == ["source1", "app1"] + + # loc2 should only have source space + assert location_filter_by_id["location_filter_loc2"].instance_spaces == ["source2"] + + # shared_asset should have source3 (LocationFilter external ID doesn't use index) + assert location_filter_by_id["location_filter_shared_asset"].instance_spaces == ["source3"] + + # Verify all InFieldLocationConfig nodes reference their LocationFilters correctly + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 3 + + node_by_id = {node.external_id: node for node in created_nodes} + assert node_by_id["loc1"].sources[0].properties["rootLocationExternalId"] == "location_filter_loc1" + assert node_by_id["loc2"].sources[0].properties["rootLocationExternalId"] == "location_filter_loc2" + # InFieldLocationConfig uses index for uniqueness (index 2 since it's the third item in the list) + # but references LocationFilter without index + assert "shared_asset_2" in node_by_id + assert node_by_id["shared_asset_2"].sources[0].properties["rootLocationExternalId"] == "location_filter_shared_asset" + diff --git a/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/test_infield_v2_config_creator.py b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/test_infield_v2_config_creator.py new file mode 100644 index 0000000000..6fc54f6929 --- /dev/null +++ b/tests/test_unit/test_cdf_tk/test_commands/test_migration_cmd/test_infield_v2_config_creator.py @@ -0,0 +1,424 @@ +from pathlib import Path +from typing import Any + +import pytest +from cognite.client.data_classes.data_modeling import DataModel, Node, NodeApply, NodeList, View + +from cognite_toolkit._cdf_tk.client.data_classes.apm_config_v1 import ( + APMConfig, + FeatureConfiguration, + RootLocationConfiguration, +) +from cognite_toolkit._cdf_tk.commands._migrate.command import MigrationCommand +from cognite_toolkit._cdf_tk.commands._migrate.creators import InfieldV2ConfigCreator +from cognite_toolkit._cdf_tk.commands._migrate.data_model import COGNITE_MIGRATION_MODEL +from cognite_toolkit._cdf_tk.data_classes import ResourceDeployResult +from tests.test_unit.approval_client import ApprovalToolkitClient + + +class TestInfieldV2ConfigCreator: + @pytest.mark.parametrize("dry_run", [pytest.param(True, id="dry_run"), pytest.param(False, id="not_dry_run")]) + def test_create_infield_v2_configs_basic( + self, dry_run: bool, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test basic migration of APMConfig to InField V2 format.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + # Create a mock APMConfig node + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "test_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "customerDataSpaceId": "APM_SourceData", + "customerDataSpaceVersion": "1", + "name": "Test Config", + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "location_1", + "assetExternalId": "asset_1", + "displayName": "Location 1", + "appDataInstanceSpace": "app_space_1", + "sourceDataInstanceSpace": "source_space_1", + "dataSetId": 123, + "featureToggles": { + "threeD": True, + "documents": True, + }, + "templateAdmins": ["admin1"], + "checklistAdmins": ["admin2"], + } + ], + }, + } + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=dry_run, + verbose=False, + output_dir=tmp_path, + ) + + # LocationFilters are now deployed separately via Location Filters API + assert "location filters" in results + location_filter_result = results["location filters"] + assert isinstance(location_filter_result, ResourceDeployResult) + assert location_filter_result.created == 1 + + # InFieldLocationConfig nodes are deployed via Data Modeling Instance API + assert "nodes" in results + node_result = results["nodes"] + assert isinstance(node_result, ResourceDeployResult) + assert node_result.created == 1 + + # Check that config files were created (both LocationFilter and Node files) + location_filter_configs = list(tmp_path.rglob("*LocationFilter.yaml")) + node_configs = list(tmp_path.rglob("*Node.yaml")) + assert len(location_filter_configs) == 1 + assert len(node_configs) == 1 + + if not dry_run: + # Check LocationFilter resource (deployed via Location Filters API) + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 1 + location_filter = created_location_filters[0] + from cognite_toolkit._cdf_tk.client.data_classes.location_filters import LocationFilterWrite + assert isinstance(location_filter, LocationFilterWrite) + assert location_filter.external_id == "location_filter_location_1" + assert location_filter.name == "Location 1" + assert location_filter.description == "InField location, migrated from old location configuration" + # Check that instanceSpaces is populated with both sourceDataInstanceSpace and appDataInstanceSpace + # Note: Detailed instanceSpaces tests are in test_infield_instance_spaces.py + assert location_filter.instance_spaces is not None + assert len(location_filter.instance_spaces) == 2 + assert "source_space_1" in location_filter.instance_spaces + assert "app_space_1" in location_filter.instance_spaces + + # Check InFieldLocationConfig node (deployed via Data Modeling Instance API) + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 1 + location_node = created_nodes[0] + assert isinstance(location_node, NodeApply) + assert location_node.external_id == "location_1" + assert location_node.space == "APM_Config" + assert len(location_node.sources) == 1 + assert location_node.sources[0].source.external_id == "InFieldLocationConfig" + + # Check properties - should have rootLocationExternalId referencing the location filter + location_props = location_node.sources[0].properties + assert location_props["rootLocationExternalId"] == "location_filter_location_1" + + def test_create_infield_v2_configs_multiple_locations( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test migration with multiple root locations.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "multi_location_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "customerDataSpaceId": "APM_SourceData", + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "loc1", + "assetExternalId": "asset1", + }, + { + "externalId": "loc2", + "assetExternalId": "asset2", + }, + ], + }, + } + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + # LocationFilters are now deployed separately via Location Filters API + assert "location filters" in results + location_filter_result = results["location filters"] + assert location_filter_result.created == 2 + + # InFieldLocationConfig nodes are deployed via Data Modeling Instance API + assert "nodes" in results + node_result = results["nodes"] + assert node_result.created == 2 + + # Check LocationFilters (they are now LocationFilter resources, not Node resources) + created_location_filters = toolkit_client_approval.created_resources.get("LocationFilter", []) + assert len(created_location_filters) == 2 + location_filter_ids = [lf.external_id for lf in created_location_filters] + assert "location_filter_loc1" in location_filter_ids + assert "location_filter_loc2" in location_filter_ids + + # Check InFieldLocationConfig nodes + created_nodes = toolkit_client_approval.created_resources.get("Node", []) + assert len(created_nodes) == 2 + + # Should have two location config nodes + location_nodes = [n for n in created_nodes if n.external_id in ["loc1", "loc2"]] + assert len(location_nodes) == 2 + + # Verify location configs reference their location filters + loc1_node = next((n for n in location_nodes if n.external_id == "loc1"), None) + loc2_node = next((n for n in location_nodes if n.external_id == "loc2"), None) + assert loc1_node is not None + assert loc2_node is not None + assert loc1_node.sources[0].properties["rootLocationExternalId"] == "location_filter_loc1" + assert loc2_node.sources[0].properties["rootLocationExternalId"] == "location_filter_loc2" + + def test_create_infield_v2_configs_empty(self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path) -> None: + """Test migration with no root locations.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "empty_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [], + }, + } + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + assert "nodes" in results + result = results["nodes"] + # No locations, so no nodes created + assert result.created == 0 + + def test_create_infield_v2_configs_no_feature_config( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test migration with missing feature configuration.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "no_feature_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": {}, + } + }, + } + ) + + # Add the APMConfig node to the approval client so it can be retrieved + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + assert "nodes" in results + result = results["nodes"] + assert result.created == 0 + + def test_external_id_generation_with_external_id( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test external ID generation when externalId exists in old config.""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "test_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + "externalId": "my_location", + "assetExternalId": "my_asset", + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources["Node"] + location_node = next((n for n in created_nodes if n.external_id == "my_location"), None) + assert location_node is not None + # Should use externalId directly when it exists + assert location_node.external_id == "my_location" + + def test_external_id_generation_with_only_asset_external_id( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test external ID generation when only assetExternalId exists (should add index postfix).""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "test_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + # No externalId, only assetExternalId + "assetExternalId": "shared_asset", + }, + { + # No externalId, only assetExternalId (same asset) + "assetExternalId": "shared_asset", + }, + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources["Node"] + # Should have 2 location config nodes with index postfix for uniqueness + location_nodes = [n for n in created_nodes if n.external_id.startswith("shared_asset_")] + assert len(location_nodes) == 2 + # Should have index 0 and 1 + external_ids = [n.external_id for n in location_nodes] + assert "shared_asset_0" in external_ids + assert "shared_asset_1" in external_ids + + def test_external_id_generation_with_neither_external_id( + self, toolkit_client_approval: ApprovalToolkitClient, tmp_path: Path + ) -> None: + """Test external ID generation when neither externalId nor assetExternalId exists (should generate UUID).""" + toolkit_client_approval.append(DataModel, COGNITE_MIGRATION_MODEL) + + apm_config_node = Node._load( + { + "space": "APM_Config", + "externalId": "test_config", + "version": 1, + "lastUpdatedTime": 1, + "createdTime": 1, + "properties": { + "APM_Config": { + "APM_Config/1": { + "featureConfiguration": { + "rootLocationConfigurations": [ + { + # No externalId, no assetExternalId + } + ], + }, + } + } + }, + } + ) + + toolkit_client_approval.append(Node, apm_config_node) + + results = MigrationCommand(silent=True).create( + client=toolkit_client_approval.client, + creator=InfieldV2ConfigCreator(toolkit_client_approval.client), + dry_run=False, + verbose=False, + output_dir=tmp_path, + ) + + created_nodes = toolkit_client_approval.created_resources["Node"] + # Should have location config with generated UUID + location_nodes = [n for n in created_nodes if n.external_id.startswith("infield_location_")] + assert len(location_nodes) == 1 + # Should start with infield_location_ prefix + assert location_nodes[0].external_id.startswith("infield_location_") + +