From 922772365f1d13f73d3a4b1f5b466b827f8583a8 Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Wed, 21 Oct 2026 14:06:25 +0100 Subject: [PATCH 1/2] FND-3879: add pyatlan_v9 creators for the 2 Dataplex types Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017CGtFm63WJtCpeBjakhwhx --- .../assets/_overlays/bigquery_routine.py | 63 ++++++++ .../_overlays/gcp_dataplex_aspect_type.py | 53 ++++++ pyatlan_v9/model/assets/bigquery_routine.py | 67 +++++++- .../model/assets/gcp_dataplex_aspect_type.py | 63 ++++++++ tests_v9/unit/model/dataplex_test.py | 153 ++++++++++++++++++ 5 files changed, 396 insertions(+), 3 deletions(-) create mode 100644 pyatlan_v9/model/assets/_overlays/bigquery_routine.py create mode 100644 pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py create mode 100644 tests_v9/unit/model/dataplex_test.py diff --git a/pyatlan_v9/model/assets/_overlays/bigquery_routine.py b/pyatlan_v9/model/assets/_overlays/bigquery_routine.py new file mode 100644 index 000000000..8fcbac0ac --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/bigquery_routine.py @@ -0,0 +1,63 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + definition: str, + schema_qualified_name: str, + schema_name: str | None = None, + database_name: str | None = None, + database_qualified_name: str | None = None, + connection_qualified_name: str | None = None, + ) -> "BigqueryRoutine": + """ + Builds the minimal object necessary to create a BigqueryRoutine. + + The qualifiedName is ``{schema_qualified_name}/_routines_/{name}``, the shape + BigqueryRoutine assets carry on tenants and that atlan-dataplex-app aspect-sync + parses. Current atlan-bigquery-app crawls emit ``Function`` / ``Procedure`` + instead of BigqueryRoutine. + """ + validate_required_fields( + ["name", "definition", "schema_qualified_name"], + [name, definition, schema_qualified_name], + ) + + fields = schema_qualified_name.split("/") + if len(fields) != 5: + raise ValueError( + f"Invalid schema_qualified_name: {schema_qualified_name}. " + "Expected format: default/bigquery/connection_id/project/dataset" + ) + + connector_name = fields[1] + connection_qn = ( + connection_qualified_name or f"{fields[0]}/{fields[1]}/{fields[2]}" + ) + db_name = database_name or fields[3] + sch_name = schema_name or fields[4] + db_qualified_name = database_qualified_name or f"{connection_qn}/{db_name}" + + return cls( + name=name, + definition=definition, + qualified_name=f"{schema_qualified_name}/_routines_/{name}", + database_name=db_name, + database_qualified_name=db_qualified_name, + schema_name=sch_name, + schema_qualified_name=schema_qualified_name, + connector_name=connector_name, + connection_qualified_name=connection_qn, + atlan_schema=RelatedSchema(qualified_name=schema_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "BigqueryRoutine": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "BigqueryRoutine": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py b/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py new file mode 100644 index 000000000..86df8be3c --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py @@ -0,0 +1,53 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + resource_name: str, + connection_qualified_name: str, + ) -> "GCPDataplexAspectType": + """ + Builds the minimal object necessary to create a GCPDataplexAspectType. + + ``resource_name`` is the full GCP resource name, + ``projects/{project}/locations/{location}/aspectTypes/{id}``. The qualifiedName + is ``{connection_qualified_name}/aspect-type/{resource_name}`` and the name is + the aspect type id, as atlan-dataplex-app emits them. + """ + validate_required_fields( + ["resource_name", "connection_qualified_name"], + [resource_name, connection_qualified_name], + ) + parts = resource_name.split("/") + if ( + len(parts) != 6 + or parts[0] != "projects" + or parts[2] != "locations" + or parts[4] != "aspectTypes" + or not all(part.strip() for part in parts) + ): + raise ValueError( + f"Invalid resource_name: {resource_name}. " + "Expected format: projects/{project}/locations/{location}/aspectTypes/{id}" + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=parts[5], + qualified_name=f"{connection_qualified_name}/aspect-type/{resource_name}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + gcp_dataplex_aspect_type_resource_name=resource_name, + gcp_dataplex_aspect_type_project=parts[1], + gcp_dataplex_aspect_type_location=parts[3], + ) + + @classmethod + def create(cls, **kwargs) -> "GCPDataplexAspectType": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "GCPDataplexAspectType": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/bigquery_routine.py b/pyatlan_v9/model/assets/bigquery_routine.py index c27a63b7a..47341eba9 100644 --- a/pyatlan_v9/model/assets/bigquery_routine.py +++ b/pyatlan_v9/model/assets/bigquery_routine.py @@ -26,6 +26,7 @@ ) from pyatlan_v9.model.serde import Serde, get_serde from pyatlan_v9.model.transform import register_asset +from pyatlan_v9.utils import init_guid, validate_required_fields from .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -509,9 +510,7 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= - _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile( - r"^.+/[^/]+/[^/]+/[^/]+$" - ) + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/_routines_/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -589,6 +588,68 @@ def relate(self) -> "RelatedBigqueryRoutine": return RelatedBigqueryRoutine(guid=self.guid) return RelatedBigqueryRoutine(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + definition: str, + schema_qualified_name: str, + schema_name: str | None = None, + database_name: str | None = None, + database_qualified_name: str | None = None, + connection_qualified_name: str | None = None, + ) -> "BigqueryRoutine": + """ + Builds the minimal object necessary to create a BigqueryRoutine. + + The qualifiedName is ``{schema_qualified_name}/_routines_/{name}``, the shape + BigqueryRoutine assets carry on tenants and that atlan-dataplex-app aspect-sync + parses. Current atlan-bigquery-app crawls emit ``Function`` / ``Procedure`` + instead of BigqueryRoutine. + """ + validate_required_fields( + ["name", "definition", "schema_qualified_name"], + [name, definition, schema_qualified_name], + ) + + fields = schema_qualified_name.split("/") + if len(fields) != 5: + raise ValueError( + f"Invalid schema_qualified_name: {schema_qualified_name}. " + "Expected format: default/bigquery/connection_id/project/dataset" + ) + + connector_name = fields[1] + connection_qn = ( + connection_qualified_name or f"{fields[0]}/{fields[1]}/{fields[2]}" + ) + db_name = database_name or fields[3] + sch_name = schema_name or fields[4] + db_qualified_name = database_qualified_name or f"{connection_qn}/{db_name}" + + return cls( + name=name, + definition=definition, + qualified_name=f"{schema_qualified_name}/_routines_/{name}", + database_name=db_name, + database_qualified_name=db_qualified_name, + schema_name=sch_name, + schema_qualified_name=schema_qualified_name, + connector_name=connector_name, + connection_qualified_name=connection_qn, + atlan_schema=RelatedSchema(qualified_name=schema_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "BigqueryRoutine": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "BigqueryRoutine": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py b/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py index 36b35625d..d41996a61 100644 --- a/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py +++ b/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py @@ -14,6 +14,7 @@ from __future__ import annotations +import re from typing import Any, ClassVar, Dict, List, Union from msgspec import UNSET, UnsetType @@ -24,6 +25,7 @@ ) from pyatlan_v9.model.serde import Serde, get_serde from pyatlan_v9.model.transform import register_asset +from pyatlan_v9.utils import init_guid, validate_required_fields from .anomalo_related import RelatedAnomaloCheck from .app_related import RelatedApplication, RelatedApplicationField @@ -229,6 +231,10 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile( + r"^.+/aspect-type/projects/[^/]+/locations/[^/]+/aspectTypes/[^/]+$" + ) + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this GCPDataplexAspectType instance. @@ -253,6 +259,11 @@ def validate(self, for_creation: bool = False) -> None: errors.append("name is required") if self.qualified_name is UNSET or self.qualified_name is None: errors.append("qualified_name is required") + elif not self._QUALIFIED_NAME_PATTERN.match(self.qualified_name): + errors.append( + f"qualified_name '{self.qualified_name}' does not match expected " + f"pattern: {self._QUALIFIED_NAME_PATTERN.pattern}" + ) if errors: raise ValueError(f"GCPDataplexAspectType validation failed: {errors}") @@ -285,6 +296,58 @@ def relate(self) -> "RelatedGCPDataplexAspectType": return RelatedGCPDataplexAspectType(guid=self.guid) return RelatedGCPDataplexAspectType(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + resource_name: str, + connection_qualified_name: str, + ) -> "GCPDataplexAspectType": + """ + Builds the minimal object necessary to create a GCPDataplexAspectType. + + ``resource_name`` is the full GCP resource name, + ``projects/{project}/locations/{location}/aspectTypes/{id}``. The qualifiedName + is ``{connection_qualified_name}/aspect-type/{resource_name}`` and the name is + the aspect type id, as atlan-dataplex-app emits them. + """ + validate_required_fields( + ["resource_name", "connection_qualified_name"], + [resource_name, connection_qualified_name], + ) + parts = resource_name.split("/") + if ( + len(parts) != 6 + or parts[0] != "projects" + or parts[2] != "locations" + or parts[4] != "aspectTypes" + or not all(part.strip() for part in parts) + ): + raise ValueError( + f"Invalid resource_name: {resource_name}. " + "Expected format: projects/{project}/locations/{location}/aspectTypes/{id}" + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=parts[5], + qualified_name=f"{connection_qualified_name}/aspect-type/{resource_name}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + gcp_dataplex_aspect_type_resource_name=resource_name, + gcp_dataplex_aspect_type_project=parts[1], + gcp_dataplex_aspect_type_location=parts[3], + ) + + @classmethod + def create(cls, **kwargs) -> "GCPDataplexAspectType": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "GCPDataplexAspectType": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/tests_v9/unit/model/dataplex_test.py b/tests_v9/unit/model/dataplex_test.py new file mode 100644 index 000000000..f48cc8d9e --- /dev/null +++ b/tests_v9/unit/model/dataplex_test.py @@ -0,0 +1,153 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for the Dataplex creators in pyatlan_v9 (GCPDataplexAspectType, BigqueryRoutine). + +The expected qualifiedNames are the exact strings the connectors emit: +GCPDataplexAspectType from atlan-dataplex-app (``aspect_type_qn``), BigqueryRoutine in the +``_routines_`` shape that atlan-dataplex-app aspect-sync parses and existing tenant assets carry. +A creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import BigqueryRoutine, GCPDataplexAspectType + +DATAPLEX_CONNECTION_QN = "default/dataplex/1712345678" +RESOURCE_NAME = "projects/my-project/locations/us-central1/aspectTypes/data-owner" +ASPECT_TYPE_QN = ( + "default/dataplex/1712345678/aspect-type/" + "projects/my-project/locations/us-central1/aspectTypes/data-owner" +) + +BQ_CONNECTION_QN = "default/bigquery/1712345678" +SCHEMA_QN = "default/bigquery/1712345678/my-project/my_dataset" +ROUTINE_NAME = "clean_orders" +DEFINITION = "CREATE PROCEDURE my_dataset.clean_orders() BEGIN SELECT 1; END" +ROUTINE_QN = "default/bigquery/1712345678/my-project/my_dataset/_routines_/clean_orders" + + +def test_aspect_type_creator_matches_app_qualified_name(): + sut = GCPDataplexAspectType.creator( + resource_name=RESOURCE_NAME, + connection_qualified_name=DATAPLEX_CONNECTION_QN, + ) + + assert sut.qualified_name == ASPECT_TYPE_QN + assert sut.name == "data-owner" + assert sut.connection_qualified_name == DATAPLEX_CONNECTION_QN + assert sut.connector_name == "dataplex" + assert sut.gcp_dataplex_aspect_type_resource_name == RESOURCE_NAME + assert sut.gcp_dataplex_aspect_type_project == "my-project" + assert sut.gcp_dataplex_aspect_type_location == "us-central1" + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +@pytest.mark.parametrize( + "resource_name, connection_qualified_name, message", + [ + (None, DATAPLEX_CONNECTION_QN, "resource_name is required"), + (RESOURCE_NAME, None, "connection_qualified_name is required"), + (" ", DATAPLEX_CONNECTION_QN, "resource_name"), + (RESOURCE_NAME, "", "connection_qualified_name"), + ("data-owner", DATAPLEX_CONNECTION_QN, "Invalid resource_name"), + ( + "projects/p/locations/l/entryTypes/data-owner", + DATAPLEX_CONNECTION_QN, + "Invalid resource_name", + ), + ( + "projects/p/locations/ /aspectTypes/data-owner", + DATAPLEX_CONNECTION_QN, + "Invalid resource_name", + ), + ], +) +def test_aspect_type_creator_rejects_missing_or_invalid_inputs( + resource_name, connection_qualified_name, message +): + with pytest.raises(ValueError, match=message): + GCPDataplexAspectType.creator( + resource_name=resource_name, + connection_qualified_name=connection_qualified_name, + ) + + +def test_aspect_type_create_alias(): + sut = GCPDataplexAspectType.create( + resource_name=RESOURCE_NAME, + connection_qualified_name=DATAPLEX_CONNECTION_QN, + ) + + assert sut.qualified_name == ASPECT_TYPE_QN + + +def test_routine_creator_matches_app_qualified_name(): + sut = BigqueryRoutine.creator( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=SCHEMA_QN, + ) + + assert sut.qualified_name == ROUTINE_QN + assert sut.name == ROUTINE_NAME + assert sut.definition == DEFINITION + assert sut.connection_qualified_name == BQ_CONNECTION_QN + assert sut.connector_name == "bigquery" + assert sut.database_name == "my-project" + assert sut.database_qualified_name == f"{BQ_CONNECTION_QN}/my-project" + assert sut.schema_name == "my_dataset" + assert sut.schema_qualified_name == SCHEMA_QN + assert sut.atlan_schema.qualified_name == SCHEMA_QN + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_routine_creator_uses_explicit_parent_names(): + sut = BigqueryRoutine.creator( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=SCHEMA_QN, + schema_name="My Dataset", + database_name="My Project", + database_qualified_name=f"{BQ_CONNECTION_QN}/my-project", + connection_qualified_name=BQ_CONNECTION_QN, + ) + + assert sut.qualified_name == ROUTINE_QN + assert sut.schema_name == "My Dataset" + assert sut.database_name == "My Project" + assert sut.database_qualified_name == f"{BQ_CONNECTION_QN}/my-project" + + +@pytest.mark.parametrize( + "name, definition, schema_qualified_name, message", + [ + (None, DEFINITION, SCHEMA_QN, "name is required"), + (ROUTINE_NAME, None, SCHEMA_QN, "definition is required"), + (ROUTINE_NAME, DEFINITION, None, "schema_qualified_name is required"), + (" ", DEFINITION, SCHEMA_QN, "name"), + (ROUTINE_NAME, DEFINITION, "", "schema_qualified_name"), + (ROUTINE_NAME, DEFINITION, BQ_CONNECTION_QN, "Invalid schema_qualified_name"), + ], +) +def test_routine_creator_rejects_missing_or_invalid_inputs( + name, definition, schema_qualified_name, message +): + with pytest.raises(ValueError, match=message): + BigqueryRoutine.creator( + name=name, + definition=definition, + schema_qualified_name=schema_qualified_name, + ) + + +def test_routine_create_alias(): + sut = BigqueryRoutine.create( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=SCHEMA_QN, + ) + + assert sut.qualified_name == ROUTINE_QN From f3b2dbc52c336cc77e53e653f9b04506bb4ef108 Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Wed, 21 Oct 2026 14:12:49 +0100 Subject: [PATCH 2/2] FND-3879: typed updater/trim, per-type unit tests, Dataplex integration test Adds AtlanConnectorType.DATAPLEX (needed by Connection.creator) and regenerates the two classes from the rebased models branch. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017CGtFm63WJtCpeBjakhwhx --- pyatlan/model/enums.py | 1 + .../assets/_overlays/bigquery_routine.py | 12 ++ .../_overlays/gcp_dataplex_aspect_type.py | 12 ++ pyatlan_v9/model/assets/bigquery_routine.py | 14 ++ .../model/assets/gcp_dataplex_aspect_type.py | 14 ++ tests_v9/integration/dataplex_asset_test.py | 140 +++++++++++++++ tests_v9/unit/model/bigquery_routine_test.py | 170 ++++++++++++++++++ tests_v9/unit/model/dataplex_test.py | 153 ---------------- .../model/gcp_dataplex_aspect_type_test.py | 139 ++++++++++++++ 9 files changed, 502 insertions(+), 153 deletions(-) create mode 100644 tests_v9/integration/dataplex_asset_test.py create mode 100644 tests_v9/unit/model/bigquery_routine_test.py delete mode 100644 tests_v9/unit/model/dataplex_test.py create mode 100644 tests_v9/unit/model/gcp_dataplex_aspect_type_test.py diff --git a/pyatlan/model/enums.py b/pyatlan/model/enums.py index 889e27bed..08699c138 100644 --- a/pyatlan/model/enums.py +++ b/pyatlan/model/enums.py @@ -268,6 +268,7 @@ def get_connector_name( API = ("api", AtlanConnectionCategory.API) DYNAMODB = ("dynamodb", AtlanConnectionCategory.WAREHOUSE) GCS = ("gcs", AtlanConnectionCategory.OBJECT_STORE) + DATAPLEX = ("dataplex", AtlanConnectionCategory.SAAS) HIVE = ("hive", AtlanConnectionCategory.WAREHOUSE) SAPHANA = ("sap-hana", AtlanConnectionCategory.WAREHOUSE) ADLS = ("adls", AtlanConnectionCategory.OBJECT_STORE) diff --git a/pyatlan_v9/model/assets/_overlays/bigquery_routine.py b/pyatlan_v9/model/assets/_overlays/bigquery_routine.py index 8fcbac0ac..67fb03e89 100644 --- a/pyatlan_v9/model/assets/_overlays/bigquery_routine.py +++ b/pyatlan_v9/model/assets/_overlays/bigquery_routine.py @@ -54,6 +54,18 @@ def creator( atlan_schema=RelatedSchema(qualified_name=schema_qualified_name), ) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "BigqueryRoutine": + """Create a BigqueryRoutine instance for update operations.""" + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + + def trim_to_required(self) -> "BigqueryRoutine": + """Return only fields required for update operations.""" + return BigqueryRoutine.updater( + qualified_name=self.qualified_name, name=self.name + ) + @classmethod def create(cls, **kwargs) -> "BigqueryRoutine": return cls.creator(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py b/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py index 86df8be3c..c8845ef90 100644 --- a/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py +++ b/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py @@ -44,6 +44,18 @@ def creator( gcp_dataplex_aspect_type_location=parts[3], ) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "GCPDataplexAspectType": + """Create a GCPDataplexAspectType instance for update operations.""" + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + + def trim_to_required(self) -> "GCPDataplexAspectType": + """Return only fields required for update operations.""" + return GCPDataplexAspectType.updater( + qualified_name=self.qualified_name, name=self.name + ) + @classmethod def create(cls, **kwargs) -> "GCPDataplexAspectType": return cls.creator(**kwargs) diff --git a/pyatlan_v9/model/assets/bigquery_routine.py b/pyatlan_v9/model/assets/bigquery_routine.py index 47341eba9..ad594970c 100644 --- a/pyatlan_v9/model/assets/bigquery_routine.py +++ b/pyatlan_v9/model/assets/bigquery_routine.py @@ -642,6 +642,18 @@ def creator( atlan_schema=RelatedSchema(qualified_name=schema_qualified_name), ) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "BigqueryRoutine": + """Create a BigqueryRoutine instance for update operations.""" + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + + def trim_to_required(self) -> "BigqueryRoutine": + """Return only fields required for update operations.""" + return BigqueryRoutine.updater( + qualified_name=self.qualified_name, name=self.name + ) + @classmethod def create(cls, **kwargs) -> "BigqueryRoutine": return cls.creator(**kwargs) @@ -1271,6 +1283,7 @@ def _bigquery_routine_to_nested( is_incomplete=bigquery_routine.is_incomplete, provenance_type=bigquery_routine.provenance_type, home_id=bigquery_routine.home_id, + delete_handler=bigquery_routine.delete_handler, depth=bigquery_routine.depth, immediate_upstream=bigquery_routine.immediate_upstream, immediate_downstream=bigquery_routine.immediate_downstream, @@ -1319,6 +1332,7 @@ def _bigquery_routine_from_nested(nested: BigqueryRoutineNested) -> BigqueryRout "is_incomplete": nested.is_incomplete, "provenance_type": nested.provenance_type, "home_id": nested.home_id, + "delete_handler": nested.delete_handler, "depth": nested.depth, "immediate_upstream": nested.immediate_upstream, "immediate_downstream": nested.immediate_downstream, diff --git a/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py b/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py index d41996a61..159c7e3aa 100644 --- a/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py +++ b/pyatlan_v9/model/assets/gcp_dataplex_aspect_type.py @@ -340,6 +340,18 @@ def creator( gcp_dataplex_aspect_type_location=parts[3], ) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "GCPDataplexAspectType": + """Create a GCPDataplexAspectType instance for update operations.""" + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + + def trim_to_required(self) -> "GCPDataplexAspectType": + """Return only fields required for update operations.""" + return GCPDataplexAspectType.updater( + qualified_name=self.qualified_name, name=self.name + ) + @classmethod def create(cls, **kwargs) -> "GCPDataplexAspectType": return cls.creator(**kwargs) @@ -670,6 +682,7 @@ def _gcp_dataplex_aspect_type_to_nested( is_incomplete=gcp_dataplex_aspect_type.is_incomplete, provenance_type=gcp_dataplex_aspect_type.provenance_type, home_id=gcp_dataplex_aspect_type.home_id, + delete_handler=gcp_dataplex_aspect_type.delete_handler, depth=gcp_dataplex_aspect_type.depth, immediate_upstream=gcp_dataplex_aspect_type.immediate_upstream, immediate_downstream=gcp_dataplex_aspect_type.immediate_downstream, @@ -720,6 +733,7 @@ def _gcp_dataplex_aspect_type_from_nested( "is_incomplete": nested.is_incomplete, "provenance_type": nested.provenance_type, "home_id": nested.home_id, + "delete_handler": nested.delete_handler, "depth": nested.depth, "immediate_upstream": nested.immediate_upstream, "immediate_downstream": nested.immediate_downstream, diff --git a/tests_v9/integration/dataplex_asset_test.py b/tests_v9/integration/dataplex_asset_test.py new file mode 100644 index 000000000..8b09c958b --- /dev/null +++ b/tests_v9/integration/dataplex_asset_test.py @@ -0,0 +1,140 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +from typing import Generator + +import pytest + +from pyatlan_v9.client.atlan import AtlanClient +from pyatlan_v9.model.assets import ( + BigqueryRoutine, + Connection, + Database, + GCPDataplexAspectType, + Schema, +) +from pyatlan_v9.model.enums import AtlanConnectorType +from tests_v9.integration.client import TestId, delete_asset +from tests_v9.integration.connection_test import create_connection + +MODULE_NAME = TestId.make_unique("DATAPLEX") + +ASPECT_TYPE_ID = f"aspect_{MODULE_NAME}" +RESOURCE_NAME = f"projects/sdk-test/locations/global/aspectTypes/{ASPECT_TYPE_ID}" +DATABASE_NAME = f"project_{MODULE_NAME}" +SCHEMA_NAME = f"dataset_{MODULE_NAME}" +ROUTINE_NAME = f"routine_{MODULE_NAME}" +ROUTINE_DEFINITION = "CREATE PROCEDURE ds.p() BEGIN SELECT 1; END" + + +@pytest.fixture(scope="module") +def dataplex_connection(client: AtlanClient) -> Generator[Connection, None, None]: + result = create_connection( + client=client, name=MODULE_NAME, connector_type=AtlanConnectorType.DATAPLEX + ) + yield result + delete_asset(client, guid=result.guid, asset_type=Connection) + + +@pytest.fixture(scope="module") +def bigquery_connection(client: AtlanClient) -> Generator[Connection, None, None]: + result = create_connection( + client=client, name=MODULE_NAME, connector_type=AtlanConnectorType.BIGQUERY + ) + yield result + delete_asset(client, guid=result.guid, asset_type=Connection) + + +@pytest.fixture(scope="module") +def aspect_type( + client: AtlanClient, dataplex_connection: Connection +) -> Generator[GCPDataplexAspectType, None, None]: + to_create = GCPDataplexAspectType.creator( + resource_name=RESOURCE_NAME, + connection_qualified_name=dataplex_connection.qualified_name, + ) + result = client.asset.save(to_create).assets_created( + asset_type=GCPDataplexAspectType + )[0] + yield result + delete_asset(client, guid=result.guid, asset_type=GCPDataplexAspectType) + + +@pytest.fixture(scope="module") +def database( + client: AtlanClient, bigquery_connection: Connection +) -> Generator[Database, None, None]: + to_create = Database.creator( + name=DATABASE_NAME, + connection_qualified_name=bigquery_connection.qualified_name, + ) + result = client.asset.save(to_create).assets_created(asset_type=Database)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=Database) + + +@pytest.fixture(scope="module") +def schema(client: AtlanClient, database: Database) -> Generator[Schema, None, None]: + to_create = Schema.creator( + name=SCHEMA_NAME, database_qualified_name=database.qualified_name + ) + result = client.asset.save(to_create).assets_created(asset_type=Schema)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=Schema) + + +@pytest.fixture(scope="module") +def routine( + client: AtlanClient, schema: Schema +) -> Generator[BigqueryRoutine, None, None]: + to_create = BigqueryRoutine.creator( + name=ROUTINE_NAME, + definition=ROUTINE_DEFINITION, + schema_qualified_name=schema.qualified_name, + ) + result = client.asset.save(to_create).assets_created(asset_type=BigqueryRoutine)[0] + yield result + delete_asset(client, guid=result.guid, asset_type=BigqueryRoutine) + + +def test_aspect_type( + client: AtlanClient, + dataplex_connection: Connection, + aspect_type: GCPDataplexAspectType, +): + assert aspect_type.guid + retrieved = client.asset.get_by_guid( + aspect_type.guid, asset_type=GCPDataplexAspectType + ) + assert retrieved.qualified_name == ( + f"{dataplex_connection.qualified_name}/aspect-type/{RESOURCE_NAME}" + ) + assert retrieved.name == ASPECT_TYPE_ID + assert retrieved.gcp_dataplex_aspect_type_resource_name == RESOURCE_NAME + assert retrieved.gcp_dataplex_aspect_type_project == "sdk-test" + assert retrieved.gcp_dataplex_aspect_type_location == "global" + assert retrieved.connector_name == AtlanConnectorType.DATAPLEX + assert retrieved.connection_qualified_name == dataplex_connection.qualified_name + + +def test_routine( + client: AtlanClient, + bigquery_connection: Connection, + schema: Schema, + routine: BigqueryRoutine, +): + assert routine.guid + retrieved = client.asset.get_by_guid( + routine.guid, asset_type=BigqueryRoutine, ignore_relationships=False + ) + assert retrieved.qualified_name == ( + f"{bigquery_connection.qualified_name}/{DATABASE_NAME}/{SCHEMA_NAME}" + f"/_routines_/{ROUTINE_NAME}" + ) + assert retrieved.name == ROUTINE_NAME + assert retrieved.definition == ROUTINE_DEFINITION + assert retrieved.schema_qualified_name == schema.qualified_name + assert retrieved.database_name == DATABASE_NAME + assert retrieved.connector_name == AtlanConnectorType.BIGQUERY + assert retrieved.connection_qualified_name == bigquery_connection.qualified_name + assert retrieved.atlan_schema.guid == schema.guid diff --git a/tests_v9/unit/model/bigquery_routine_test.py b/tests_v9/unit/model/bigquery_routine_test.py new file mode 100644 index 000000000..35a4ca182 --- /dev/null +++ b/tests_v9/unit/model/bigquery_routine_test.py @@ -0,0 +1,170 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for BigqueryRoutine model in pyatlan_v9. + +The parity test uses the exact ``{schemaQN}/_routines_/{name}`` qualifiedName that +BigqueryRoutine assets carry and that atlan-dataplex-app aspect-sync parses +(``entry_names._ROUTINES_MARKER``), so a creator change that would re-key existing +routines fails here. +""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import BigqueryRoutine + +CONNECTION_QUALIFIED_NAME = "default/bigquery/123456789" +CONNECTOR_NAME = "bigquery" +DATABASE_NAME = "my-project" +DATABASE_QUALIFIED_NAME = f"{CONNECTION_QUALIFIED_NAME}/{DATABASE_NAME}" +SCHEMA_NAME = "my_dataset" +SCHEMA_QUALIFIED_NAME = f"{DATABASE_QUALIFIED_NAME}/{SCHEMA_NAME}" +ROUTINE_NAME = "clean_orders" +DEFINITION = "CREATE PROCEDURE my_dataset.clean_orders() BEGIN SELECT 1; END" +ROUTINE_QUALIFIED_NAME = f"{SCHEMA_QUALIFIED_NAME}/_routines_/{ROUTINE_NAME}" + + +def test_creator_matches_app_qualified_name(): + """Mirror the routine qualifiedName atlan-dataplex-app aspect-sync parses.""" + sut = BigqueryRoutine.creator( + name="clean_orders", + definition=DEFINITION, + schema_qualified_name="default/bigquery/1700000000/acme-prod/sales", + ) + + assert ( + sut.qualified_name + == "default/bigquery/1700000000/acme-prod/sales/_routines_/clean_orders" + ) + sut.validate(for_creation=True) + + +@pytest.mark.parametrize( + "name, definition, schema_qualified_name, message", + [ + (None, DEFINITION, SCHEMA_QUALIFIED_NAME, "name is required"), + (" ", DEFINITION, SCHEMA_QUALIFIED_NAME, "name cannot be blank"), + (ROUTINE_NAME, None, SCHEMA_QUALIFIED_NAME, "definition is required"), + (ROUTINE_NAME, " ", SCHEMA_QUALIFIED_NAME, "definition cannot be blank"), + (ROUTINE_NAME, DEFINITION, None, "schema_qualified_name is required"), + (ROUTINE_NAME, DEFINITION, " ", "schema_qualified_name cannot be blank"), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + name: str, definition: str, schema_qualified_name: str, message: str +): + """Test creator validates required parameters.""" + with pytest.raises(ValueError, match=message): + BigqueryRoutine.creator( + name=name, + definition=definition, + schema_qualified_name=schema_qualified_name, + ) + + +@pytest.mark.parametrize( + "schema_qualified_name", + [CONNECTION_QUALIFIED_NAME, f"{SCHEMA_QUALIFIED_NAME}/extra"], +) +def test_creator_with_malformed_schema_qualified_name_raises_value_error( + schema_qualified_name: str, +): + """Test creator rejects a parent qualifiedName that is not a BigQuery dataset.""" + with pytest.raises(ValueError, match="Invalid schema_qualified_name"): + BigqueryRoutine.creator( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=schema_qualified_name, + ) + + +def test_creator_derives_parent_attributes(): + """Test creator derives the denormalised parent attributes from the schema qn.""" + sut = BigqueryRoutine.creator( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=SCHEMA_QUALIFIED_NAME, + ) + + assert sut.type_name == "BigqueryRoutine" + assert sut.name == ROUTINE_NAME + assert sut.definition == DEFINITION + assert sut.qualified_name == ROUTINE_QUALIFIED_NAME + assert sut.schema_name == SCHEMA_NAME + assert sut.schema_qualified_name == SCHEMA_QUALIFIED_NAME + assert sut.database_name == DATABASE_NAME + assert sut.database_qualified_name == DATABASE_QUALIFIED_NAME + assert sut.connection_qualified_name == CONNECTION_QUALIFIED_NAME + assert sut.connector_name == CONNECTOR_NAME + assert sut.atlan_schema.qualified_name == SCHEMA_QUALIFIED_NAME + assert sut.guid + sut.validate(for_creation=True) + + +def test_creator_with_explicit_parent_attributes(): + """Test explicit parent names override the derived ones without changing the qn.""" + sut = BigqueryRoutine.creator( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=SCHEMA_QUALIFIED_NAME, + schema_name="My Dataset", + database_name="My Project", + database_qualified_name=DATABASE_QUALIFIED_NAME, + connection_qualified_name=CONNECTION_QUALIFIED_NAME, + ) + + assert sut.qualified_name == ROUTINE_QUALIFIED_NAME + assert sut.schema_name == "My Dataset" + assert sut.database_name == "My Project" + assert sut.database_qualified_name == DATABASE_QUALIFIED_NAME + assert sut.connection_qualified_name == CONNECTION_QUALIFIED_NAME + + +def test_create_alias(): + """Test the deprecated create alias delegates to creator.""" + sut = BigqueryRoutine.create( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=SCHEMA_QUALIFIED_NAME, + ) + + assert sut.qualified_name == ROUTINE_QUALIFIED_NAME + + +@pytest.mark.parametrize( + "qualified_name, name, message", + [ + (None, ROUTINE_NAME, "qualified_name is required"), + (ROUTINE_QUALIFIED_NAME, None, "name is required"), + ], +) +def test_updater_with_invalid_parameter_raises_value_error( + qualified_name: str, name: str, message: str +): + """Test updater validates required parameters.""" + with pytest.raises(ValueError, match=message): + BigqueryRoutine.updater(qualified_name=qualified_name, name=name) + + +def test_updater(): + """Test updater creates a BigqueryRoutine for modification.""" + sut = BigqueryRoutine.updater( + qualified_name=ROUTINE_QUALIFIED_NAME, name=ROUTINE_NAME + ) + + assert sut.qualified_name == ROUTINE_QUALIFIED_NAME + assert sut.name == ROUTINE_NAME + + +def test_trim_to_required(): + """Test trim_to_required keeps only updater-required fields.""" + sut = BigqueryRoutine.creator( + name=ROUTINE_NAME, + definition=DEFINITION, + schema_qualified_name=SCHEMA_QUALIFIED_NAME, + ).trim_to_required() + + assert sut.qualified_name == ROUTINE_QUALIFIED_NAME + assert sut.name == ROUTINE_NAME + assert sut.definition is UNSET diff --git a/tests_v9/unit/model/dataplex_test.py b/tests_v9/unit/model/dataplex_test.py deleted file mode 100644 index f48cc8d9e..000000000 --- a/tests_v9/unit/model/dataplex_test.py +++ /dev/null @@ -1,153 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# Copyright 2026 Atlan Pte. Ltd. - -"""Unit tests for the Dataplex creators in pyatlan_v9 (GCPDataplexAspectType, BigqueryRoutine). - -The expected qualifiedNames are the exact strings the connectors emit: -GCPDataplexAspectType from atlan-dataplex-app (``aspect_type_qn``), BigqueryRoutine in the -``_routines_`` shape that atlan-dataplex-app aspect-sync parses and existing tenant assets carry. -A creator change that would re-key existing assets fails here. -""" - -import pytest - -from pyatlan_v9.model import BigqueryRoutine, GCPDataplexAspectType - -DATAPLEX_CONNECTION_QN = "default/dataplex/1712345678" -RESOURCE_NAME = "projects/my-project/locations/us-central1/aspectTypes/data-owner" -ASPECT_TYPE_QN = ( - "default/dataplex/1712345678/aspect-type/" - "projects/my-project/locations/us-central1/aspectTypes/data-owner" -) - -BQ_CONNECTION_QN = "default/bigquery/1712345678" -SCHEMA_QN = "default/bigquery/1712345678/my-project/my_dataset" -ROUTINE_NAME = "clean_orders" -DEFINITION = "CREATE PROCEDURE my_dataset.clean_orders() BEGIN SELECT 1; END" -ROUTINE_QN = "default/bigquery/1712345678/my-project/my_dataset/_routines_/clean_orders" - - -def test_aspect_type_creator_matches_app_qualified_name(): - sut = GCPDataplexAspectType.creator( - resource_name=RESOURCE_NAME, - connection_qualified_name=DATAPLEX_CONNECTION_QN, - ) - - assert sut.qualified_name == ASPECT_TYPE_QN - assert sut.name == "data-owner" - assert sut.connection_qualified_name == DATAPLEX_CONNECTION_QN - assert sut.connector_name == "dataplex" - assert sut.gcp_dataplex_aspect_type_resource_name == RESOURCE_NAME - assert sut.gcp_dataplex_aspect_type_project == "my-project" - assert sut.gcp_dataplex_aspect_type_location == "us-central1" - assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) - sut.validate(for_creation=True) - - -@pytest.mark.parametrize( - "resource_name, connection_qualified_name, message", - [ - (None, DATAPLEX_CONNECTION_QN, "resource_name is required"), - (RESOURCE_NAME, None, "connection_qualified_name is required"), - (" ", DATAPLEX_CONNECTION_QN, "resource_name"), - (RESOURCE_NAME, "", "connection_qualified_name"), - ("data-owner", DATAPLEX_CONNECTION_QN, "Invalid resource_name"), - ( - "projects/p/locations/l/entryTypes/data-owner", - DATAPLEX_CONNECTION_QN, - "Invalid resource_name", - ), - ( - "projects/p/locations/ /aspectTypes/data-owner", - DATAPLEX_CONNECTION_QN, - "Invalid resource_name", - ), - ], -) -def test_aspect_type_creator_rejects_missing_or_invalid_inputs( - resource_name, connection_qualified_name, message -): - with pytest.raises(ValueError, match=message): - GCPDataplexAspectType.creator( - resource_name=resource_name, - connection_qualified_name=connection_qualified_name, - ) - - -def test_aspect_type_create_alias(): - sut = GCPDataplexAspectType.create( - resource_name=RESOURCE_NAME, - connection_qualified_name=DATAPLEX_CONNECTION_QN, - ) - - assert sut.qualified_name == ASPECT_TYPE_QN - - -def test_routine_creator_matches_app_qualified_name(): - sut = BigqueryRoutine.creator( - name=ROUTINE_NAME, - definition=DEFINITION, - schema_qualified_name=SCHEMA_QN, - ) - - assert sut.qualified_name == ROUTINE_QN - assert sut.name == ROUTINE_NAME - assert sut.definition == DEFINITION - assert sut.connection_qualified_name == BQ_CONNECTION_QN - assert sut.connector_name == "bigquery" - assert sut.database_name == "my-project" - assert sut.database_qualified_name == f"{BQ_CONNECTION_QN}/my-project" - assert sut.schema_name == "my_dataset" - assert sut.schema_qualified_name == SCHEMA_QN - assert sut.atlan_schema.qualified_name == SCHEMA_QN - assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) - sut.validate(for_creation=True) - - -def test_routine_creator_uses_explicit_parent_names(): - sut = BigqueryRoutine.creator( - name=ROUTINE_NAME, - definition=DEFINITION, - schema_qualified_name=SCHEMA_QN, - schema_name="My Dataset", - database_name="My Project", - database_qualified_name=f"{BQ_CONNECTION_QN}/my-project", - connection_qualified_name=BQ_CONNECTION_QN, - ) - - assert sut.qualified_name == ROUTINE_QN - assert sut.schema_name == "My Dataset" - assert sut.database_name == "My Project" - assert sut.database_qualified_name == f"{BQ_CONNECTION_QN}/my-project" - - -@pytest.mark.parametrize( - "name, definition, schema_qualified_name, message", - [ - (None, DEFINITION, SCHEMA_QN, "name is required"), - (ROUTINE_NAME, None, SCHEMA_QN, "definition is required"), - (ROUTINE_NAME, DEFINITION, None, "schema_qualified_name is required"), - (" ", DEFINITION, SCHEMA_QN, "name"), - (ROUTINE_NAME, DEFINITION, "", "schema_qualified_name"), - (ROUTINE_NAME, DEFINITION, BQ_CONNECTION_QN, "Invalid schema_qualified_name"), - ], -) -def test_routine_creator_rejects_missing_or_invalid_inputs( - name, definition, schema_qualified_name, message -): - with pytest.raises(ValueError, match=message): - BigqueryRoutine.creator( - name=name, - definition=definition, - schema_qualified_name=schema_qualified_name, - ) - - -def test_routine_create_alias(): - sut = BigqueryRoutine.create( - name=ROUTINE_NAME, - definition=DEFINITION, - schema_qualified_name=SCHEMA_QN, - ) - - assert sut.qualified_name == ROUTINE_QN diff --git a/tests_v9/unit/model/gcp_dataplex_aspect_type_test.py b/tests_v9/unit/model/gcp_dataplex_aspect_type_test.py new file mode 100644 index 000000000..58119588c --- /dev/null +++ b/tests_v9/unit/model/gcp_dataplex_aspect_type_test.py @@ -0,0 +1,139 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for GCPDataplexAspectType model in pyatlan_v9. + +The parity test uses the exact qualifiedName atlan-dataplex-app emits today +(``aspect_type_qn``: ``{connectionQN}/aspect-type/{resourceName}``, FND-3838 census), +so a creator change that would re-key existing aspect types fails here. +""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import GCPDataplexAspectType + +CONNECTION_QUALIFIED_NAME = "default/dataplex/123456789" +CONNECTOR_NAME = "dataplex" +RESOURCE_NAME = "projects/my-project/locations/us-central1/aspectTypes/data-owner" +ASPECT_TYPE_NAME = "data-owner" +ASPECT_TYPE_QUALIFIED_NAME = f"{CONNECTION_QUALIFIED_NAME}/aspect-type/{RESOURCE_NAME}" + + +def test_creator_matches_app_qualified_name(): + """Mirror atlan-dataplex-app's extract_aspect_types inputs and output.""" + sut = GCPDataplexAspectType.creator( + resource_name="projects/acme-prod/locations/global/aspectTypes/pii", + connection_qualified_name="default/dataplex/1700000000", + ) + + assert ( + sut.qualified_name == "default/dataplex/1700000000/aspect-type/" + "projects/acme-prod/locations/global/aspectTypes/pii" + ) + assert sut.name == "pii" + sut.validate(for_creation=True) + + +@pytest.mark.parametrize( + "resource_name, connection_qualified_name, message", + [ + (None, CONNECTION_QUALIFIED_NAME, "resource_name is required"), + (" ", CONNECTION_QUALIFIED_NAME, "resource_name cannot be blank"), + (RESOURCE_NAME, None, "connection_qualified_name is required"), + (RESOURCE_NAME, " ", "connection_qualified_name cannot be blank"), + ], +) +def test_creator_with_missing_parameters_raise_value_error( + resource_name: str, connection_qualified_name: str, message: str +): + """Test creator validates required parameters.""" + with pytest.raises(ValueError, match=message): + GCPDataplexAspectType.creator( + resource_name=resource_name, + connection_qualified_name=connection_qualified_name, + ) + + +@pytest.mark.parametrize( + "resource_name", + [ + "data-owner", + "projects/p/locations/l/entryTypes/data-owner", + "projects/p/locations/ /aspectTypes/data-owner", + "projects/p/locations/l/aspectTypes/data-owner/extra", + ], +) +def test_creator_with_malformed_resource_name_raises_value_error(resource_name: str): + """Test creator rejects a resource name that is not an aspect type resource.""" + with pytest.raises(ValueError, match="Invalid resource_name"): + GCPDataplexAspectType.creator( + resource_name=resource_name, + connection_qualified_name=CONNECTION_QUALIFIED_NAME, + ) + + +def test_creator(): + """Test creator sets the resource attributes and denormalised connection attributes.""" + sut = GCPDataplexAspectType.creator( + resource_name=RESOURCE_NAME, + connection_qualified_name=CONNECTION_QUALIFIED_NAME, + ) + + assert sut.type_name == "GCPDataplexAspectType" + assert sut.name == ASPECT_TYPE_NAME + assert sut.qualified_name == ASPECT_TYPE_QUALIFIED_NAME + assert sut.gcp_dataplex_aspect_type_resource_name == RESOURCE_NAME + assert sut.gcp_dataplex_aspect_type_project == "my-project" + assert sut.gcp_dataplex_aspect_type_location == "us-central1" + assert sut.connection_qualified_name == CONNECTION_QUALIFIED_NAME + assert sut.connector_name == CONNECTOR_NAME + assert sut.guid + sut.validate(for_creation=True) + + +def test_create_alias(): + """Test the deprecated create alias delegates to creator.""" + sut = GCPDataplexAspectType.create( + resource_name=RESOURCE_NAME, + connection_qualified_name=CONNECTION_QUALIFIED_NAME, + ) + + assert sut.qualified_name == ASPECT_TYPE_QUALIFIED_NAME + + +@pytest.mark.parametrize( + "qualified_name, name, message", + [ + (None, ASPECT_TYPE_NAME, "qualified_name is required"), + (ASPECT_TYPE_QUALIFIED_NAME, None, "name is required"), + ], +) +def test_updater_with_invalid_parameter_raises_value_error( + qualified_name: str, name: str, message: str +): + """Test updater validates required parameters.""" + with pytest.raises(ValueError, match=message): + GCPDataplexAspectType.updater(qualified_name=qualified_name, name=name) + + +def test_updater(): + """Test updater creates a GCPDataplexAspectType for modification.""" + sut = GCPDataplexAspectType.updater( + qualified_name=ASPECT_TYPE_QUALIFIED_NAME, name=ASPECT_TYPE_NAME + ) + + assert sut.qualified_name == ASPECT_TYPE_QUALIFIED_NAME + assert sut.name == ASPECT_TYPE_NAME + + +def test_trim_to_required(): + """Test trim_to_required keeps only updater-required fields.""" + sut = GCPDataplexAspectType.creator( + resource_name=RESOURCE_NAME, + connection_qualified_name=CONNECTION_QUALIFIED_NAME, + ).trim_to_required() + + assert sut.qualified_name == ASPECT_TYPE_QUALIFIED_NAME + assert sut.name == ASPECT_TYPE_NAME + assert sut.gcp_dataplex_aspect_type_resource_name is UNSET