From 6544f642ffe9cd0dc52b11b6df6c4963391e7808 Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Wed, 14 Oct 2026 16:44:15 +0100 Subject: [PATCH 1/2] feat(v9): add creators for the 7 Sigma asset types (FND-3856) Overlays for SigmaWorkbook, SigmaPage, SigmaDataElement, SigmaDataElementField, SigmaDataset, SigmaDataModel and SigmaDataModelColumn. Each creator emits the qualifiedName atlan-sigma-app emits today, sets the parent relationship and the denormalised parent attributes, and raises on blank inputs. Classes regenerated from atlanhq/models#2082. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0166vmu2bYFX2P4LFGH3GR4F --- .../assets/_overlays/sigma_data_element.py | 48 ++++ .../_overlays/sigma_data_element_field.py | 62 +++++ .../assets/_overlays/sigma_data_model.py | 32 +++ .../_overlays/sigma_data_model_column.py | 49 ++++ .../model/assets/_overlays/sigma_dataset.py | 31 +++ .../model/assets/_overlays/sigma_page.py | 39 +++ .../model/assets/_overlays/sigma_workbook.py | 31 +++ pyatlan_v9/model/assets/sigma_data_element.py | 48 ++++ .../model/assets/sigma_data_element_field.py | 62 +++++ pyatlan_v9/model/assets/sigma_data_model.py | 40 +++ .../model/assets/sigma_data_model_column.py | 53 +++- pyatlan_v9/model/assets/sigma_dataset.py | 39 +++ pyatlan_v9/model/assets/sigma_page.py | 39 +++ pyatlan_v9/model/assets/sigma_workbook.py | 39 +++ tests_v9/integration/sigma_asset_test.py | 240 ++++++++++++++++++ tests_v9/unit/model/sigma_test.py | 230 +++++++++++++++++ 16 files changed, 1081 insertions(+), 1 deletion(-) create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_element.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_model.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_dataset.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_page.py create mode 100644 pyatlan_v9/model/assets/_overlays/sigma_workbook.py create mode 100644 tests_v9/integration/sigma_asset_test.py create mode 100644 tests_v9/unit/model/sigma_test.py diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_element.py b/pyatlan_v9/model/assets/_overlays/sigma_data_element.py new file mode 100644 index 000000000..30537a3a7 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_element.py @@ -0,0 +1,48 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + page_qualified_name: str, + page_name: str, + workbook_name: str, + data_element_id: str, + ) -> "SigmaDataElement": + validate_required_fields( + [ + "name", + "page_qualified_name", + "page_name", + "workbook_name", + "data_element_id", + ], + [name, page_qualified_name, page_name, workbook_name, data_element_id], + ) + fields = page_qualified_name.split("/") + if len(fields) != 5: + raise ValueError( + f"Invalid page_qualified_name: {page_qualified_name}. " + "Expected format: default/sigma///" + ) + return cls( + name=name, + qualified_name=f"{page_qualified_name}/{data_element_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name=page_qualified_name, + sigma_page_name=page_name, + sigma_page=RelatedSigmaPage(qualified_name=page_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElement": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElement": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py b/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py new file mode 100644 index 000000000..d2fa9af2a --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py @@ -0,0 +1,62 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_element_qualified_name: str, + data_element_name: str, + page_name: str, + workbook_name: str, + field_id: str, + ) -> "SigmaDataElementField": + validate_required_fields( + [ + "name", + "data_element_qualified_name", + "data_element_name", + "page_name", + "workbook_name", + "field_id", + ], + [ + name, + data_element_qualified_name, + data_element_name, + page_name, + workbook_name, + field_id, + ], + ) + fields = data_element_qualified_name.split("/") + if len(fields) != 6: + raise ValueError( + f"Invalid data_element_qualified_name: {data_element_qualified_name}. " + "Expected format: " + "default/sigma////" + ) + return cls( + name=name, + qualified_name=f"{data_element_qualified_name}/{field_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name="/".join(fields[:5]), + sigma_page_name=page_name, + sigma_data_element_qualified_name=data_element_qualified_name, + sigma_data_element_name=data_element_name, + sigma_data_element=RelatedSigmaDataElement( + qualified_name=data_element_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElementField": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElementField": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_model.py b/pyatlan_v9/model/assets/_overlays/sigma_data_model.py new file mode 100644 index 000000000..556e27102 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_model.py @@ -0,0 +1,32 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + data_model_id: str, + ) -> "SigmaDataModel": + validate_required_fields( + ["name", "connection_qualified_name", "data_model_id"], + [name, connection_qualified_name, data_model_id], + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=name, + qualified_name=f"{connection_qualified_name}/dataModels/{data_model_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + asset_source_id=data_model_id, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModel": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModel": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py b/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py new file mode 100644 index 000000000..c77aa6782 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py @@ -0,0 +1,49 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_model_qualified_name: str, + data_model_name: str, + element_id: str, + column_id: str, + ) -> "SigmaDataModelColumn": + validate_required_fields( + [ + "name", + "data_model_qualified_name", + "data_model_name", + "element_id", + "column_id", + ], + [name, data_model_qualified_name, data_model_name, element_id, column_id], + ) + fields = data_model_qualified_name.split("/") + if len(fields) != 5 or fields[3] != "dataModels": + raise ValueError( + f"Invalid data_model_qualified_name: {data_model_qualified_name}. " + "Expected format: default/sigma//dataModels/" + ) + return cls( + name=name, + qualified_name=f"{data_model_qualified_name}/{element_id}/{column_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + asset_source_id=column_id, + sigma_data_model_qualified_name=data_model_qualified_name, + sigma_data_model_name=data_model_name, + sigma_data_model=RelatedSigmaDataModel( + qualified_name=data_model_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_dataset.py b/pyatlan_v9/model/assets/_overlays/sigma_dataset.py new file mode 100644 index 000000000..7f24f6969 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_dataset.py @@ -0,0 +1,31 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + dataset_id: str, + ) -> "SigmaDataset": + validate_required_fields( + ["name", "connection_qualified_name", "dataset_id"], + [name, connection_qualified_name, dataset_id], + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=name, + qualified_name=f"{connection_qualified_name}/{dataset_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataset": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_page.py b/pyatlan_v9/model/assets/_overlays/sigma_page.py new file mode 100644 index 000000000..3cbfef750 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_page.py @@ -0,0 +1,39 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + workbook_qualified_name: str, + workbook_name: str, + page_id: str, + ) -> "SigmaPage": + validate_required_fields( + ["name", "workbook_qualified_name", "workbook_name", "page_id"], + [name, workbook_qualified_name, workbook_name, page_id], + ) + fields = workbook_qualified_name.split("/") + if len(fields) != 4: + raise ValueError( + f"Invalid workbook_qualified_name: {workbook_qualified_name}. " + "Expected format: default/sigma//" + ) + return cls( + name=name, + qualified_name=f"{workbook_qualified_name}/{page_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name=workbook_qualified_name, + sigma_workbook_name=workbook_name, + sigma_workbook=RelatedSigmaWorkbook(qualified_name=workbook_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaPage": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaPage": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_workbook.py b/pyatlan_v9/model/assets/_overlays/sigma_workbook.py new file mode 100644 index 000000000..95cb9f9d4 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/sigma_workbook.py @@ -0,0 +1,31 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + workbook_id: str, + ) -> "SigmaWorkbook": + validate_required_fields( + ["name", "connection_qualified_name", "workbook_id"], + [name, connection_qualified_name, workbook_id], + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=name, + qualified_name=f"{connection_qualified_name}/{workbook_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaWorkbook": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaWorkbook": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/sigma_data_element.py b/pyatlan_v9/model/assets/sigma_data_element.py index 4027390d5..96571c530 100644 --- a/pyatlan_v9/model/assets/sigma_data_element.py +++ b/pyatlan_v9/model/assets/sigma_data_element.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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -349,6 +350,53 @@ def relate(self) -> "RelatedSigmaDataElement": return RelatedSigmaDataElement(guid=self.guid) return RelatedSigmaDataElement(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + page_qualified_name: str, + page_name: str, + workbook_name: str, + data_element_id: str, + ) -> "SigmaDataElement": + validate_required_fields( + [ + "name", + "page_qualified_name", + "page_name", + "workbook_name", + "data_element_id", + ], + [name, page_qualified_name, page_name, workbook_name, data_element_id], + ) + fields = page_qualified_name.split("/") + if len(fields) != 5: + raise ValueError( + f"Invalid page_qualified_name: {page_qualified_name}. " + "Expected format: default/sigma///" + ) + return cls( + name=name, + qualified_name=f"{page_qualified_name}/{data_element_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name=page_qualified_name, + sigma_page_name=page_name, + sigma_page=RelatedSigmaPage(qualified_name=page_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElement": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElement": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_element_field.py b/pyatlan_v9/model/assets/sigma_data_element_field.py index 57ee9beeb..01fc813d2 100644 --- a/pyatlan_v9/model/assets/sigma_data_element_field.py +++ b/pyatlan_v9/model/assets/sigma_data_element_field.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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -341,6 +342,67 @@ def relate(self) -> "RelatedSigmaDataElementField": return RelatedSigmaDataElementField(guid=self.guid) return RelatedSigmaDataElementField(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_element_qualified_name: str, + data_element_name: str, + page_name: str, + workbook_name: str, + field_id: str, + ) -> "SigmaDataElementField": + validate_required_fields( + [ + "name", + "data_element_qualified_name", + "data_element_name", + "page_name", + "workbook_name", + "field_id", + ], + [ + name, + data_element_qualified_name, + data_element_name, + page_name, + workbook_name, + field_id, + ], + ) + fields = data_element_qualified_name.split("/") + if len(fields) != 6: + raise ValueError( + f"Invalid data_element_qualified_name: {data_element_qualified_name}. " + "Expected format: " + "default/sigma////" + ) + return cls( + name=name, + qualified_name=f"{data_element_qualified_name}/{field_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name="/".join(fields[:4]), + sigma_workbook_name=workbook_name, + sigma_page_qualified_name="/".join(fields[:5]), + sigma_page_name=page_name, + sigma_data_element_qualified_name=data_element_qualified_name, + sigma_data_element_name=data_element_name, + sigma_data_element=RelatedSigmaDataElement( + qualified_name=data_element_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataElementField": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataElementField": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_model.py b/pyatlan_v9/model/assets/sigma_data_model.py index ac2a2cda0..18e43e1ae 100644 --- a/pyatlan_v9/model/assets/sigma_data_model.py +++ b/pyatlan_v9/model/assets/sigma_data_model.py @@ -14,6 +14,7 @@ from __future__ import annotations +import re from typing import Any, ClassVar, 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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -278,6 +280,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/dataModels/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this SigmaDataModel instance. @@ -302,6 +306,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"SigmaDataModel validation failed: {errors}") @@ -334,6 +343,37 @@ def relate(self) -> "RelatedSigmaDataModel": return RelatedSigmaDataModel(guid=self.guid) return RelatedSigmaDataModel(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + data_model_id: str, + ) -> "SigmaDataModel": + validate_required_fields( + ["name", "connection_qualified_name", "data_model_id"], + [name, connection_qualified_name, data_model_id], + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=name, + qualified_name=f"{connection_qualified_name}/dataModels/{data_model_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + asset_source_id=data_model_id, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModel": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModel": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_model_column.py b/pyatlan_v9/model/assets/sigma_data_model_column.py index 0917f1ff4..52f274f9a 100644 --- a/pyatlan_v9/model/assets/sigma_data_model_column.py +++ b/pyatlan_v9/model/assets/sigma_data_model_column.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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -265,7 +266,9 @@ 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"^.+/dataModels/[^/]+/[^/]+/[^/]+$" + ) def validate(self, for_creation: bool = False) -> None: """ @@ -339,6 +342,54 @@ def relate(self) -> "RelatedSigmaDataModelColumn": return RelatedSigmaDataModelColumn(guid=self.guid) return RelatedSigmaDataModelColumn(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + data_model_qualified_name: str, + data_model_name: str, + element_id: str, + column_id: str, + ) -> "SigmaDataModelColumn": + validate_required_fields( + [ + "name", + "data_model_qualified_name", + "data_model_name", + "element_id", + "column_id", + ], + [name, data_model_qualified_name, data_model_name, element_id, column_id], + ) + fields = data_model_qualified_name.split("/") + if len(fields) != 5 or fields[3] != "dataModels": + raise ValueError( + f"Invalid data_model_qualified_name: {data_model_qualified_name}. " + "Expected format: default/sigma//dataModels/" + ) + return cls( + name=name, + qualified_name=f"{data_model_qualified_name}/{element_id}/{column_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + asset_source_id=column_id, + sigma_data_model_qualified_name=data_model_qualified_name, + sigma_data_model_name=data_model_name, + sigma_data_model=RelatedSigmaDataModel( + qualified_name=data_model_qualified_name + ), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataModelColumn": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_dataset.py b/pyatlan_v9/model/assets/sigma_dataset.py index 867902120..8978a6cdc 100644 --- a/pyatlan_v9/model/assets/sigma_dataset.py +++ b/pyatlan_v9/model/assets/sigma_dataset.py @@ -14,6 +14,7 @@ from __future__ import annotations +import re from typing import Any, ClassVar, 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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -254,6 +256,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this SigmaDataset instance. @@ -278,6 +282,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"SigmaDataset validation failed: {errors}") @@ -310,6 +319,36 @@ def relate(self) -> "RelatedSigmaDataset": return RelatedSigmaDataset(guid=self.guid) return RelatedSigmaDataset(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + dataset_id: str, + ) -> "SigmaDataset": + validate_required_fields( + ["name", "connection_qualified_name", "dataset_id"], + [name, connection_qualified_name, dataset_id], + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=name, + qualified_name=f"{connection_qualified_name}/{dataset_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaDataset": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_page.py b/pyatlan_v9/model/assets/sigma_page.py index cb7ee07c4..d6d6f2dcb 100644 --- a/pyatlan_v9/model/assets/sigma_page.py +++ b/pyatlan_v9/model/assets/sigma_page.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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -333,6 +334,44 @@ def relate(self) -> "RelatedSigmaPage": return RelatedSigmaPage(guid=self.guid) return RelatedSigmaPage(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + workbook_qualified_name: str, + workbook_name: str, + page_id: str, + ) -> "SigmaPage": + validate_required_fields( + ["name", "workbook_qualified_name", "workbook_name", "page_id"], + [name, workbook_qualified_name, workbook_name, page_id], + ) + fields = workbook_qualified_name.split("/") + if len(fields) != 4: + raise ValueError( + f"Invalid workbook_qualified_name: {workbook_qualified_name}. " + "Expected format: default/sigma//" + ) + return cls( + name=name, + qualified_name=f"{workbook_qualified_name}/{page_id}", + connection_qualified_name="/".join(fields[:3]), + connector_name=fields[1], + sigma_workbook_qualified_name=workbook_qualified_name, + sigma_workbook_name=workbook_name, + sigma_workbook=RelatedSigmaWorkbook(qualified_name=workbook_qualified_name), + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaPage": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaPage": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_workbook.py b/pyatlan_v9/model/assets/sigma_workbook.py index a3cff7bc3..598ba7ff0 100644 --- a/pyatlan_v9/model/assets/sigma_workbook.py +++ b/pyatlan_v9/model/assets/sigma_workbook.py @@ -14,6 +14,7 @@ from __future__ import annotations +import re from typing import Any, ClassVar, 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 .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -252,6 +254,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this SigmaWorkbook instance. @@ -276,6 +280,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"SigmaWorkbook validation failed: {errors}") @@ -308,6 +317,36 @@ def relate(self) -> "RelatedSigmaWorkbook": return RelatedSigmaWorkbook(guid=self.guid) return RelatedSigmaWorkbook(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + workbook_id: str, + ) -> "SigmaWorkbook": + validate_required_fields( + ["name", "connection_qualified_name", "workbook_id"], + [name, connection_qualified_name, workbook_id], + ) + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + return cls( + name=name, + qualified_name=f"{connection_qualified_name}/{workbook_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + ) + + @classmethod + def create(cls, **kwargs) -> "SigmaWorkbook": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "SigmaWorkbook": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/tests_v9/integration/sigma_asset_test.py b/tests_v9/integration/sigma_asset_test.py new file mode 100644 index 000000000..7414de3ec --- /dev/null +++ b/tests_v9/integration/sigma_asset_test.py @@ -0,0 +1,240 @@ +# 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 ( + Connection, + SigmaDataElement, + SigmaDataElementField, + SigmaDataModel, + SigmaDataModelColumn, + SigmaDataset, + SigmaPage, + SigmaWorkbook, +) +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("SIGMA") + +WORKBOOK_ID = f"wb-{MODULE_NAME}" +PAGE_ID = f"pg-{MODULE_NAME}" +DATA_ELEMENT_ID = f"el-{MODULE_NAME}" +FIELD_ID = f"fld-{MODULE_NAME}" +DATASET_ID = f"ds-{MODULE_NAME}" +DATA_MODEL_ID = f"dm-{MODULE_NAME}" +ELEMENT_ID = f"dmel-{MODULE_NAME}" +COLUMN_ID = f"col-{MODULE_NAME}" + + +def _save(client: AtlanClient, to_create, asset_type): + return client.asset.save(to_create).assets_created(asset_type=asset_type)[0] + + +@pytest.fixture(scope="module") +def connection(client: AtlanClient) -> Generator[Connection, None, None]: + result = create_connection( + client=client, name=MODULE_NAME, connector_type=AtlanConnectorType.SIGMA + ) + yield result + delete_asset(client, guid=result.guid, asset_type=Connection) + + +@pytest.fixture(scope="module") +def workbook( + client: AtlanClient, connection: Connection +) -> Generator[SigmaWorkbook, None, None]: + to_create = SigmaWorkbook.creator( + name=f"workbook_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + workbook_id=WORKBOOK_ID, + ) + result = _save(client, to_create, SigmaWorkbook) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaWorkbook) + + +@pytest.fixture(scope="module") +def page( + client: AtlanClient, workbook: SigmaWorkbook +) -> Generator[SigmaPage, None, None]: + to_create = SigmaPage.creator( + name=f"page_{MODULE_NAME}", + workbook_qualified_name=workbook.qualified_name, + workbook_name=workbook.name, + page_id=PAGE_ID, + ) + result = _save(client, to_create, SigmaPage) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaPage) + + +@pytest.fixture(scope="module") +def data_element( + client: AtlanClient, workbook: SigmaWorkbook, page: SigmaPage +) -> Generator[SigmaDataElement, None, None]: + to_create = SigmaDataElement.creator( + name=f"data_element_{MODULE_NAME}", + page_qualified_name=page.qualified_name, + page_name=page.name, + workbook_name=workbook.name, + data_element_id=DATA_ELEMENT_ID, + ) + result = _save(client, to_create, SigmaDataElement) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataElement) + + +@pytest.fixture(scope="module") +def data_element_field( + client: AtlanClient, + workbook: SigmaWorkbook, + page: SigmaPage, + data_element: SigmaDataElement, +) -> Generator[SigmaDataElementField, None, None]: + to_create = SigmaDataElementField.creator( + name=f"field_{MODULE_NAME}", + data_element_qualified_name=data_element.qualified_name, + data_element_name=data_element.name, + page_name=page.name, + workbook_name=workbook.name, + field_id=FIELD_ID, + ) + result = _save(client, to_create, SigmaDataElementField) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataElementField) + + +@pytest.fixture(scope="module") +def dataset( + client: AtlanClient, connection: Connection +) -> Generator[SigmaDataset, None, None]: + to_create = SigmaDataset.creator( + name=f"dataset_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + dataset_id=DATASET_ID, + ) + result = _save(client, to_create, SigmaDataset) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataset) + + +@pytest.fixture(scope="module") +def data_model( + client: AtlanClient, connection: Connection +) -> Generator[SigmaDataModel, None, None]: + to_create = SigmaDataModel.creator( + name=f"data_model_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + data_model_id=DATA_MODEL_ID, + ) + result = _save(client, to_create, SigmaDataModel) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataModel) + + +@pytest.fixture(scope="module") +def data_model_column( + client: AtlanClient, data_model: SigmaDataModel +) -> Generator[SigmaDataModelColumn, None, None]: + to_create = SigmaDataModelColumn.creator( + name=f"column_{MODULE_NAME}", + data_model_qualified_name=data_model.qualified_name, + data_model_name=data_model.name, + element_id=ELEMENT_ID, + column_id=COLUMN_ID, + ) + result = _save(client, to_create, SigmaDataModelColumn) + yield result + delete_asset(client, guid=result.guid, asset_type=SigmaDataModelColumn) + + +def _assert_retrieved(client, asset, asset_type, connection, expected_suffix): + retrieved = client.asset.get_by_guid( + asset.guid, asset_type=asset_type, ignore_relationships=False + ) + assert retrieved.qualified_name == f"{connection.qualified_name}/{expected_suffix}" + assert retrieved.name == asset.name + assert retrieved.connector_name == AtlanConnectorType.SIGMA + assert retrieved.connection_qualified_name == connection.qualified_name + return retrieved + + +def test_workbook(client: AtlanClient, connection: Connection, workbook): + _assert_retrieved(client, workbook, SigmaWorkbook, connection, WORKBOOK_ID) + + +def test_page(client: AtlanClient, connection: Connection, workbook, page): + retrieved = _assert_retrieved( + client, page, SigmaPage, connection, f"{WORKBOOK_ID}/{PAGE_ID}" + ) + assert retrieved.sigma_workbook.guid == workbook.guid + assert retrieved.sigma_workbook_qualified_name == workbook.qualified_name + assert retrieved.sigma_workbook_name == workbook.name + + +def test_data_element( + client: AtlanClient, connection: Connection, workbook, page, data_element +): + retrieved = _assert_retrieved( + client, + data_element, + SigmaDataElement, + connection, + f"{WORKBOOK_ID}/{PAGE_ID}/{DATA_ELEMENT_ID}", + ) + assert retrieved.sigma_page.guid == page.guid + assert retrieved.sigma_page_qualified_name == page.qualified_name + assert retrieved.sigma_workbook_qualified_name == workbook.qualified_name + + +def test_data_element_field( + client: AtlanClient, + connection: Connection, + workbook, + page, + data_element, + data_element_field, +): + retrieved = _assert_retrieved( + client, + data_element_field, + SigmaDataElementField, + connection, + f"{WORKBOOK_ID}/{PAGE_ID}/{DATA_ELEMENT_ID}/{FIELD_ID}", + ) + assert retrieved.sigma_data_element.guid == data_element.guid + assert retrieved.sigma_data_element_qualified_name == data_element.qualified_name + assert retrieved.sigma_page_qualified_name == page.qualified_name + assert retrieved.sigma_workbook_qualified_name == workbook.qualified_name + + +def test_dataset(client: AtlanClient, connection: Connection, dataset): + _assert_retrieved(client, dataset, SigmaDataset, connection, DATASET_ID) + + +def test_data_model(client: AtlanClient, connection: Connection, data_model): + retrieved = _assert_retrieved( + client, data_model, SigmaDataModel, connection, f"dataModels/{DATA_MODEL_ID}" + ) + assert retrieved.asset_source_id == DATA_MODEL_ID + + +def test_data_model_column( + client: AtlanClient, connection: Connection, data_model, data_model_column +): + retrieved = _assert_retrieved( + client, + data_model_column, + SigmaDataModelColumn, + connection, + f"dataModels/{DATA_MODEL_ID}/{ELEMENT_ID}/{COLUMN_ID}", + ) + assert retrieved.sigma_data_model.guid == data_model.guid + assert retrieved.sigma_data_model_qualified_name == data_model.qualified_name + assert retrieved.asset_source_id == COLUMN_ID diff --git a/tests_v9/unit/model/sigma_test.py b/tests_v9/unit/model/sigma_test.py new file mode 100644 index 000000000..fd17e4dca --- /dev/null +++ b/tests_v9/unit/model/sigma_test.py @@ -0,0 +1,230 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for the Sigma creators in pyatlan_v9. + +The expected qualifiedNames are the exact strings atlan-sigma-app emits today +(FND-3838 census), so a creator change that would re-key existing Sigma assets fails here. +""" + +import pytest + +from pyatlan_v9.model import ( + SigmaDataElement, + SigmaDataElementField, + SigmaDataModel, + SigmaDataModelColumn, + SigmaDataset, + SigmaPage, + SigmaWorkbook, +) + +CONNECTION_QN = "default/sigma/1712345678" +WORKBOOK_ID = "3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" +PAGE_ID = "pg_Ab12Cd34" +DATA_ELEMENT_ID = "el-Xy9Zw8Vu" +FIELD_ID = "fld_Qr7St6" +DATASET_ID = "9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b" +DATA_MODEL_ID = "dm-1a2b3c4d" +ELEMENT_ID = "dmel_Mn5Op4" +COLUMN_ID = "col_Gh3Ij2" + +WORKBOOK_QN = f"{CONNECTION_QN}/{WORKBOOK_ID}" +PAGE_QN = f"{WORKBOOK_QN}/{PAGE_ID}" +DATA_ELEMENT_QN = f"{PAGE_QN}/{DATA_ELEMENT_ID}" +DATA_MODEL_QN = f"{CONNECTION_QN}/dataModels/{DATA_MODEL_ID}" + +NAME = "Revenue" +WORKBOOK_NAME = "Sales Overview" +PAGE_NAME = "Summary" +DATA_ELEMENT_NAME = "Revenue by Region" +DATA_MODEL_NAME = "Orders Model" + +# (type, creator kwargs, the literal qualifiedName atlan-sigma-app emits) +ALL_TYPES = [ + ( + SigmaWorkbook, + dict(connection_qualified_name=CONNECTION_QN, workbook_id=WORKBOOK_ID), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f", + ), + ( + SigmaDataset, + dict(connection_qualified_name=CONNECTION_QN, dataset_id=DATASET_ID), + "default/sigma/1712345678/9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b", + ), + ( + SigmaDataModel, + dict(connection_qualified_name=CONNECTION_QN, data_model_id=DATA_MODEL_ID), + "default/sigma/1712345678/dataModels/dm-1a2b3c4d", + ), + ( + SigmaPage, + dict( + workbook_qualified_name=WORKBOOK_QN, + workbook_name=WORKBOOK_NAME, + page_id=PAGE_ID, + ), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34", + ), + ( + SigmaDataElement, + dict( + page_qualified_name=PAGE_QN, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + data_element_id=DATA_ELEMENT_ID, + ), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + "/el-Xy9Zw8Vu", + ), + ( + SigmaDataElementField, + dict( + data_element_qualified_name=DATA_ELEMENT_QN, + data_element_name=DATA_ELEMENT_NAME, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + field_id=FIELD_ID, + ), + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + "/el-Xy9Zw8Vu/fld_Qr7St6", + ), + ( + SigmaDataModelColumn, + dict( + data_model_qualified_name=DATA_MODEL_QN, + data_model_name=DATA_MODEL_NAME, + element_id=ELEMENT_ID, + column_id=COLUMN_ID, + ), + "default/sigma/1712345678/dataModels/dm-1a2b3c4d/dmel_Mn5Op4/col_Gh3Ij2", + ), +] +IDS = [t[0].__name__ for t in ALL_TYPES] + + +@pytest.mark.parametrize("asset_type, kwargs, expected_qn", ALL_TYPES, ids=IDS) +def test_creator_matches_app_qualified_name(asset_type, kwargs, expected_qn): + sut = asset_type.creator(name=NAME, **kwargs) + + assert sut.qualified_name == expected_qn + assert sut.name == NAME + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "sigma" + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +@pytest.mark.parametrize("asset_type, kwargs, expected_qn", ALL_TYPES, ids=IDS) +def test_create_alias(asset_type, kwargs, expected_qn): + assert asset_type.create(name=NAME, **kwargs).qualified_name == expected_qn + + +@pytest.mark.parametrize("asset_type, kwargs, _", ALL_TYPES, ids=IDS) +def test_creator_rejects_missing_or_blank_inputs(asset_type, kwargs, _): + for param in ["name", *kwargs]: + args = {"name": NAME, **kwargs} + args[param] = None + with pytest.raises(ValueError, match=f"{param} is required"): + asset_type.creator(**args) + args[param] = " " + with pytest.raises(ValueError, match=f"{param} cannot be blank"): + asset_type.creator(**args) + + +def test_page_sets_workbook(): + sut = SigmaPage.creator( + name=NAME, + workbook_qualified_name=WORKBOOK_QN, + workbook_name=WORKBOOK_NAME, + page_id=PAGE_ID, + ) + + assert sut.sigma_workbook.qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_name == WORKBOOK_NAME + + +def test_data_element_sets_page_and_workbook(): + sut = SigmaDataElement.creator( + name=NAME, + page_qualified_name=PAGE_QN, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + data_element_id=DATA_ELEMENT_ID, + ) + + assert sut.sigma_page.qualified_name == PAGE_QN + assert sut.sigma_page_qualified_name == PAGE_QN + assert sut.sigma_page_name == PAGE_NAME + assert sut.sigma_workbook_qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_name == WORKBOOK_NAME + + +def test_data_element_field_sets_all_ancestors(): + sut = SigmaDataElementField.creator( + name=NAME, + data_element_qualified_name=DATA_ELEMENT_QN, + data_element_name=DATA_ELEMENT_NAME, + page_name=PAGE_NAME, + workbook_name=WORKBOOK_NAME, + field_id=FIELD_ID, + ) + + assert sut.sigma_data_element.qualified_name == DATA_ELEMENT_QN + assert sut.sigma_data_element_qualified_name == DATA_ELEMENT_QN + assert sut.sigma_data_element_name == DATA_ELEMENT_NAME + assert sut.sigma_page_qualified_name == PAGE_QN + assert sut.sigma_page_name == PAGE_NAME + assert sut.sigma_workbook_qualified_name == WORKBOOK_QN + assert sut.sigma_workbook_name == WORKBOOK_NAME + + +def test_data_model_sets_asset_source_id(): + sut = SigmaDataModel.creator( + name=NAME, connection_qualified_name=CONNECTION_QN, data_model_id=DATA_MODEL_ID + ) + + assert sut.asset_source_id == DATA_MODEL_ID + + +def test_data_model_column_sets_data_model(): + sut = SigmaDataModelColumn.creator( + name=NAME, + data_model_qualified_name=DATA_MODEL_QN, + data_model_name=DATA_MODEL_NAME, + element_id=ELEMENT_ID, + column_id=COLUMN_ID, + ) + + assert sut.sigma_data_model.qualified_name == DATA_MODEL_QN + assert sut.sigma_data_model_qualified_name == DATA_MODEL_QN + assert sut.sigma_data_model_name == DATA_MODEL_NAME + assert sut.asset_source_id == COLUMN_ID + + +@pytest.mark.parametrize( + "asset_type, parent_param, bad_parent_qn", + [ + (SigmaPage, "workbook_qualified_name", CONNECTION_QN), + (SigmaPage, "workbook_qualified_name", PAGE_QN), + (SigmaDataElement, "page_qualified_name", WORKBOOK_QN), + (SigmaDataElement, "page_qualified_name", DATA_ELEMENT_QN), + (SigmaDataElementField, "data_element_qualified_name", PAGE_QN), + (SigmaDataElementField, "data_element_qualified_name", f"{DATA_ELEMENT_QN}/x"), + (SigmaDataModelColumn, "data_model_qualified_name", CONNECTION_QN), + # a page qn has the data model's segment count but not the `dataModels` literal + (SigmaDataModelColumn, "data_model_qualified_name", PAGE_QN), + ], +) +def test_creator_rejects_malformed_parent_qualified_name( + asset_type, parent_param, bad_parent_qn +): + kwargs = next(k for t, k, _ in ALL_TYPES if t is asset_type) + with pytest.raises(ValueError, match=f"Invalid {parent_param}"): + asset_type.creator(name=NAME, **{**kwargs, parent_param: bad_parent_qn}) + + +def test_data_model_patterns_require_data_models_segment(): + assert not SigmaDataModel._QUALIFIED_NAME_PATTERN.match(PAGE_QN) + assert not SigmaDataModelColumn._QUALIFIED_NAME_PATTERN.match(DATA_ELEMENT_QN) From 486facbc1823fdb5b701720f36f8a294ec370ccb Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Mon, 19 Oct 2026 16:31:25 +0100 Subject: [PATCH 2/2] FND-3856: per-type unit tests, unique_attributes parent refs, updater/trim - Parent relationships use Related(unique_attributes=...), which serialises to exactly the {typeName, uniqueAttributes} ref atlan-sigma-app emits (qualified_name= also added a top-level qualifiedName key). - Typed updater() and trim_to_required() on every Sigma overlay. - Split sigma_test.py into one tests_v9/unit/model/_test.py per type, adding updater, trim and missing/blank cases for every param. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0166vmu2bYFX2P4LFGH3GR4F --- .../assets/_overlays/sigma_data_element.py | 12 +- .../_overlays/sigma_data_element_field.py | 10 +- .../assets/_overlays/sigma_data_model.py | 8 + .../_overlays/sigma_data_model_column.py | 10 +- .../model/assets/_overlays/sigma_dataset.py | 8 + .../model/assets/_overlays/sigma_page.py | 12 +- .../model/assets/_overlays/sigma_workbook.py | 8 + pyatlan_v9/model/assets/sigma_data_element.py | 14 +- .../model/assets/sigma_data_element_field.py | 12 +- pyatlan_v9/model/assets/sigma_data_model.py | 10 + .../model/assets/sigma_data_model_column.py | 12 +- pyatlan_v9/model/assets/sigma_dataset.py | 8 + pyatlan_v9/model/assets/sigma_page.py | 12 +- pyatlan_v9/model/assets/sigma_workbook.py | 8 + .../model/sigma_data_element_field_test.py | 134 ++++++++++ .../unit/model/sigma_data_element_test.py | 117 +++++++++ .../model/sigma_data_model_column_test.py | 110 +++++++++ tests_v9/unit/model/sigma_data_model_test.py | 85 +++++++ tests_v9/unit/model/sigma_dataset_test.py | 75 ++++++ tests_v9/unit/model/sigma_page_test.py | 108 ++++++++ tests_v9/unit/model/sigma_test.py | 230 ------------------ tests_v9/unit/model/sigma_workbook_test.py | 75 ++++++ 22 files changed, 840 insertions(+), 238 deletions(-) create mode 100644 tests_v9/unit/model/sigma_data_element_field_test.py create mode 100644 tests_v9/unit/model/sigma_data_element_test.py create mode 100644 tests_v9/unit/model/sigma_data_model_column_test.py create mode 100644 tests_v9/unit/model/sigma_data_model_test.py create mode 100644 tests_v9/unit/model/sigma_dataset_test.py create mode 100644 tests_v9/unit/model/sigma_page_test.py delete mode 100644 tests_v9/unit/model/sigma_test.py create mode 100644 tests_v9/unit/model/sigma_workbook_test.py diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_element.py b/pyatlan_v9/model/assets/_overlays/sigma_data_element.py index 30537a3a7..e3d9c5ee2 100644 --- a/pyatlan_v9/model/assets/_overlays/sigma_data_element.py +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_element.py @@ -36,13 +36,23 @@ def creator( sigma_workbook_name=workbook_name, sigma_page_qualified_name=page_qualified_name, sigma_page_name=page_name, - sigma_page=RelatedSigmaPage(qualified_name=page_qualified_name), + sigma_page=RelatedSigmaPage( + unique_attributes={"qualifiedName": page_qualified_name} + ), ) @classmethod def create(cls, **kwargs) -> "SigmaDataElement": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataElement": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataElement": return cls.updater(**kwargs) + + def trim_to_required(self) -> "SigmaDataElement": + return SigmaDataElement.updater(qualified_name=self.qualified_name, name=self.name) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py b/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py index d2fa9af2a..fca2cb030 100644 --- a/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_element_field.py @@ -49,7 +49,7 @@ def creator( sigma_data_element_qualified_name=data_element_qualified_name, sigma_data_element_name=data_element_name, sigma_data_element=RelatedSigmaDataElement( - qualified_name=data_element_qualified_name + unique_attributes={"qualifiedName": data_element_qualified_name} ), ) @@ -57,6 +57,14 @@ def creator( def create(cls, **kwargs) -> "SigmaDataElementField": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataElementField": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataElementField": return cls.updater(**kwargs) + + def trim_to_required(self) -> "SigmaDataElementField": + return SigmaDataElementField.updater(qualified_name=self.qualified_name, name=self.name) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_model.py b/pyatlan_v9/model/assets/_overlays/sigma_data_model.py index 556e27102..552f72d98 100644 --- a/pyatlan_v9/model/assets/_overlays/sigma_data_model.py +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_model.py @@ -27,6 +27,14 @@ def creator( def create(cls, **kwargs) -> "SigmaDataModel": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataModel": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataModel": return cls.updater(**kwargs) + + def trim_to_required(self) -> "SigmaDataModel": + return SigmaDataModel.updater(qualified_name=self.qualified_name, name=self.name) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py b/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py index c77aa6782..1a5810f56 100644 --- a/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py +++ b/pyatlan_v9/model/assets/_overlays/sigma_data_model_column.py @@ -36,7 +36,7 @@ def creator( sigma_data_model_qualified_name=data_model_qualified_name, sigma_data_model_name=data_model_name, sigma_data_model=RelatedSigmaDataModel( - qualified_name=data_model_qualified_name + unique_attributes={"qualifiedName": data_model_qualified_name} ), ) @@ -44,6 +44,14 @@ def creator( def create(cls, **kwargs) -> "SigmaDataModelColumn": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataModelColumn": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataModelColumn": return cls.updater(**kwargs) + + def trim_to_required(self) -> "SigmaDataModelColumn": + return SigmaDataModelColumn.updater(qualified_name=self.qualified_name, name=self.name) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_dataset.py b/pyatlan_v9/model/assets/_overlays/sigma_dataset.py index 7f24f6969..0891292e6 100644 --- a/pyatlan_v9/model/assets/_overlays/sigma_dataset.py +++ b/pyatlan_v9/model/assets/_overlays/sigma_dataset.py @@ -26,6 +26,14 @@ def creator( def create(cls, **kwargs) -> "SigmaDataset": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataset": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataset": return cls.updater(**kwargs) + + def trim_to_required(self) -> "SigmaDataset": + return SigmaDataset.updater(qualified_name=self.qualified_name, name=self.name) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_page.py b/pyatlan_v9/model/assets/_overlays/sigma_page.py index 3cbfef750..b95f4a9e3 100644 --- a/pyatlan_v9/model/assets/_overlays/sigma_page.py +++ b/pyatlan_v9/model/assets/_overlays/sigma_page.py @@ -27,13 +27,23 @@ def creator( connector_name=fields[1], sigma_workbook_qualified_name=workbook_qualified_name, sigma_workbook_name=workbook_name, - sigma_workbook=RelatedSigmaWorkbook(qualified_name=workbook_qualified_name), + sigma_workbook=RelatedSigmaWorkbook( + unique_attributes={"qualifiedName": workbook_qualified_name} + ), ) @classmethod def create(cls, **kwargs) -> "SigmaPage": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaPage": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaPage": return cls.updater(**kwargs) + + def trim_to_required(self) -> "SigmaPage": + return SigmaPage.updater(qualified_name=self.qualified_name, name=self.name) diff --git a/pyatlan_v9/model/assets/_overlays/sigma_workbook.py b/pyatlan_v9/model/assets/_overlays/sigma_workbook.py index 95cb9f9d4..b61af9129 100644 --- a/pyatlan_v9/model/assets/_overlays/sigma_workbook.py +++ b/pyatlan_v9/model/assets/_overlays/sigma_workbook.py @@ -26,6 +26,14 @@ def creator( def create(cls, **kwargs) -> "SigmaWorkbook": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaWorkbook": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaWorkbook": return cls.updater(**kwargs) + + def trim_to_required(self) -> "SigmaWorkbook": + return SigmaWorkbook.updater(qualified_name=self.qualified_name, name=self.name) diff --git a/pyatlan_v9/model/assets/sigma_data_element.py b/pyatlan_v9/model/assets/sigma_data_element.py index 96571c530..775892e31 100644 --- a/pyatlan_v9/model/assets/sigma_data_element.py +++ b/pyatlan_v9/model/assets/sigma_data_element.py @@ -386,17 +386,29 @@ def creator( sigma_workbook_name=workbook_name, sigma_page_qualified_name=page_qualified_name, sigma_page_name=page_name, - sigma_page=RelatedSigmaPage(qualified_name=page_qualified_name), + sigma_page=RelatedSigmaPage( + unique_attributes={"qualifiedName": page_qualified_name} + ), ) @classmethod def create(cls, **kwargs) -> "SigmaDataElement": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataElement": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataElement": return cls.updater(**kwargs) + def trim_to_required(self) -> "SigmaDataElement": + return SigmaDataElement.updater( + qualified_name=self.qualified_name, name=self.name + ) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_element_field.py b/pyatlan_v9/model/assets/sigma_data_element_field.py index 01fc813d2..3df02790a 100644 --- a/pyatlan_v9/model/assets/sigma_data_element_field.py +++ b/pyatlan_v9/model/assets/sigma_data_element_field.py @@ -391,7 +391,7 @@ def creator( sigma_data_element_qualified_name=data_element_qualified_name, sigma_data_element_name=data_element_name, sigma_data_element=RelatedSigmaDataElement( - qualified_name=data_element_qualified_name + unique_attributes={"qualifiedName": data_element_qualified_name} ), ) @@ -399,10 +399,20 @@ def creator( def create(cls, **kwargs) -> "SigmaDataElementField": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataElementField": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataElementField": return cls.updater(**kwargs) + def trim_to_required(self) -> "SigmaDataElementField": + return SigmaDataElementField.updater( + qualified_name=self.qualified_name, name=self.name + ) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_model.py b/pyatlan_v9/model/assets/sigma_data_model.py index 18e43e1ae..8902fac8a 100644 --- a/pyatlan_v9/model/assets/sigma_data_model.py +++ b/pyatlan_v9/model/assets/sigma_data_model.py @@ -370,10 +370,20 @@ def creator( def create(cls, **kwargs) -> "SigmaDataModel": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataModel": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataModel": return cls.updater(**kwargs) + def trim_to_required(self) -> "SigmaDataModel": + return SigmaDataModel.updater( + qualified_name=self.qualified_name, name=self.name + ) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_data_model_column.py b/pyatlan_v9/model/assets/sigma_data_model_column.py index 52f274f9a..57d128d5c 100644 --- a/pyatlan_v9/model/assets/sigma_data_model_column.py +++ b/pyatlan_v9/model/assets/sigma_data_model_column.py @@ -378,7 +378,7 @@ def creator( sigma_data_model_qualified_name=data_model_qualified_name, sigma_data_model_name=data_model_name, sigma_data_model=RelatedSigmaDataModel( - qualified_name=data_model_qualified_name + unique_attributes={"qualifiedName": data_model_qualified_name} ), ) @@ -386,10 +386,20 @@ def creator( def create(cls, **kwargs) -> "SigmaDataModelColumn": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataModelColumn": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataModelColumn": return cls.updater(**kwargs) + def trim_to_required(self) -> "SigmaDataModelColumn": + return SigmaDataModelColumn.updater( + qualified_name=self.qualified_name, name=self.name + ) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_dataset.py b/pyatlan_v9/model/assets/sigma_dataset.py index 8978a6cdc..6f90b4bc4 100644 --- a/pyatlan_v9/model/assets/sigma_dataset.py +++ b/pyatlan_v9/model/assets/sigma_dataset.py @@ -345,10 +345,18 @@ def creator( def create(cls, **kwargs) -> "SigmaDataset": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaDataset": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaDataset": return cls.updater(**kwargs) + def trim_to_required(self) -> "SigmaDataset": + return SigmaDataset.updater(qualified_name=self.qualified_name, name=self.name) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_page.py b/pyatlan_v9/model/assets/sigma_page.py index d6d6f2dcb..13ce0d3f0 100644 --- a/pyatlan_v9/model/assets/sigma_page.py +++ b/pyatlan_v9/model/assets/sigma_page.py @@ -361,17 +361,27 @@ def creator( connector_name=fields[1], sigma_workbook_qualified_name=workbook_qualified_name, sigma_workbook_name=workbook_name, - sigma_workbook=RelatedSigmaWorkbook(qualified_name=workbook_qualified_name), + sigma_workbook=RelatedSigmaWorkbook( + unique_attributes={"qualifiedName": workbook_qualified_name} + ), ) @classmethod def create(cls, **kwargs) -> "SigmaPage": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaPage": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaPage": return cls.updater(**kwargs) + def trim_to_required(self) -> "SigmaPage": + return SigmaPage.updater(qualified_name=self.qualified_name, name=self.name) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/sigma_workbook.py b/pyatlan_v9/model/assets/sigma_workbook.py index 598ba7ff0..a63dd0f52 100644 --- a/pyatlan_v9/model/assets/sigma_workbook.py +++ b/pyatlan_v9/model/assets/sigma_workbook.py @@ -343,10 +343,18 @@ def creator( def create(cls, **kwargs) -> "SigmaWorkbook": return cls.creator(**kwargs) + @classmethod + def updater(cls, *, qualified_name: str, name: str) -> "SigmaWorkbook": + validate_required_fields(["qualified_name", "name"], [qualified_name, name]) + return cls(qualified_name=qualified_name, name=name) + @classmethod def create_for_modification(cls, **kwargs) -> "SigmaWorkbook": return cls.updater(**kwargs) + def trim_to_required(self) -> "SigmaWorkbook": + return SigmaWorkbook.updater(qualified_name=self.qualified_name, name=self.name) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/tests_v9/unit/model/sigma_data_element_field_test.py b/tests_v9/unit/model/sigma_data_element_field_test.py new file mode 100644 index 000000000..2e0ec42aa --- /dev/null +++ b/tests_v9/unit/model/sigma_data_element_field_test.py @@ -0,0 +1,134 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for SigmaDataElementField.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-sigma-app emits for a `data_element_fields` record +(FND-3838 census), so a creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import SigmaDataElementField + +NAME = "Revenue" +DATA_ELEMENT_QUALIFIED_NAME = "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34/el-Xy9Zw8Vu" +DATA_ELEMENT_NAME = "Revenue by Region" +PAGE_NAME = "Summary" +WORKBOOK_NAME = "Sales Overview" +FIELD_ID = "fld_Qr7St6" +EXPECTED_QN = ( + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + "/el-Xy9Zw8Vu/fld_Qr7St6" +) + + +def _creator_kwargs(**overrides): + return { + "name": NAME, + "data_element_qualified_name": DATA_ELEMENT_QUALIFIED_NAME, + "data_element_name": DATA_ELEMENT_NAME, + "page_name": PAGE_NAME, + "workbook_name": WORKBOOK_NAME, + "field_id": FIELD_ID, + **overrides, + } + + +def _creator(**overrides): + return SigmaDataElementField.creator(**_creator_kwargs(**overrides)) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.connection_qualified_name == "default/sigma/1712345678" + assert sut.connector_name == "sigma" + assert ( + sut.sigma_data_element_qualified_name + == "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34/el-Xy9Zw8Vu" + ) + assert sut.sigma_data_element_name == "Revenue by Region" + assert ( + sut.sigma_page_qualified_name + == "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + ) + assert sut.sigma_page_name == "Summary" + assert ( + sut.sigma_workbook_qualified_name + == "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" + ) + assert sut.sigma_workbook_name == "Sales Overview" + assert sut.sigma_data_element.unique_attributes == { + "qualifiedName": "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34/el-Xy9Zw8Vu" + } + sut.validate(for_creation=True) + + +def test_create_alias(): + assert ( + SigmaDataElementField.create(**_creator_kwargs()).qualified_name == EXPECTED_QN + ) + + +@pytest.mark.parametrize( + "param", + [ + "name", + "data_element_qualified_name", + "data_element_name", + "page_name", + "workbook_name", + "field_id", + ], +) +@pytest.mark.parametrize( + "value, message", [(None, "is required"), (" ", "cannot be blank")] +) +def test_creator_rejects_missing_or_blank_inputs(param, value, message): + with pytest.raises(ValueError, match=f"{param} {message}"): + _creator(**{param: value}) + + +@pytest.mark.parametrize( + "param, bad_qn", + [ + # page qn, one segment short + ( + "data_element_qualified_name", + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34", + ), + # field qn, one segment long + ( + "data_element_qualified_name", + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34/el-Xy9Zw8Vu/fld_Qr7St6", + ), + ], +) +def test_creator_rejects_malformed_parent_qualified_name(param, bad_qn): + with pytest.raises(ValueError, match=f"Invalid {param}"): + _creator(**{param: bad_qn}) + + +def test_updater(): + sut = SigmaDataElementField.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +@pytest.mark.parametrize("param", ["qualified_name", "name"]) +def test_updater_rejects_missing_inputs(param): + args = {"qualified_name": EXPECTED_QN, "name": NAME, param: None} + with pytest.raises(ValueError, match=f"{param} is required"): + SigmaDataElementField.updater(**args) + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert isinstance(sut, SigmaDataElementField) + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/sigma_data_element_test.py b/tests_v9/unit/model/sigma_data_element_test.py new file mode 100644 index 000000000..138e37931 --- /dev/null +++ b/tests_v9/unit/model/sigma_data_element_test.py @@ -0,0 +1,117 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for SigmaDataElement.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-sigma-app emits for a `data_elements` record +(FND-3838 census), so a creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import SigmaDataElement + +NAME = "Revenue" +PAGE_QUALIFIED_NAME = ( + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" +) +PAGE_NAME = "Summary" +WORKBOOK_NAME = "Sales Overview" +DATA_ELEMENT_ID = "el-Xy9Zw8Vu" +EXPECTED_QN = "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34/el-Xy9Zw8Vu" + + +def _creator_kwargs(**overrides): + return { + "name": NAME, + "page_qualified_name": PAGE_QUALIFIED_NAME, + "page_name": PAGE_NAME, + "workbook_name": WORKBOOK_NAME, + "data_element_id": DATA_ELEMENT_ID, + **overrides, + } + + +def _creator(**overrides): + return SigmaDataElement.creator(**_creator_kwargs(**overrides)) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.connection_qualified_name == "default/sigma/1712345678" + assert sut.connector_name == "sigma" + assert ( + sut.sigma_page_qualified_name + == "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + ) + assert sut.sigma_page_name == "Summary" + assert ( + sut.sigma_workbook_qualified_name + == "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" + ) + assert sut.sigma_workbook_name == "Sales Overview" + assert sut.sigma_page.unique_attributes == { + "qualifiedName": "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + } + sut.validate(for_creation=True) + + +def test_create_alias(): + assert SigmaDataElement.create(**_creator_kwargs()).qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize( + "param", + ["name", "page_qualified_name", "page_name", "workbook_name", "data_element_id"], +) +@pytest.mark.parametrize( + "value, message", [(None, "is required"), (" ", "cannot be blank")] +) +def test_creator_rejects_missing_or_blank_inputs(param, value, message): + with pytest.raises(ValueError, match=f"{param} {message}"): + _creator(**{param: value}) + + +@pytest.mark.parametrize( + "param, bad_qn", + [ + # workbook qn, one segment short + ( + "page_qualified_name", + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f", + ), + # data element qn, one segment long + ( + "page_qualified_name", + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34/el-Xy9Zw8Vu", + ), + ], +) +def test_creator_rejects_malformed_parent_qualified_name(param, bad_qn): + with pytest.raises(ValueError, match=f"Invalid {param}"): + _creator(**{param: bad_qn}) + + +def test_updater(): + sut = SigmaDataElement.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +@pytest.mark.parametrize("param", ["qualified_name", "name"]) +def test_updater_rejects_missing_inputs(param): + args = {"qualified_name": EXPECTED_QN, "name": NAME, param: None} + with pytest.raises(ValueError, match=f"{param} is required"): + SigmaDataElement.updater(**args) + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert isinstance(sut, SigmaDataElement) + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/sigma_data_model_column_test.py b/tests_v9/unit/model/sigma_data_model_column_test.py new file mode 100644 index 000000000..689180666 --- /dev/null +++ b/tests_v9/unit/model/sigma_data_model_column_test.py @@ -0,0 +1,110 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for SigmaDataModelColumn.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-sigma-app emits for a `data_model_columns` record +(FND-3838 census), so a creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import SigmaDataModelColumn + +NAME = "Revenue" +DATA_MODEL_QUALIFIED_NAME = "default/sigma/1712345678/dataModels/dm-1a2b3c4d" +DATA_MODEL_NAME = "Orders Model" +ELEMENT_ID = "dmel_Mn5Op4" +COLUMN_ID = "col_Gh3Ij2" +EXPECTED_QN = "default/sigma/1712345678/dataModels/dm-1a2b3c4d/dmel_Mn5Op4/col_Gh3Ij2" + + +def _creator_kwargs(**overrides): + return { + "name": NAME, + "data_model_qualified_name": DATA_MODEL_QUALIFIED_NAME, + "data_model_name": DATA_MODEL_NAME, + "element_id": ELEMENT_ID, + "column_id": COLUMN_ID, + **overrides, + } + + +def _creator(**overrides): + return SigmaDataModelColumn.creator(**_creator_kwargs(**overrides)) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.connection_qualified_name == "default/sigma/1712345678" + assert sut.connector_name == "sigma" + assert ( + sut.sigma_data_model_qualified_name + == "default/sigma/1712345678/dataModels/dm-1a2b3c4d" + ) + assert sut.sigma_data_model_name == "Orders Model" + assert sut.asset_source_id == "col_Gh3Ij2" + assert sut.sigma_data_model.unique_attributes == { + "qualifiedName": "default/sigma/1712345678/dataModels/dm-1a2b3c4d" + } + sut.validate(for_creation=True) + + +def test_create_alias(): + assert ( + SigmaDataModelColumn.create(**_creator_kwargs()).qualified_name == EXPECTED_QN + ) + + +@pytest.mark.parametrize( + "param", + ["name", "data_model_qualified_name", "data_model_name", "element_id", "column_id"], +) +@pytest.mark.parametrize( + "value, message", [(None, "is required"), (" ", "cannot be blank")] +) +def test_creator_rejects_missing_or_blank_inputs(param, value, message): + with pytest.raises(ValueError, match=f"{param} {message}"): + _creator(**{param: value}) + + +@pytest.mark.parametrize( + "param, bad_qn", + [ + # connection qn, one segment short + ("data_model_qualified_name", "default/sigma/1712345678"), + # right length, no dataModels segment + ( + "data_model_qualified_name", + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34", + ), + ], +) +def test_creator_rejects_malformed_parent_qualified_name(param, bad_qn): + with pytest.raises(ValueError, match=f"Invalid {param}"): + _creator(**{param: bad_qn}) + + +def test_updater(): + sut = SigmaDataModelColumn.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +@pytest.mark.parametrize("param", ["qualified_name", "name"]) +def test_updater_rejects_missing_inputs(param): + args = {"qualified_name": EXPECTED_QN, "name": NAME, param: None} + with pytest.raises(ValueError, match=f"{param} is required"): + SigmaDataModelColumn.updater(**args) + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert isinstance(sut, SigmaDataModelColumn) + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/sigma_data_model_test.py b/tests_v9/unit/model/sigma_data_model_test.py new file mode 100644 index 000000000..57f05fbf4 --- /dev/null +++ b/tests_v9/unit/model/sigma_data_model_test.py @@ -0,0 +1,85 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for SigmaDataModel.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-sigma-app emits for a `data_models` record +(FND-3838 census), so a creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import SigmaDataModel + +NAME = "Revenue" +CONNECTION_QUALIFIED_NAME = "default/sigma/1712345678" +DATA_MODEL_ID = "dm-1a2b3c4d" +EXPECTED_QN = "default/sigma/1712345678/dataModels/dm-1a2b3c4d" + + +def _creator_kwargs(**overrides): + return { + "name": NAME, + "connection_qualified_name": CONNECTION_QUALIFIED_NAME, + "data_model_id": DATA_MODEL_ID, + **overrides, + } + + +def _creator(**overrides): + return SigmaDataModel.creator(**_creator_kwargs(**overrides)) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.connection_qualified_name == "default/sigma/1712345678" + assert sut.connector_name == "sigma" + assert sut.asset_source_id == "dm-1a2b3c4d" + sut.validate(for_creation=True) + + +def test_create_alias(): + assert SigmaDataModel.create(**_creator_kwargs()).qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize( + "param", ["name", "connection_qualified_name", "data_model_id"] +) +@pytest.mark.parametrize( + "value, message", [(None, "is required"), (" ", "cannot be blank")] +) +def test_creator_rejects_missing_or_blank_inputs(param, value, message): + with pytest.raises(ValueError, match=f"{param} {message}"): + _creator(**{param: value}) + + +def test_updater(): + sut = SigmaDataModel.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +@pytest.mark.parametrize("param", ["qualified_name", "name"]) +def test_updater_rejects_missing_inputs(param): + args = {"qualified_name": EXPECTED_QN, "name": NAME, param: None} + with pytest.raises(ValueError, match=f"{param} is required"): + SigmaDataModel.updater(**args) + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert isinstance(sut, SigmaDataModel) + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +def test_pattern_requires_data_models_segment(): + page_qn = ( + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" + ) + assert not SigmaDataModel._QUALIFIED_NAME_PATTERN.match(page_qn) diff --git a/tests_v9/unit/model/sigma_dataset_test.py b/tests_v9/unit/model/sigma_dataset_test.py new file mode 100644 index 000000000..226a66047 --- /dev/null +++ b/tests_v9/unit/model/sigma_dataset_test.py @@ -0,0 +1,75 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for SigmaDataset.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-sigma-app emits for a `datasets` record +(FND-3838 census), so a creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import SigmaDataset + +NAME = "Revenue" +CONNECTION_QUALIFIED_NAME = "default/sigma/1712345678" +DATASET_ID = "9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b" +EXPECTED_QN = "default/sigma/1712345678/9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b" + + +def _creator_kwargs(**overrides): + return { + "name": NAME, + "connection_qualified_name": CONNECTION_QUALIFIED_NAME, + "dataset_id": DATASET_ID, + **overrides, + } + + +def _creator(**overrides): + return SigmaDataset.creator(**_creator_kwargs(**overrides)) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.connection_qualified_name == "default/sigma/1712345678" + assert sut.connector_name == "sigma" + sut.validate(for_creation=True) + + +def test_create_alias(): + assert SigmaDataset.create(**_creator_kwargs()).qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize("param", ["name", "connection_qualified_name", "dataset_id"]) +@pytest.mark.parametrize( + "value, message", [(None, "is required"), (" ", "cannot be blank")] +) +def test_creator_rejects_missing_or_blank_inputs(param, value, message): + with pytest.raises(ValueError, match=f"{param} {message}"): + _creator(**{param: value}) + + +def test_updater(): + sut = SigmaDataset.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +@pytest.mark.parametrize("param", ["qualified_name", "name"]) +def test_updater_rejects_missing_inputs(param): + args = {"qualified_name": EXPECTED_QN, "name": NAME, param: None} + with pytest.raises(ValueError, match=f"{param} is required"): + SigmaDataset.updater(**args) + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert isinstance(sut, SigmaDataset) + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/sigma_page_test.py b/tests_v9/unit/model/sigma_page_test.py new file mode 100644 index 000000000..f5743542f --- /dev/null +++ b/tests_v9/unit/model/sigma_page_test.py @@ -0,0 +1,108 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for SigmaPage.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-sigma-app emits for a `pages` record +(FND-3838 census), so a creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import SigmaPage + +NAME = "Revenue" +WORKBOOK_QUALIFIED_NAME = ( + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" +) +WORKBOOK_NAME = "Sales Overview" +PAGE_ID = "pg_Ab12Cd34" +EXPECTED_QN = ( + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" +) + + +def _creator_kwargs(**overrides): + return { + "name": NAME, + "workbook_qualified_name": WORKBOOK_QUALIFIED_NAME, + "workbook_name": WORKBOOK_NAME, + "page_id": PAGE_ID, + **overrides, + } + + +def _creator(**overrides): + return SigmaPage.creator(**_creator_kwargs(**overrides)) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.connection_qualified_name == "default/sigma/1712345678" + assert sut.connector_name == "sigma" + assert ( + sut.sigma_workbook_qualified_name + == "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" + ) + assert sut.sigma_workbook_name == "Sales Overview" + assert sut.sigma_workbook.unique_attributes == { + "qualifiedName": "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" + } + sut.validate(for_creation=True) + + +def test_create_alias(): + assert SigmaPage.create(**_creator_kwargs()).qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize( + "param", ["name", "workbook_qualified_name", "workbook_name", "page_id"] +) +@pytest.mark.parametrize( + "value, message", [(None, "is required"), (" ", "cannot be blank")] +) +def test_creator_rejects_missing_or_blank_inputs(param, value, message): + with pytest.raises(ValueError, match=f"{param} {message}"): + _creator(**{param: value}) + + +@pytest.mark.parametrize( + "param, bad_qn", + [ + # connection qn, one segment short + ("workbook_qualified_name", "default/sigma/1712345678"), + # page qn, one segment long + ( + "workbook_qualified_name", + "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34", + ), + ], +) +def test_creator_rejects_malformed_parent_qualified_name(param, bad_qn): + with pytest.raises(ValueError, match=f"Invalid {param}"): + _creator(**{param: bad_qn}) + + +def test_updater(): + sut = SigmaPage.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +@pytest.mark.parametrize("param", ["qualified_name", "name"]) +def test_updater_rejects_missing_inputs(param): + args = {"qualified_name": EXPECTED_QN, "name": NAME, param: None} + with pytest.raises(ValueError, match=f"{param} is required"): + SigmaPage.updater(**args) + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert isinstance(sut, SigmaPage) + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/sigma_test.py b/tests_v9/unit/model/sigma_test.py deleted file mode 100644 index fd17e4dca..000000000 --- a/tests_v9/unit/model/sigma_test.py +++ /dev/null @@ -1,230 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# Copyright 2026 Atlan Pte. Ltd. - -"""Unit tests for the Sigma creators in pyatlan_v9. - -The expected qualifiedNames are the exact strings atlan-sigma-app emits today -(FND-3838 census), so a creator change that would re-key existing Sigma assets fails here. -""" - -import pytest - -from pyatlan_v9.model import ( - SigmaDataElement, - SigmaDataElementField, - SigmaDataModel, - SigmaDataModelColumn, - SigmaDataset, - SigmaPage, - SigmaWorkbook, -) - -CONNECTION_QN = "default/sigma/1712345678" -WORKBOOK_ID = "3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" -PAGE_ID = "pg_Ab12Cd34" -DATA_ELEMENT_ID = "el-Xy9Zw8Vu" -FIELD_ID = "fld_Qr7St6" -DATASET_ID = "9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b" -DATA_MODEL_ID = "dm-1a2b3c4d" -ELEMENT_ID = "dmel_Mn5Op4" -COLUMN_ID = "col_Gh3Ij2" - -WORKBOOK_QN = f"{CONNECTION_QN}/{WORKBOOK_ID}" -PAGE_QN = f"{WORKBOOK_QN}/{PAGE_ID}" -DATA_ELEMENT_QN = f"{PAGE_QN}/{DATA_ELEMENT_ID}" -DATA_MODEL_QN = f"{CONNECTION_QN}/dataModels/{DATA_MODEL_ID}" - -NAME = "Revenue" -WORKBOOK_NAME = "Sales Overview" -PAGE_NAME = "Summary" -DATA_ELEMENT_NAME = "Revenue by Region" -DATA_MODEL_NAME = "Orders Model" - -# (type, creator kwargs, the literal qualifiedName atlan-sigma-app emits) -ALL_TYPES = [ - ( - SigmaWorkbook, - dict(connection_qualified_name=CONNECTION_QN, workbook_id=WORKBOOK_ID), - "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f", - ), - ( - SigmaDataset, - dict(connection_qualified_name=CONNECTION_QN, dataset_id=DATASET_ID), - "default/sigma/1712345678/9e8d7c6b-5a4f-4e3d-8c2b-1a0f9e8d7c6b", - ), - ( - SigmaDataModel, - dict(connection_qualified_name=CONNECTION_QN, data_model_id=DATA_MODEL_ID), - "default/sigma/1712345678/dataModels/dm-1a2b3c4d", - ), - ( - SigmaPage, - dict( - workbook_qualified_name=WORKBOOK_QN, - workbook_name=WORKBOOK_NAME, - page_id=PAGE_ID, - ), - "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34", - ), - ( - SigmaDataElement, - dict( - page_qualified_name=PAGE_QN, - page_name=PAGE_NAME, - workbook_name=WORKBOOK_NAME, - data_element_id=DATA_ELEMENT_ID, - ), - "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" - "/el-Xy9Zw8Vu", - ), - ( - SigmaDataElementField, - dict( - data_element_qualified_name=DATA_ELEMENT_QN, - data_element_name=DATA_ELEMENT_NAME, - page_name=PAGE_NAME, - workbook_name=WORKBOOK_NAME, - field_id=FIELD_ID, - ), - "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f/pg_Ab12Cd34" - "/el-Xy9Zw8Vu/fld_Qr7St6", - ), - ( - SigmaDataModelColumn, - dict( - data_model_qualified_name=DATA_MODEL_QN, - data_model_name=DATA_MODEL_NAME, - element_id=ELEMENT_ID, - column_id=COLUMN_ID, - ), - "default/sigma/1712345678/dataModels/dm-1a2b3c4d/dmel_Mn5Op4/col_Gh3Ij2", - ), -] -IDS = [t[0].__name__ for t in ALL_TYPES] - - -@pytest.mark.parametrize("asset_type, kwargs, expected_qn", ALL_TYPES, ids=IDS) -def test_creator_matches_app_qualified_name(asset_type, kwargs, expected_qn): - sut = asset_type.creator(name=NAME, **kwargs) - - assert sut.qualified_name == expected_qn - assert sut.name == NAME - assert sut.connection_qualified_name == CONNECTION_QN - assert sut.connector_name == "sigma" - assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) - sut.validate(for_creation=True) - - -@pytest.mark.parametrize("asset_type, kwargs, expected_qn", ALL_TYPES, ids=IDS) -def test_create_alias(asset_type, kwargs, expected_qn): - assert asset_type.create(name=NAME, **kwargs).qualified_name == expected_qn - - -@pytest.mark.parametrize("asset_type, kwargs, _", ALL_TYPES, ids=IDS) -def test_creator_rejects_missing_or_blank_inputs(asset_type, kwargs, _): - for param in ["name", *kwargs]: - args = {"name": NAME, **kwargs} - args[param] = None - with pytest.raises(ValueError, match=f"{param} is required"): - asset_type.creator(**args) - args[param] = " " - with pytest.raises(ValueError, match=f"{param} cannot be blank"): - asset_type.creator(**args) - - -def test_page_sets_workbook(): - sut = SigmaPage.creator( - name=NAME, - workbook_qualified_name=WORKBOOK_QN, - workbook_name=WORKBOOK_NAME, - page_id=PAGE_ID, - ) - - assert sut.sigma_workbook.qualified_name == WORKBOOK_QN - assert sut.sigma_workbook_qualified_name == WORKBOOK_QN - assert sut.sigma_workbook_name == WORKBOOK_NAME - - -def test_data_element_sets_page_and_workbook(): - sut = SigmaDataElement.creator( - name=NAME, - page_qualified_name=PAGE_QN, - page_name=PAGE_NAME, - workbook_name=WORKBOOK_NAME, - data_element_id=DATA_ELEMENT_ID, - ) - - assert sut.sigma_page.qualified_name == PAGE_QN - assert sut.sigma_page_qualified_name == PAGE_QN - assert sut.sigma_page_name == PAGE_NAME - assert sut.sigma_workbook_qualified_name == WORKBOOK_QN - assert sut.sigma_workbook_name == WORKBOOK_NAME - - -def test_data_element_field_sets_all_ancestors(): - sut = SigmaDataElementField.creator( - name=NAME, - data_element_qualified_name=DATA_ELEMENT_QN, - data_element_name=DATA_ELEMENT_NAME, - page_name=PAGE_NAME, - workbook_name=WORKBOOK_NAME, - field_id=FIELD_ID, - ) - - assert sut.sigma_data_element.qualified_name == DATA_ELEMENT_QN - assert sut.sigma_data_element_qualified_name == DATA_ELEMENT_QN - assert sut.sigma_data_element_name == DATA_ELEMENT_NAME - assert sut.sigma_page_qualified_name == PAGE_QN - assert sut.sigma_page_name == PAGE_NAME - assert sut.sigma_workbook_qualified_name == WORKBOOK_QN - assert sut.sigma_workbook_name == WORKBOOK_NAME - - -def test_data_model_sets_asset_source_id(): - sut = SigmaDataModel.creator( - name=NAME, connection_qualified_name=CONNECTION_QN, data_model_id=DATA_MODEL_ID - ) - - assert sut.asset_source_id == DATA_MODEL_ID - - -def test_data_model_column_sets_data_model(): - sut = SigmaDataModelColumn.creator( - name=NAME, - data_model_qualified_name=DATA_MODEL_QN, - data_model_name=DATA_MODEL_NAME, - element_id=ELEMENT_ID, - column_id=COLUMN_ID, - ) - - assert sut.sigma_data_model.qualified_name == DATA_MODEL_QN - assert sut.sigma_data_model_qualified_name == DATA_MODEL_QN - assert sut.sigma_data_model_name == DATA_MODEL_NAME - assert sut.asset_source_id == COLUMN_ID - - -@pytest.mark.parametrize( - "asset_type, parent_param, bad_parent_qn", - [ - (SigmaPage, "workbook_qualified_name", CONNECTION_QN), - (SigmaPage, "workbook_qualified_name", PAGE_QN), - (SigmaDataElement, "page_qualified_name", WORKBOOK_QN), - (SigmaDataElement, "page_qualified_name", DATA_ELEMENT_QN), - (SigmaDataElementField, "data_element_qualified_name", PAGE_QN), - (SigmaDataElementField, "data_element_qualified_name", f"{DATA_ELEMENT_QN}/x"), - (SigmaDataModelColumn, "data_model_qualified_name", CONNECTION_QN), - # a page qn has the data model's segment count but not the `dataModels` literal - (SigmaDataModelColumn, "data_model_qualified_name", PAGE_QN), - ], -) -def test_creator_rejects_malformed_parent_qualified_name( - asset_type, parent_param, bad_parent_qn -): - kwargs = next(k for t, k, _ in ALL_TYPES if t is asset_type) - with pytest.raises(ValueError, match=f"Invalid {parent_param}"): - asset_type.creator(name=NAME, **{**kwargs, parent_param: bad_parent_qn}) - - -def test_data_model_patterns_require_data_models_segment(): - assert not SigmaDataModel._QUALIFIED_NAME_PATTERN.match(PAGE_QN) - assert not SigmaDataModelColumn._QUALIFIED_NAME_PATTERN.match(DATA_ELEMENT_QN) diff --git a/tests_v9/unit/model/sigma_workbook_test.py b/tests_v9/unit/model/sigma_workbook_test.py new file mode 100644 index 000000000..924820a42 --- /dev/null +++ b/tests_v9/unit/model/sigma_workbook_test.py @@ -0,0 +1,75 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for SigmaWorkbook.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-sigma-app emits for a `workbooks` record +(FND-3838 census), so a creator change that would re-key existing assets fails here. +""" + +import pytest + +from pyatlan_v9.model import SigmaWorkbook + +NAME = "Revenue" +CONNECTION_QUALIFIED_NAME = "default/sigma/1712345678" +WORKBOOK_ID = "3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" +EXPECTED_QN = "default/sigma/1712345678/3f2b9c1e-7a4d-4e8b-9c0d-1a2b3c4d5e6f" + + +def _creator_kwargs(**overrides): + return { + "name": NAME, + "connection_qualified_name": CONNECTION_QUALIFIED_NAME, + "workbook_id": WORKBOOK_ID, + **overrides, + } + + +def _creator(**overrides): + return SigmaWorkbook.creator(**_creator_kwargs(**overrides)) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.connection_qualified_name == "default/sigma/1712345678" + assert sut.connector_name == "sigma" + sut.validate(for_creation=True) + + +def test_create_alias(): + assert SigmaWorkbook.create(**_creator_kwargs()).qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize("param", ["name", "connection_qualified_name", "workbook_id"]) +@pytest.mark.parametrize( + "value, message", [(None, "is required"), (" ", "cannot be blank")] +) +def test_creator_rejects_missing_or_blank_inputs(param, value, message): + with pytest.raises(ValueError, match=f"{param} {message}"): + _creator(**{param: value}) + + +def test_updater(): + sut = SigmaWorkbook.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +@pytest.mark.parametrize("param", ["qualified_name", "name"]) +def test_updater_rejects_missing_inputs(param): + args = {"qualified_name": EXPECTED_QN, "name": NAME, param: None} + with pytest.raises(ValueError, match=f"{param} is required"): + SigmaWorkbook.updater(**args) + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert isinstance(sut, SigmaWorkbook) + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME