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 new file mode 100644 index 000000000..67fb03e89 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/bigquery_routine.py @@ -0,0 +1,75 @@ +# 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 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) + + @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..c8845ef90 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/gcp_dataplex_aspect_type.py @@ -0,0 +1,65 @@ +# 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 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) + + @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..ad594970c 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,80 @@ 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 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) + + @classmethod + def create_for_modification(cls, **kwargs) -> "BigqueryRoutine": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= @@ -1210,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, @@ -1258,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 36b35625d..159c7e3aa 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,70 @@ 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 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) + + @classmethod + def create_for_modification(cls, **kwargs) -> "GCPDataplexAspectType": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= @@ -607,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, @@ -657,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/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