diff --git a/pyatlan_v9/model/assets/_overlays/flow_control_operation.py b/pyatlan_v9/model/assets/_overlays/flow_control_operation.py new file mode 100644 index 000000000..0d95b69e5 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/flow_control_operation.py @@ -0,0 +1,77 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + source_id: str, + scope: str | None = None, + parent_qualified_name: str | None = None, + flow_id: str | None = None, + ) -> "FlowControlOperation": + """ + Create a new FlowControlOperation asset. + + Exactly one of ``scope`` or ``parent_qualified_name`` must be given: + + - top-level operation: ``{connection_qualified_name}/{scope}/{source_id}`` + (e.g. scope = Fabric workspace ID, or SageMaker region) + - child operation: ``{parent_qualified_name}/jobs/{source_id}``, linked to + its parent through ``flow_controlled_by`` + + Args: + name: Name of the operation + connection_qualified_name: Unique name of the connection + source_id: Identifier of the operation in the source system; the last + segment of the qualified name + scope: Source-system container of a top-level operation + parent_qualified_name: Unique name of the controlling operation + flow_id: Value for ``flow_id`` (e.g. an ARN); not part of the + qualified name + + Returns: + FlowControlOperation instance ready to be created + + Raises: + ValueError: If required parameters are missing or invalid + """ + validate_required_fields( + ["name", "connection_qualified_name", "source_id"], + [name, connection_qualified_name, source_id], + ) + fields = connection_qualified_name.split("/") + if len(fields) != 3: + raise ValueError("Invalid connection_qualified_name") + if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "": + raise ValueError("Invalid connection_qualified_name") + if (scope is None) == (parent_qualified_name is None): + raise ValueError( + "Exactly one of scope or parent_qualified_name must be provided" + ) + + flow_controlled_by: RelatedFlowControlOperation | UnsetType = UNSET + if parent_qualified_name is not None: + validate_required_fields(["parent_qualified_name"], [parent_qualified_name]) + if not parent_qualified_name.startswith(f"{connection_qualified_name}/"): + raise ValueError( + "parent_qualified_name must belong to connection_qualified_name" + ) + qualified_name = f"{parent_qualified_name}/jobs/{source_id}" + flow_controlled_by = RelatedFlowControlOperation( + qualified_name=parent_qualified_name + ) + else: + validate_required_fields(["scope"], [scope]) + qualified_name = f"{connection_qualified_name}/{scope}/{source_id}" + + return cls( + name=name, + qualified_name=qualified_name, + connection_qualified_name=connection_qualified_name, + connector_name=fields[1], + flow_id=flow_id if flow_id is not None else UNSET, + flow_controlled_by=flow_controlled_by, + ) diff --git a/pyatlan_v9/model/assets/_overlays/s3_prefix.py b/pyatlan_v9/model/assets/_overlays/s3_prefix.py new file mode 100644 index 000000000..04b83c429 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/s3_prefix.py @@ -0,0 +1,89 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + connection_qualified_name: str, + s3_bucket_name: str, + s3_bucket_qualified_name: str, + prefix: str, + name: str | None = None, + ) -> "S3Prefix": + """ + Create a new S3Prefix asset. + + The qualified name is ``{s3_bucket_qualified_name}/{prefix}/`` — the full + key from the bucket root, always with a trailing slash (native S3 + convention, shared by every app that builds prefixes). + + The parent prefix is recorded by qualified name + (``s3_parent_prefix_qualified_name`` and ``s3_prefix_hierarchy``); the + ``s3_parent_prefix`` relationship is left for the caller to set once the + parent prefix exists. + + Args: + connection_qualified_name: Unique name of the connection + s3_bucket_name: Simple name of the bucket + s3_bucket_qualified_name: Unique name of the bucket + prefix: Full key of the prefix from the bucket root, e.g. ``a/b/`` + (leading and trailing slashes are optional) + name: Name of the prefix; defaults to the last segment of ``prefix`` + + Returns: + S3Prefix instance ready to be created + + Raises: + ValueError: If required parameters are missing or invalid + """ + validate_required_fields( + [ + "connection_qualified_name", + "s3_bucket_name", + "s3_bucket_qualified_name", + "prefix", + ], + [ + connection_qualified_name, + s3_bucket_name, + s3_bucket_qualified_name, + prefix, + ], + ) + fields = connection_qualified_name.split("/") + if len(fields) != 3: + raise ValueError("Invalid connection_qualified_name") + if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "": + raise ValueError("Invalid connection_qualified_name") + if fields[1].lower() != "s3": + raise ValueError("Invalid connection_qualified_name") + + segments = prefix.strip("/").split("/") + if any(segment.strip() == "" for segment in segments): + raise ValueError("prefix cannot contain empty segments") + if name is None: + name = segments[-1] + validate_required_fields(["name"], [name]) + + def _prefix_qn(depth: int) -> str: + return f"{s3_bucket_qualified_name}/{'/'.join(segments[:depth])}/" + + # Ancestors from immediate parent up to the root prefix. + hierarchy = [ + {"name": segments[depth - 1], "qualifiedName": _prefix_qn(depth)} + for depth in range(len(segments) - 1, 0, -1) + ] + return cls( + name=name, + qualified_name=_prefix_qn(len(segments)), + connection_qualified_name=connection_qualified_name, + connector_name=fields[1], + s3_bucket_name=s3_bucket_name, + s3_bucket_qualified_name=s3_bucket_qualified_name, + s3_bucket=RelatedS3Bucket(qualified_name=s3_bucket_qualified_name), + s3_parent_prefix_qualified_name=( + hierarchy[0]["qualifiedName"] if hierarchy else UNSET + ), + s3_prefix_hierarchy=hierarchy or UNSET, + ) diff --git a/pyatlan_v9/model/assets/_overlays/tag_attachment.py b/pyatlan_v9/model/assets/_overlays/tag_attachment.py new file mode 100644 index 000000000..f65434d22 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/tag_attachment.py @@ -0,0 +1,98 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + _QN_STYLES: ClassVar[tuple[str, ...]] = ( + "tag_object", + "tag_attachments_object", + "object_tag", + ) + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + tag_qualified_name: str, + object_qualified_name: str, + qn_style: str = "tag_object", + object_type_name: str | None = None, + source_tag_type_name: str | None = None, + value: str | None = None, + ) -> "TagAttachment": + """ + Create a new TagAttachment asset, linking a source tag to an asset. + + Apps already in production build the qualified name in different shapes, + so ``qn_style`` selects one; changing it changes the attachment's identity: + + - ``tag_object`` (default): ``{tag_qualified_name}/{object_qualified_name}`` + - ``tag_attachments_object``: + ``{tag_qualified_name}/attachments/{object_qualified_name}`` + - ``object_tag``: ``{object_qualified_name}/{name}`` + + Args: + name: Name of the source tag + connection_qualified_name: Unique name of the connection + tag_qualified_name: Unique name of the source tag + object_qualified_name: Unique name of the asset the tag is attached to + qn_style: Shape of the qualified name (see above) + object_type_name: Type name of the asset the tag is attached to + source_tag_type_name: Type name of the source tag + value: String value of the tag on this asset, if any + + Returns: + TagAttachment instance ready to be created + + Raises: + ValueError: If required parameters are missing or invalid + """ + validate_required_fields( + [ + "name", + "connection_qualified_name", + "tag_qualified_name", + "object_qualified_name", + ], + [ + name, + connection_qualified_name, + tag_qualified_name, + object_qualified_name, + ], + ) + fields = connection_qualified_name.split("/") + if len(fields) != 3: + raise ValueError("Invalid connection_qualified_name") + if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "": + raise ValueError("Invalid connection_qualified_name") + + if qn_style == "tag_object": + qualified_name = f"{tag_qualified_name}/{object_qualified_name}" + elif qn_style == "tag_attachments_object": + qualified_name = f"{tag_qualified_name}/attachments/{object_qualified_name}" + elif qn_style == "object_tag": + qualified_name = f"{object_qualified_name}/{name}" + else: + raise ValueError(f"qn_style must be one of {cls._QN_STYLES}") + + return cls( + name=name, + qualified_name=qualified_name, + connection_qualified_name=connection_qualified_name, + connector_name=fields[1], + tag_qualified_name=tag_qualified_name, + source_tag_qualified_name=tag_qualified_name, + source_tag_display_name=name, + source_tag_type_name=( + source_tag_type_name if source_tag_type_name is not None else UNSET + ), + object_qualified_name=object_qualified_name, + object_type_name=object_type_name + if object_type_name is not None + else UNSET, + value_type="NONE" if value is None else "STRING", + value=value, + tag_attachment_string_value=value if value is not None else UNSET, + propagate=False, + ) diff --git a/pyatlan_v9/model/assets/flow_control_operation.py b/pyatlan_v9/model/assets/flow_control_operation.py index e4a382596..0e2aa062a 100644 --- a/pyatlan_v9/model/assets/flow_control_operation.py +++ b/pyatlan_v9/model/assets/flow_control_operation.py @@ -25,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 @@ -315,6 +316,82 @@ def relate(self) -> "RelatedFlowControlOperation": return RelatedFlowControlOperation(guid=self.guid) return RelatedFlowControlOperation(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + source_id: str, + scope: str | None = None, + parent_qualified_name: str | None = None, + flow_id: str | None = None, + ) -> "FlowControlOperation": + """ + Create a new FlowControlOperation asset. + + Exactly one of ``scope`` or ``parent_qualified_name`` must be given: + + - top-level operation: ``{connection_qualified_name}/{scope}/{source_id}`` + (e.g. scope = Fabric workspace ID, or SageMaker region) + - child operation: ``{parent_qualified_name}/jobs/{source_id}``, linked to + its parent through ``flow_controlled_by`` + + Args: + name: Name of the operation + connection_qualified_name: Unique name of the connection + source_id: Identifier of the operation in the source system; the last + segment of the qualified name + scope: Source-system container of a top-level operation + parent_qualified_name: Unique name of the controlling operation + flow_id: Value for ``flow_id`` (e.g. an ARN); not part of the + qualified name + + Returns: + FlowControlOperation instance ready to be created + + Raises: + ValueError: If required parameters are missing or invalid + """ + validate_required_fields( + ["name", "connection_qualified_name", "source_id"], + [name, connection_qualified_name, source_id], + ) + fields = connection_qualified_name.split("/") + if len(fields) != 3: + raise ValueError("Invalid connection_qualified_name") + if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "": + raise ValueError("Invalid connection_qualified_name") + if (scope is None) == (parent_qualified_name is None): + raise ValueError( + "Exactly one of scope or parent_qualified_name must be provided" + ) + + flow_controlled_by: RelatedFlowControlOperation | UnsetType = UNSET + if parent_qualified_name is not None: + validate_required_fields(["parent_qualified_name"], [parent_qualified_name]) + if not parent_qualified_name.startswith(f"{connection_qualified_name}/"): + raise ValueError( + "parent_qualified_name must belong to connection_qualified_name" + ) + qualified_name = f"{parent_qualified_name}/jobs/{source_id}" + flow_controlled_by = RelatedFlowControlOperation( + qualified_name=parent_qualified_name + ) + else: + validate_required_fields(["scope"], [scope]) + qualified_name = f"{connection_qualified_name}/{scope}/{source_id}" + + return cls( + name=name, + qualified_name=qualified_name, + connection_qualified_name=connection_qualified_name, + connector_name=fields[1], + flow_id=flow_id if flow_id is not None else UNSET, + flow_controlled_by=flow_controlled_by, + ) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/s3_prefix.py b/pyatlan_v9/model/assets/s3_prefix.py index 81f4e26ba..73e66f48b 100644 --- a/pyatlan_v9/model/assets/s3_prefix.py +++ b/pyatlan_v9/model/assets/s3_prefix.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 @@ -382,6 +383,94 @@ def relate(self) -> "RelatedS3Prefix": return RelatedS3Prefix(guid=self.guid) return RelatedS3Prefix(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + connection_qualified_name: str, + s3_bucket_name: str, + s3_bucket_qualified_name: str, + prefix: str, + name: str | None = None, + ) -> "S3Prefix": + """ + Create a new S3Prefix asset. + + The qualified name is ``{s3_bucket_qualified_name}/{prefix}/`` — the full + key from the bucket root, always with a trailing slash (native S3 + convention, shared by every app that builds prefixes). + + The parent prefix is recorded by qualified name + (``s3_parent_prefix_qualified_name`` and ``s3_prefix_hierarchy``); the + ``s3_parent_prefix`` relationship is left for the caller to set once the + parent prefix exists. + + Args: + connection_qualified_name: Unique name of the connection + s3_bucket_name: Simple name of the bucket + s3_bucket_qualified_name: Unique name of the bucket + prefix: Full key of the prefix from the bucket root, e.g. ``a/b/`` + (leading and trailing slashes are optional) + name: Name of the prefix; defaults to the last segment of ``prefix`` + + Returns: + S3Prefix instance ready to be created + + Raises: + ValueError: If required parameters are missing or invalid + """ + validate_required_fields( + [ + "connection_qualified_name", + "s3_bucket_name", + "s3_bucket_qualified_name", + "prefix", + ], + [ + connection_qualified_name, + s3_bucket_name, + s3_bucket_qualified_name, + prefix, + ], + ) + fields = connection_qualified_name.split("/") + if len(fields) != 3: + raise ValueError("Invalid connection_qualified_name") + if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "": + raise ValueError("Invalid connection_qualified_name") + if fields[1].lower() != "s3": + raise ValueError("Invalid connection_qualified_name") + + segments = prefix.strip("/").split("/") + if any(segment.strip() == "" for segment in segments): + raise ValueError("prefix cannot contain empty segments") + if name is None: + name = segments[-1] + validate_required_fields(["name"], [name]) + + def _prefix_qn(depth: int) -> str: + return f"{s3_bucket_qualified_name}/{'/'.join(segments[:depth])}/" + + # Ancestors from immediate parent up to the root prefix. + hierarchy = [ + {"name": segments[depth - 1], "qualifiedName": _prefix_qn(depth)} + for depth in range(len(segments) - 1, 0, -1) + ] + return cls( + name=name, + qualified_name=_prefix_qn(len(segments)), + connection_qualified_name=connection_qualified_name, + connector_name=fields[1], + s3_bucket_name=s3_bucket_name, + s3_bucket_qualified_name=s3_bucket_qualified_name, + s3_bucket=RelatedS3Bucket(qualified_name=s3_bucket_qualified_name), + s3_parent_prefix_qualified_name=( + hierarchy[0]["qualifiedName"] if hierarchy else UNSET + ), + s3_prefix_hierarchy=hierarchy or UNSET, + ) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/tag_attachment.py b/pyatlan_v9/model/assets/tag_attachment.py index 4fe9161cd..2246e8589 100644 --- a/pyatlan_v9/model/assets/tag_attachment.py +++ b/pyatlan_v9/model/assets/tag_attachment.py @@ -24,6 +24,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 @@ -265,6 +266,103 @@ def relate(self) -> "RelatedTagAttachment": return RelatedTagAttachment(guid=self.guid) return RelatedTagAttachment(qualified_name=self.qualified_name) + _QN_STYLES: ClassVar[tuple[str, ...]] = ( + "tag_object", + "tag_attachments_object", + "object_tag", + ) + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + tag_qualified_name: str, + object_qualified_name: str, + qn_style: str = "tag_object", + object_type_name: str | None = None, + source_tag_type_name: str | None = None, + value: str | None = None, + ) -> "TagAttachment": + """ + Create a new TagAttachment asset, linking a source tag to an asset. + + Apps already in production build the qualified name in different shapes, + so ``qn_style`` selects one; changing it changes the attachment's identity: + + - ``tag_object`` (default): ``{tag_qualified_name}/{object_qualified_name}`` + - ``tag_attachments_object``: + ``{tag_qualified_name}/attachments/{object_qualified_name}`` + - ``object_tag``: ``{object_qualified_name}/{name}`` + + Args: + name: Name of the source tag + connection_qualified_name: Unique name of the connection + tag_qualified_name: Unique name of the source tag + object_qualified_name: Unique name of the asset the tag is attached to + qn_style: Shape of the qualified name (see above) + object_type_name: Type name of the asset the tag is attached to + source_tag_type_name: Type name of the source tag + value: String value of the tag on this asset, if any + + Returns: + TagAttachment instance ready to be created + + Raises: + ValueError: If required parameters are missing or invalid + """ + validate_required_fields( + [ + "name", + "connection_qualified_name", + "tag_qualified_name", + "object_qualified_name", + ], + [ + name, + connection_qualified_name, + tag_qualified_name, + object_qualified_name, + ], + ) + fields = connection_qualified_name.split("/") + if len(fields) != 3: + raise ValueError("Invalid connection_qualified_name") + if fields[0].replace(" ", "") == "" or fields[2].replace(" ", "") == "": + raise ValueError("Invalid connection_qualified_name") + + if qn_style == "tag_object": + qualified_name = f"{tag_qualified_name}/{object_qualified_name}" + elif qn_style == "tag_attachments_object": + qualified_name = f"{tag_qualified_name}/attachments/{object_qualified_name}" + elif qn_style == "object_tag": + qualified_name = f"{object_qualified_name}/{name}" + else: + raise ValueError(f"qn_style must be one of {cls._QN_STYLES}") + + return cls( + name=name, + qualified_name=qualified_name, + connection_qualified_name=connection_qualified_name, + connector_name=fields[1], + tag_qualified_name=tag_qualified_name, + source_tag_qualified_name=tag_qualified_name, + source_tag_display_name=name, + source_tag_type_name=( + source_tag_type_name if source_tag_type_name is not None else UNSET + ), + object_qualified_name=object_qualified_name, + object_type_name=object_type_name + if object_type_name is not None + else UNSET, + value_type="NONE" if value is None else "STRING", + value=value, + tag_attachment_string_value=value if value is not None else UNSET, + propagate=False, + ) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/tests_v9/unit/model/flow_control_operation_test.py b/tests_v9/unit/model/flow_control_operation_test.py new file mode 100644 index 000000000..222928545 --- /dev/null +++ b/tests_v9/unit/model/flow_control_operation_test.py @@ -0,0 +1,109 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for FlowControlOperation.creator in pyatlan_v9.""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import FlowControlOperation + +FABRIC_CONNECTION_QN = "default/fabric/1700000000" +SAGEMAKER_CONNECTION_QN = "default/sagemaker/1700000000" +WORKSPACE_ID = "11111111-2222-3333-4444-555555555555" +NOTEBOOK_ID = "66666666-7777-8888-9999-000000000000" +NOTEBOOK_QN = f"{FABRIC_CONNECTION_QN}/{WORKSPACE_ID}/{NOTEBOOK_ID}" + + +@pytest.mark.parametrize( + "kwargs, message", + [ + ({"name": None}, "name is required"), + ({"connection_qualified_name": None}, "connection_qualified_name is required"), + ({"source_id": " "}, "source_id cannot be blank"), + ({"scope": None}, "Exactly one of scope or parent_qualified_name"), + ({"parent_qualified_name": NOTEBOOK_QN}, "Exactly one of scope"), + ({"scope": ""}, "scope cannot be blank"), + ({"connection_qualified_name": "default/fabric"}, "Invalid connection"), + ], +) +def test_creator_with_invalid_parameters_raises_value_error(kwargs, message): + params = { + "name": "nb", + "connection_qualified_name": FABRIC_CONNECTION_QN, + "source_id": NOTEBOOK_ID, + "scope": WORKSPACE_ID, + **kwargs, + } + with pytest.raises(ValueError, match=message): + FlowControlOperation.creator(**params) + + +def test_creator_child_requires_parent_in_same_connection(): + with pytest.raises(ValueError, match="must belong to connection_qualified_name"): + FlowControlOperation.creator( + name="job", + connection_qualified_name=FABRIC_CONNECTION_QN, + source_id="abc", + parent_qualified_name=f"{SAGEMAKER_CONNECTION_QN}/x/y", + ) + + +def test_parity_fabric_notebook(): + """atlan-fabric-app: {conn}/{workspace_id}/{notebook_id}.""" + sut = FlowControlOperation.creator( + name="My notebook", + connection_qualified_name=FABRIC_CONNECTION_QN, + scope=WORKSPACE_ID, + source_id=NOTEBOOK_ID, + flow_id=NOTEBOOK_ID, + ) + assert sut.qualified_name == NOTEBOOK_QN + assert sut.name == "My notebook" + assert sut.connection_qualified_name == FABRIC_CONNECTION_QN + assert sut.connector_name == "fabric" + assert sut.flow_id == NOTEBOOK_ID + assert sut.flow_controlled_by is UNSET + sut.validate(for_creation=True) + + +def test_parity_fabric_spark_job(): + """atlan-fabric-app: {notebook_qn}/jobs/{sha256(namespace\\0jobName)}.""" + key = "a" * 64 + sut = FlowControlOperation.creator( + name="job", + connection_qualified_name=FABRIC_CONNECTION_QN, + parent_qualified_name=NOTEBOOK_QN, + source_id=key, + flow_id=key, + ) + assert sut.qualified_name == f"{NOTEBOOK_QN}/jobs/{key}" + assert sut.flow_controlled_by.qualified_name == NOTEBOOK_QN + assert sut.flow_id == key + sut.validate(for_creation=True) + + +def test_parity_sagemaker_job(): + """atlan-sagemaker-app: {conn}/{region}/{job_name}; flowId is the job ARN.""" + arn = "arn:aws:sagemaker:us-east-1:000000000000:training-job/example-job" + sut = FlowControlOperation.creator( + name="example-job", + connection_qualified_name=SAGEMAKER_CONNECTION_QN, + scope="us-east-1", + source_id="example-job", + flow_id=arn, + ) + assert sut.qualified_name == f"{SAGEMAKER_CONNECTION_QN}/us-east-1/example-job" + assert sut.connector_name == "sagemaker" + assert sut.flow_id == arn + sut.validate(for_creation=True) + + +def test_flow_id_optional(): + sut = FlowControlOperation.creator( + name="j", + connection_qualified_name=SAGEMAKER_CONNECTION_QN, + scope="us-east-1", + source_id="j", + ) + assert sut.flow_id is UNSET diff --git a/tests_v9/unit/model/s3_prefix_test.py b/tests_v9/unit/model/s3_prefix_test.py new file mode 100644 index 000000000..6ced7ef8b --- /dev/null +++ b/tests_v9/unit/model/s3_prefix_test.py @@ -0,0 +1,117 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for S3Prefix.creator in pyatlan_v9.""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import S3Prefix + +CONNECTION_QN = "default/s3/1700000000" +BUCKET_NAME = "example-bucket" +BUCKET_QN = f"{CONNECTION_QN}/{BUCKET_NAME}" + + +def _create(prefix: str, **kwargs) -> S3Prefix: + return S3Prefix.creator( + connection_qualified_name=CONNECTION_QN, + s3_bucket_name=BUCKET_NAME, + s3_bucket_qualified_name=BUCKET_QN, + prefix=prefix, + **kwargs, + ) + + +@pytest.mark.parametrize( + "connection_qualified_name, s3_bucket_name, s3_bucket_qualified_name, prefix, message", + [ + (None, BUCKET_NAME, BUCKET_QN, "a/", "connection_qualified_name is required"), + (CONNECTION_QN, None, BUCKET_QN, "a/", "s3_bucket_name is required"), + ( + CONNECTION_QN, + BUCKET_NAME, + None, + "a/", + "s3_bucket_qualified_name is required", + ), + (CONNECTION_QN, BUCKET_NAME, BUCKET_QN, None, "prefix is required"), + (CONNECTION_QN, BUCKET_NAME, BUCKET_QN, " ", "prefix cannot be blank"), + (CONNECTION_QN, BUCKET_NAME, BUCKET_QN, "a//b/", "empty segments"), + (CONNECTION_QN, BUCKET_NAME, BUCKET_QN, "/", "empty segments"), + ("default/s3", BUCKET_NAME, BUCKET_QN, "a/", "Invalid connection"), + ("default/gcs/1", BUCKET_NAME, BUCKET_QN, "a/", "Invalid connection"), + ], +) +def test_creator_with_invalid_parameters_raises_value_error( + connection_qualified_name, s3_bucket_name, s3_bucket_qualified_name, prefix, message +): + with pytest.raises(ValueError, match=message): + S3Prefix.creator( + connection_qualified_name=connection_qualified_name, + s3_bucket_name=s3_bucket_name, + s3_bucket_qualified_name=s3_bucket_qualified_name, + prefix=prefix, + ) + + +def test_creator_root_prefix(): + sut = _create("raw/") + + assert sut.name == "raw" + assert sut.qualified_name == f"{BUCKET_QN}/raw/" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "s3" + assert sut.s3_bucket_name == BUCKET_NAME + assert sut.s3_bucket_qualified_name == BUCKET_QN + assert sut.s3_bucket.qualified_name == BUCKET_QN + assert sut.s3_parent_prefix_qualified_name is UNSET + assert sut.s3_prefix_hierarchy is UNSET + + +def test_creator_nested_prefix_sets_parent_and_hierarchy(): + sut = _create("raw/2025/data/") + + assert sut.name == "data" + assert sut.qualified_name == f"{BUCKET_QN}/raw/2025/data/" + assert sut.s3_parent_prefix_qualified_name == f"{BUCKET_QN}/raw/2025/" + assert sut.s3_prefix_hierarchy == [ + {"name": "2025", "qualifiedName": f"{BUCKET_QN}/raw/2025/"}, + {"name": "raw", "qualifiedName": f"{BUCKET_QN}/raw/"}, + ] + + +@pytest.mark.parametrize("prefix", ["a/b", "a/b/", "/a/b/"]) +def test_creator_normalises_slashes(prefix): + assert _create(prefix).qualified_name == f"{BUCKET_QN}/a/b/" + + +def test_creator_explicit_name(): + assert _create("a/b/", name="B folder").name == "B folder" + + +def test_parity_s3_app(): + """atlan-s3-app: {bucket_qn}/{prefix_key} where prefix_key keeps its trailing slash.""" + prefix_key = "a/b/" + sut = _create(prefix_key) + assert sut.qualified_name == f"{BUCKET_QN}/{prefix_key}" + assert sut.name == prefix_key.split("/")[-2] + assert sut.s3_parent_prefix_qualified_name == f"{BUCKET_QN}/a/" + + +def test_parity_bridge_app(): + """atlan-bridge-app: {conn}/{bucket}/{parent_folder_path}/{name}/.""" + parent_folder_path, name = "/raw/2025/", "data" + full_path = f"{parent_folder_path.strip('/')}/{name.strip('/')}".strip("/") + sut = _create(full_path) + assert sut.qualified_name == f"{CONNECTION_QN}/{BUCKET_NAME}/{full_path}/" + assert sut.name == name + + +@pytest.mark.xfail( + strict=True, + reason="Generated _QUALIFIED_NAME_PATTERN rejects the trailing slash; needs " + "qualifiedNameFormula on S3Prefix in atlanhq/models (FND-3846)", +) +def test_creator_output_passes_generated_validation(): + _create("raw/2025/").validate(for_creation=True) diff --git a/tests_v9/unit/model/tag_attachment_test.py b/tests_v9/unit/model/tag_attachment_test.py new file mode 100644 index 000000000..4c59179eb --- /dev/null +++ b/tests_v9/unit/model/tag_attachment_test.py @@ -0,0 +1,110 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for TagAttachment.creator in pyatlan_v9.""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import TagAttachment + +DATABRICKS_CONN = "default/databricks/1700000000" +DBT_CONN = "default/dbt/1700000000" +BRIDGE_CONN = "default/example-provider/1700000000" + + +def _create(**kwargs) -> TagAttachment: + params = { + "name": "pii", + "connection_qualified_name": DATABRICKS_CONN, + "tag_qualified_name": f"{DATABRICKS_CONN}/tag/pii", + "object_qualified_name": f"{DATABRICKS_CONN}/cat/sch/tbl", + **kwargs, + } + return TagAttachment.creator(**params) + + +@pytest.mark.parametrize( + "kwargs, message", + [ + ({"name": None}, "name is required"), + ({"name": " "}, "name cannot be blank"), + ({"connection_qualified_name": None}, "connection_qualified_name is required"), + ({"tag_qualified_name": ""}, "tag_qualified_name cannot be blank"), + ({"object_qualified_name": None}, "object_qualified_name is required"), + ({"connection_qualified_name": "default/dbt"}, "Invalid connection"), + ({"qn_style": "unknown"}, "qn_style must be one of"), + ], +) +def test_creator_with_invalid_parameters_raises_value_error(kwargs, message): + with pytest.raises(ValueError, match=message): + _create(**kwargs) + + +def test_parity_databricks(): + """atlan-databricks-app: {conn}/tag/{tag_name}/{object_qn} (default style).""" + tag_qn = f"{DATABRICKS_CONN}/tag/pii" + obj_qn = f"{DATABRICKS_CONN}/cat/sch/tbl/col" + sut = _create( + object_qualified_name=obj_qn, + object_type_name="Column", + source_tag_type_name="DatabricksUnityCatalogTag", + value="high", + ) + assert sut.qualified_name == f"{tag_qn}/{obj_qn}" + assert sut.name == "pii" + assert sut.connector_name == "databricks" + assert sut.connection_qualified_name == DATABRICKS_CONN + assert sut.tag_qualified_name == tag_qn + assert sut.source_tag_qualified_name == tag_qn + assert sut.object_qualified_name == obj_qn + assert sut.object_type_name == "Column" + assert sut.source_tag_type_name == "DatabricksUnityCatalogTag" + assert sut.value == "high" + assert sut.value_type == "STRING" + assert sut.tag_attachment_string_value == "high" + assert sut.propagate is False + sut.validate(for_creation=True) + + +def test_parity_dbt(): + """atlan-dbt-app: {dbt_tag_qn}/attachments/{object_qn}.""" + tag_qn = f"{DBT_CONN}/account/1/project/2/tag/finance" + obj_qn = f"{DBT_CONN}/account/1/project/2/env/3/model.pkg.orders" + sut = _create( + name="finance", + connection_qualified_name=DBT_CONN, + tag_qualified_name=tag_qn, + object_qualified_name=obj_qn, + qn_style="tag_attachments_object", + source_tag_type_name="DbtTag", + ) + assert sut.qualified_name == f"{tag_qn}/attachments/{obj_qn}" + assert sut.connector_name == "dbt" + assert sut.value is None + assert sut.value_type == "NONE" + assert sut.tag_attachment_string_value is UNSET + + +def test_parity_bridge(): + """atlan-bridge-app: {object_qn}/{tag_name}.""" + tag_qn = f"{BRIDGE_CONN}/source-tag/42" + obj_qn = f"{BRIDGE_CONN}/db/sch/tbl" + sut = _create( + name="Confidential", + connection_qualified_name=BRIDGE_CONN, + tag_qualified_name=tag_qn, + object_qualified_name=obj_qn, + qn_style="object_tag", + source_tag_type_name="SourceTag", + ) + assert sut.qualified_name == f"{obj_qn}/Confidential" + assert sut.source_tag_display_name == "Confidential" + assert sut.connector_name == "example-provider" + + +def test_round_trip_json(): + sut = _create(value="v") + restored = TagAttachment.from_json(sut.to_json()) + assert restored.qualified_name == sut.qualified_name + assert restored.tag_qualified_name == sut.tag_qualified_name