diff --git a/pyatlan_v9/model/assets/_overlays/qlik_app.py b/pyatlan_v9/model/assets/_overlays/qlik_app.py new file mode 100644 index 000000000..f67872d3b --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_app.py @@ -0,0 +1,52 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + space_qualified_name: str, + ) -> "QlikApp": + """ + Build a Qlik app; qualifiedName is {connection}/app/{qlik_id}. + + space_qualified_name is the qualifiedName of the QlikSpace + ({connection}/space/{id}) or QlikStream ({connection}/stream/{id}) + the app lives in. Its segment decides which type the qlik_space + relationship points at. + """ + validate_required_fields( + ["name", "qlik_id", "space_qualified_name"], + [name, qlik_id, space_qualified_name], + ) + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + connection_qualified_name, space_kind, space_id = space_parts + 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}/app/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_space_id=space_id, + qlik_space_qualified_name=space_qualified_name, + qlik_space=RelatedQlikSpace( + type_name="QlikStream" if space_kind == "stream" else "QlikSpace", + qualified_name=space_qualified_name, + ), + ) + + @classmethod + def create(cls, **kwargs) -> "QlikApp": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikApp": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/qlik_chart.py b/pyatlan_v9/model/assets/_overlays/qlik_chart.py new file mode 100644 index 000000000..993a6a129 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_chart.py @@ -0,0 +1,58 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + sheet_qualified_name: str, + space_qualified_name: Union[str, None] = None, + ) -> "QlikChart": + """ + Build a Qlik chart; qualifiedName is {app_qualified_name}/chart/{qlik_id}. + + The chart's qualifiedName hangs off the app, not the sheet: the app is + taken from sheet_qualified_name ({app_qualified_name}/sheet/{id}). + space_qualified_name is optional and only fills the denormalised + qlik_space_id / qlik_space_qualified_name of the app's space or stream. + """ + validate_required_fields( + ["name", "qlik_id", "sheet_qualified_name"], + [name, qlik_id, sheet_qualified_name], + ) + sheet_parts = sheet_qualified_name.rsplit("/", 4) + if len(sheet_parts) != 5 or sheet_parts[1] != "app" or sheet_parts[3] != "sheet": + raise ValueError("sheet_qualified_name must end in /app//sheet/") + connection_qualified_name, _, app_id, _, _ = sheet_parts + app_qualified_name = f"{connection_qualified_name}/app/{app_id}" + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + asset = cls( + name=name, + qualified_name=f"{app_qualified_name}/chart/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_app_id=app_id, + qlik_app_qualified_name=app_qualified_name, + qlik_sheet=RelatedQlikSheet(qualified_name=sheet_qualified_name), + ) + if space_qualified_name: + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + asset.qlik_space_id = space_parts[2] + asset.qlik_space_qualified_name = space_qualified_name + return asset + + @classmethod + def create(cls, **kwargs) -> "QlikChart": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikChart": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/qlik_column.py b/pyatlan_v9/model/assets/_overlays/qlik_column.py new file mode 100644 index 000000000..a68cc7073 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_column.py @@ -0,0 +1,81 @@ +# STDLIB_IMPORT: import hashlib +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + parent_qualified_name: str, + space_qualified_name: Union[str, None] = None, + ) -> "QlikColumn": + """ + Build a Qlik column under a sheet, chart or dataset. + + qualifiedName is {parent_qualified_name}/column/{md5}, where md5 is taken + over "{parent path}/column/{name}" and the parent path is the parent's + qualifiedName without the connection (app//sheet/, + app//chart/ or dataset/), exactly as atlan-qlik-sense-app + builds it. The parent type is read from parent_qualified_name. + space_qualified_name is optional and only fills the denormalised + qlik_space_id / qlik_space_qualified_name of the space or stream. + """ + validate_required_fields( + ["name", "parent_qualified_name"], [name, parent_qualified_name] + ) + dataset_parts = parent_qualified_name.rsplit("/", 2) + app_parts = parent_qualified_name.rsplit("/", 4) + if len(dataset_parts) == 3 and dataset_parts[1] == "dataset": + connection_qualified_name, parent_kind = dataset_parts[0], "dataset" + elif ( + len(app_parts) == 5 + and app_parts[1] == "app" + and app_parts[3] in ("sheet", "chart") + ): + connection_qualified_name, parent_kind = app_parts[0], app_parts[3] + else: + raise ValueError( + "parent_qualified_name must end in /app//sheet/, " + "/app//chart/ or /dataset/" + ) + parent_path = parent_qualified_name[len(connection_qualified_name) + 1 :] + # deepcode ignore InsecureHash/test: an identifier, not a security key + column_hash = hashlib.md5( # noqa: S324 + f"{parent_path}/column/{name}".encode("utf-8") + ).hexdigest() + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + asset = cls( + name=name, + qualified_name=f"{parent_qualified_name}/column/{column_hash}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_column_name=name, + qlik_parent_qualified_name=parent_qualified_name, + ) + if parent_kind == "sheet": + asset.qlik_sheet = RelatedQlikSheet(qualified_name=parent_qualified_name) + elif parent_kind == "chart": + asset.qlik_chart = RelatedQlikChart(qualified_name=parent_qualified_name) + else: + asset.qlik_dataset = RelatedQlikDataset( + qualified_name=parent_qualified_name + ) + if space_qualified_name: + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + asset.qlik_space_id = space_parts[2] + asset.qlik_space_qualified_name = space_qualified_name + return asset + + @classmethod + def create(cls, **kwargs) -> "QlikColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikColumn": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/qlik_dataset.py b/pyatlan_v9/model/assets/_overlays/qlik_dataset.py new file mode 100644 index 000000000..93e983558 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_dataset.py @@ -0,0 +1,52 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + space_qualified_name: str, + ) -> "QlikDataset": + """ + Build a Qlik dataset; qualifiedName is {connection}/dataset/{qlik_id}. + + space_qualified_name is the qualifiedName of the QlikSpace + ({connection}/space/{id}) or QlikStream ({connection}/stream/{id}) + the dataset lives in. Its segment decides which type the qlik_space + relationship points at. + """ + validate_required_fields( + ["name", "qlik_id", "space_qualified_name"], + [name, qlik_id, space_qualified_name], + ) + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + connection_qualified_name, space_kind, space_id = space_parts + 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/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_space_id=space_id, + qlik_space_qualified_name=space_qualified_name, + qlik_space=RelatedQlikSpace( + type_name="QlikStream" if space_kind == "stream" else "QlikSpace", + qualified_name=space_qualified_name, + ), + ) + + @classmethod + def create(cls, **kwargs) -> "QlikDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikDataset": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/qlik_sheet.py b/pyatlan_v9/model/assets/_overlays/qlik_sheet.py new file mode 100644 index 000000000..8363ef6ab --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_sheet.py @@ -0,0 +1,55 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + app_qualified_name: str, + space_qualified_name: Union[str, None] = None, + ) -> "QlikSheet": + """ + Build a Qlik sheet; qualifiedName is {app_qualified_name}/sheet/{qlik_id}. + + space_qualified_name is optional and only fills the denormalised + qlik_space_id / qlik_space_qualified_name of the app's space or stream. + """ + validate_required_fields( + ["name", "qlik_id", "app_qualified_name"], + [name, qlik_id, app_qualified_name], + ) + app_parts = app_qualified_name.rsplit("/", 2) + if len(app_parts) != 3 or app_parts[1] != "app": + raise ValueError("app_qualified_name must end in /app/") + connection_qualified_name, _, app_id = app_parts + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + asset = cls( + name=name, + qualified_name=f"{app_qualified_name}/sheet/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_app_id=app_id, + qlik_app_qualified_name=app_qualified_name, + qlik_app=RelatedQlikApp(qualified_name=app_qualified_name), + ) + if space_qualified_name: + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + asset.qlik_space_id = space_parts[2] + asset.qlik_space_qualified_name = space_qualified_name + return asset + + @classmethod + def create(cls, **kwargs) -> "QlikSheet": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikSheet": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/qlik_space.py b/pyatlan_v9/model/assets/_overlays/qlik_space.py new file mode 100644 index 000000000..3b3ceaf50 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_space.py @@ -0,0 +1,33 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + qlik_id: str, + ) -> "QlikSpace": + """Build a Qlik space; qualifiedName is {connection}/space/{qlik_id}.""" + validate_required_fields( + ["name", "connection_qualified_name", "qlik_id"], + [name, connection_qualified_name, qlik_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}/space/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + ) + + @classmethod + def create(cls, **kwargs) -> "QlikSpace": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikSpace": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/_overlays/qlik_stream.py b/pyatlan_v9/model/assets/_overlays/qlik_stream.py new file mode 100644 index 000000000..a6ea13659 --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_stream.py @@ -0,0 +1,33 @@ +# INTERNAL_IMPORT: from pyatlan.utils import init_guid, validate_required_fields + + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + qlik_id: str, + ) -> "QlikStream": + """Build a Qlik stream; qualifiedName is {connection}/stream/{qlik_id}.""" + validate_required_fields( + ["name", "connection_qualified_name", "qlik_id"], + [name, connection_qualified_name, qlik_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}/stream/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + ) + + @classmethod + def create(cls, **kwargs) -> "QlikStream": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikStream": + return cls.updater(**kwargs) diff --git a/pyatlan_v9/model/assets/qlik_app.py b/pyatlan_v9/model/assets/qlik_app.py index 0395229f7..86a074896 100644 --- a/pyatlan_v9/model/assets/qlik_app.py +++ b/pyatlan_v9/model/assets/qlik_app.py @@ -26,6 +26,7 @@ ) from pyatlan_v9.model.serde import Serde, get_serde from pyatlan_v9.model.transform import register_asset +from pyatlan_v9.utils import init_guid, validate_required_fields from .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -282,7 +283,7 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= - _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/[^/]+/[^/]+$") + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/app/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -352,6 +353,57 @@ def relate(self) -> "RelatedQlikApp": return RelatedQlikApp(guid=self.guid) return RelatedQlikApp(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + space_qualified_name: str, + ) -> "QlikApp": + """ + Build a Qlik app; qualifiedName is {connection}/app/{qlik_id}. + + space_qualified_name is the qualifiedName of the QlikSpace + ({connection}/space/{id}) or QlikStream ({connection}/stream/{id}) + the app lives in. Its segment decides which type the qlik_space + relationship points at. + """ + validate_required_fields( + ["name", "qlik_id", "space_qualified_name"], + [name, qlik_id, space_qualified_name], + ) + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + connection_qualified_name, space_kind, space_id = space_parts + 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}/app/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_space_id=space_id, + qlik_space_qualified_name=space_qualified_name, + qlik_space=RelatedQlikSpace( + type_name="QlikStream" if space_kind == "stream" else "QlikSpace", + qualified_name=space_qualified_name, + ), + ) + + @classmethod + def create(cls, **kwargs) -> "QlikApp": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikApp": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/qlik_chart.py b/pyatlan_v9/model/assets/qlik_chart.py index ec177d130..b189917b9 100644 --- a/pyatlan_v9/model/assets/qlik_chart.py +++ b/pyatlan_v9/model/assets/qlik_chart.py @@ -26,6 +26,7 @@ ) from pyatlan_v9.model.serde import Serde, get_serde from pyatlan_v9.model.transform import register_asset +from pyatlan_v9.utils import init_guid, validate_required_fields from .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -278,9 +279,7 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= - _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile( - r"^.+/[^/]+/[^/]+/[^/]+/[^/]+$" - ) + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/chart/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -352,6 +351,67 @@ def relate(self) -> "RelatedQlikChart": return RelatedQlikChart(guid=self.guid) return RelatedQlikChart(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + sheet_qualified_name: str, + space_qualified_name: Union[str, None] = None, + ) -> "QlikChart": + """ + Build a Qlik chart; qualifiedName is {app_qualified_name}/chart/{qlik_id}. + + The chart's qualifiedName hangs off the app, not the sheet: the app is + taken from sheet_qualified_name ({app_qualified_name}/sheet/{id}). + space_qualified_name is optional and only fills the denormalised + qlik_space_id / qlik_space_qualified_name of the app's space or stream. + """ + validate_required_fields( + ["name", "qlik_id", "sheet_qualified_name"], + [name, qlik_id, sheet_qualified_name], + ) + sheet_parts = sheet_qualified_name.rsplit("/", 4) + if ( + len(sheet_parts) != 5 + or sheet_parts[1] != "app" + or sheet_parts[3] != "sheet" + ): + raise ValueError("sheet_qualified_name must end in /app//sheet/") + connection_qualified_name, _, app_id, _, _ = sheet_parts + app_qualified_name = f"{connection_qualified_name}/app/{app_id}" + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + asset = cls( + name=name, + qualified_name=f"{app_qualified_name}/chart/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_app_id=app_id, + qlik_app_qualified_name=app_qualified_name, + qlik_sheet=RelatedQlikSheet(qualified_name=sheet_qualified_name), + ) + if space_qualified_name: + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + asset.qlik_space_id = space_parts[2] + asset.qlik_space_qualified_name = space_qualified_name + return asset + + @classmethod + def create(cls, **kwargs) -> "QlikChart": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikChart": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/qlik_column.py b/pyatlan_v9/model/assets/qlik_column.py index 72b76fc0e..89443065b 100644 --- a/pyatlan_v9/model/assets/qlik_column.py +++ b/pyatlan_v9/model/assets/qlik_column.py @@ -14,6 +14,7 @@ from __future__ import annotations +import hashlib import re from typing import Any, ClassVar, List, Union @@ -26,6 +27,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 @@ -291,9 +293,7 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= - _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile( - r"^.+/[^/]+/[^/]+/[^/]+/[^/]+$" - ) + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/column/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -327,12 +327,16 @@ def validate(self, for_creation: bool = False) -> None: if for_creation: if self.connection_qualified_name is UNSET: errors.append("connection_qualified_name is required for creation") - if self.qlik_sheet is UNSET: - errors.append("qlik_sheet is required for creation") - if self.qlik_app_qualified_name is UNSET: - errors.append("qlik_app_qualified_name is required for creation") if self.qlik_space_qualified_name is UNSET: errors.append("qlik_space_qualified_name is required for creation") + if ( + self.qlik_sheet is UNSET + and self.qlik_chart is UNSET + and self.qlik_dataset is UNSET + ): + errors.append( + "one of qlik_sheet, qlik_chart, qlik_dataset is required for creation" + ) if errors: raise ValueError(f"QlikColumn validation failed: {errors}") @@ -365,6 +369,85 @@ def relate(self) -> "RelatedQlikColumn": return RelatedQlikColumn(guid=self.guid) return RelatedQlikColumn(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + parent_qualified_name: str, + space_qualified_name: Union[str, None] = None, + ) -> "QlikColumn": + """ + Build a Qlik column under a sheet, chart or dataset. + + qualifiedName is {parent_qualified_name}/column/{md5}, where md5 is taken + over "{parent path}/column/{name}" and the parent path is the parent's + qualifiedName without the connection (app//sheet/, + app//chart/ or dataset/), exactly as atlan-qlik-sense-app + builds it. The parent type is read from parent_qualified_name. + space_qualified_name is optional and only fills the denormalised + qlik_space_id / qlik_space_qualified_name of the space or stream. + """ + validate_required_fields( + ["name", "parent_qualified_name"], [name, parent_qualified_name] + ) + dataset_parts = parent_qualified_name.rsplit("/", 2) + app_parts = parent_qualified_name.rsplit("/", 4) + if len(dataset_parts) == 3 and dataset_parts[1] == "dataset": + connection_qualified_name, parent_kind = dataset_parts[0], "dataset" + elif ( + len(app_parts) == 5 + and app_parts[1] == "app" + and app_parts[3] in ("sheet", "chart") + ): + connection_qualified_name, parent_kind = app_parts[0], app_parts[3] + else: + raise ValueError( + "parent_qualified_name must end in /app//sheet/, " + "/app//chart/ or /dataset/" + ) + parent_path = parent_qualified_name[len(connection_qualified_name) + 1 :] + # deepcode ignore InsecureHash/test: an identifier, not a security key + column_hash = hashlib.md5( # noqa: S324 + f"{parent_path}/column/{name}".encode("utf-8") + ).hexdigest() + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + asset = cls( + name=name, + qualified_name=f"{parent_qualified_name}/column/{column_hash}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_column_name=name, + qlik_parent_qualified_name=parent_qualified_name, + ) + if parent_kind == "sheet": + asset.qlik_sheet = RelatedQlikSheet(qualified_name=parent_qualified_name) + elif parent_kind == "chart": + asset.qlik_chart = RelatedQlikChart(qualified_name=parent_qualified_name) + else: + asset.qlik_dataset = RelatedQlikDataset( + qualified_name=parent_qualified_name + ) + if space_qualified_name: + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + asset.qlik_space_id = space_parts[2] + asset.qlik_space_qualified_name = space_qualified_name + return asset + + @classmethod + def create(cls, **kwargs) -> "QlikColumn": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikColumn": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/qlik_dataset.py b/pyatlan_v9/model/assets/qlik_dataset.py index f2e856004..159e488ee 100644 --- a/pyatlan_v9/model/assets/qlik_dataset.py +++ b/pyatlan_v9/model/assets/qlik_dataset.py @@ -26,6 +26,7 @@ ) from pyatlan_v9.model.serde import Serde, get_serde from pyatlan_v9.model.transform import register_asset +from pyatlan_v9.utils import init_guid, validate_required_fields from .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -282,7 +283,7 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= - _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/[^/]+/[^/]+$") + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/dataset/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -352,6 +353,57 @@ def relate(self) -> "RelatedQlikDataset": return RelatedQlikDataset(guid=self.guid) return RelatedQlikDataset(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + space_qualified_name: str, + ) -> "QlikDataset": + """ + Build a Qlik dataset; qualifiedName is {connection}/dataset/{qlik_id}. + + space_qualified_name is the qualifiedName of the QlikSpace + ({connection}/space/{id}) or QlikStream ({connection}/stream/{id}) + the dataset lives in. Its segment decides which type the qlik_space + relationship points at. + """ + validate_required_fields( + ["name", "qlik_id", "space_qualified_name"], + [name, qlik_id, space_qualified_name], + ) + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + connection_qualified_name, space_kind, space_id = space_parts + 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/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_space_id=space_id, + qlik_space_qualified_name=space_qualified_name, + qlik_space=RelatedQlikSpace( + type_name="QlikStream" if space_kind == "stream" else "QlikSpace", + qualified_name=space_qualified_name, + ), + ) + + @classmethod + def create(cls, **kwargs) -> "QlikDataset": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikDataset": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/qlik_sheet.py b/pyatlan_v9/model/assets/qlik_sheet.py index 09f1f2126..b924a3ba4 100644 --- a/pyatlan_v9/model/assets/qlik_sheet.py +++ b/pyatlan_v9/model/assets/qlik_sheet.py @@ -26,6 +26,7 @@ ) from pyatlan_v9.model.serde import Serde, get_serde from pyatlan_v9.model.transform import register_asset +from pyatlan_v9.utils import init_guid, validate_required_fields from .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -275,9 +276,7 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= - _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile( - r"^.+/[^/]+/[^/]+/[^/]+$" - ) + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/sheet/[^/]+$") def validate(self, for_creation: bool = False) -> None: """ @@ -349,6 +348,60 @@ def relate(self) -> "RelatedQlikSheet": return RelatedQlikSheet(guid=self.guid) return RelatedQlikSheet(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + qlik_id: str, + app_qualified_name: str, + space_qualified_name: Union[str, None] = None, + ) -> "QlikSheet": + """ + Build a Qlik sheet; qualifiedName is {app_qualified_name}/sheet/{qlik_id}. + + space_qualified_name is optional and only fills the denormalised + qlik_space_id / qlik_space_qualified_name of the app's space or stream. + """ + validate_required_fields( + ["name", "qlik_id", "app_qualified_name"], + [name, qlik_id, app_qualified_name], + ) + app_parts = app_qualified_name.rsplit("/", 2) + if len(app_parts) != 3 or app_parts[1] != "app": + raise ValueError("app_qualified_name must end in /app/") + connection_qualified_name, _, app_id = app_parts + fields = connection_qualified_name.split("/") + connector_name = fields[1] if len(fields) > 1 else None + asset = cls( + name=name, + qualified_name=f"{app_qualified_name}/sheet/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + qlik_app_id=app_id, + qlik_app_qualified_name=app_qualified_name, + qlik_app=RelatedQlikApp(qualified_name=app_qualified_name), + ) + if space_qualified_name: + space_parts = space_qualified_name.rsplit("/", 2) + if len(space_parts) != 3 or space_parts[1] not in ("space", "stream"): + raise ValueError( + "space_qualified_name must end in /space/ or /stream/" + ) + asset.qlik_space_id = space_parts[2] + asset.qlik_space_qualified_name = space_qualified_name + return asset + + @classmethod + def create(cls, **kwargs) -> "QlikSheet": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikSheet": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/qlik_space.py b/pyatlan_v9/model/assets/qlik_space.py index df5bc44b4..c48c0ad1f 100644 --- a/pyatlan_v9/model/assets/qlik_space.py +++ b/pyatlan_v9/model/assets/qlik_space.py @@ -14,6 +14,7 @@ from __future__ import annotations +import re from typing import Any, ClassVar, List, Union import msgspec @@ -25,6 +26,7 @@ ) from pyatlan_v9.model.serde import Serde, get_serde from pyatlan_v9.model.transform import register_asset +from pyatlan_v9.utils import init_guid, validate_required_fields from .airflow_related import RelatedAirflowTask from .anomalo_related import RelatedAnomaloCheck @@ -265,6 +267,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/space/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this QlikSpace instance. @@ -289,6 +293,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"QlikSpace validation failed: {errors}") @@ -321,6 +330,38 @@ def relate(self) -> "RelatedQlikSpace": return RelatedQlikSpace(guid=self.guid) return RelatedQlikSpace(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + qlik_id: str, + ) -> "QlikSpace": + """Build a Qlik space; qualifiedName is {connection}/space/{qlik_id}.""" + validate_required_fields( + ["name", "connection_qualified_name", "qlik_id"], + [name, connection_qualified_name, qlik_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}/space/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + ) + + @classmethod + def create(cls, **kwargs) -> "QlikSpace": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikSpace": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/pyatlan_v9/model/assets/qlik_stream.py b/pyatlan_v9/model/assets/qlik_stream.py index ce5d669f1..094b6cc22 100644 --- a/pyatlan_v9/model/assets/qlik_stream.py +++ b/pyatlan_v9/model/assets/qlik_stream.py @@ -14,11 +14,20 @@ from __future__ import annotations +import re from typing import Any, ClassVar, List, Union import msgspec from msgspec import UNSET, UnsetType +from pyatlan_v9.model.conversion_utils import ( + categorize_relationships, + merge_relationships, +) +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 from .app_related import RelatedApplication, RelatedApplicationField @@ -42,19 +51,12 @@ from .monte_carlo_related import RelatedMCIncident, RelatedMCMonitor from .partial_related import RelatedPartialField, RelatedPartialObject from .process_related import RelatedProcess +from .qlik_related import RelatedQlikApp, RelatedQlikDataset, RelatedQlikStream from .referenceable_related import RelatedReferenceable from .resource_related import RelatedFile, RelatedLink, RelatedReadme from .schema_registry_related import RelatedSchemaRegistrySubject from .soda_related import RelatedSodaCheck from .spark_related import RelatedSparkJob -from pyatlan_v9.model.conversion_utils import ( - categorize_relationships, - merge_relationships, -) -from pyatlan_v9.model.serde import Serde, get_serde -from pyatlan_v9.model.transform import register_asset - -from .qlik_related import RelatedQlikApp, RelatedQlikDataset, RelatedQlikStream # ============================================================================= # FLAT ASSET CLASS @@ -265,6 +267,8 @@ def __post_init__(self) -> None: # SDK Methods # ========================================================================= + _QUALIFIED_NAME_PATTERN: ClassVar[re.Pattern] = re.compile(r"^.+/stream/[^/]+$") + def validate(self, for_creation: bool = False) -> None: """ Dry-run validation of this QlikStream instance. @@ -289,6 +293,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"QlikStream validation failed: {errors}") @@ -321,6 +330,38 @@ def relate(self) -> "RelatedQlikStream": return RelatedQlikStream(guid=self.guid) return RelatedQlikStream(qualified_name=self.qualified_name) + @classmethod + @init_guid + def creator( + cls, + *, + name: str, + connection_qualified_name: str, + qlik_id: str, + ) -> "QlikStream": + """Build a Qlik stream; qualifiedName is {connection}/stream/{qlik_id}.""" + validate_required_fields( + ["name", "connection_qualified_name", "qlik_id"], + [name, connection_qualified_name, qlik_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}/stream/{qlik_id}", + connection_qualified_name=connection_qualified_name, + connector_name=connector_name, + qlik_id=qlik_id, + ) + + @classmethod + def create(cls, **kwargs) -> "QlikStream": + return cls.creator(**kwargs) + + @classmethod + def create_for_modification(cls, **kwargs) -> "QlikStream": + return cls.updater(**kwargs) + # ========================================================================= # Optimized Serialization Methods (override Asset base class) # ========================================================================= diff --git a/tests_v9/integration/qlik_asset_test.py b/tests_v9/integration/qlik_asset_test.py new file mode 100644 index 000000000..41791ddcd --- /dev/null +++ b/tests_v9/integration/qlik_asset_test.py @@ -0,0 +1,288 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Integration tests for the Qlik Sense creators in pyatlan_v9 (FND-3857). + +Creates a Qlik Sense connection, then every Qlik type through its creator, parents +before children, the way atlan-qlik-sense-app links them: a space and a stream, an +app in each, a sheet and a chart under the space's app, a dataset in the space, and +a column under each of the sheet, chart and dataset. Each asset is retrieved by guid +and checked for its exact qualifiedName and parent relationship, then purged. +""" + +from typing import Generator + +import pytest + +from pyatlan_v9.client.atlan import AtlanClient +from pyatlan_v9.model.assets import ( + Connection, + QlikApp, + QlikChart, + QlikColumn, + QlikDataset, + QlikSheet, + QlikSpace, + QlikStream, +) +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("QLIK") + +SPACE_ID = f"sp-{MODULE_NAME}" +STREAM_ID = f"st-{MODULE_NAME}" +APP_ID = f"ap-{MODULE_NAME}" +STREAM_APP_ID = f"sa-{MODULE_NAME}" +SHEET_ID = f"sh-{MODULE_NAME}" +CHART_ID = f"ch-{MODULE_NAME}" +DATASET_ID = f"ds-{MODULE_NAME}" +FIELD = f"field_{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.QLIKSENSE + ) + yield result + delete_asset(client, guid=result.guid, asset_type=Connection) + + +@pytest.fixture(scope="module") +def space( + client: AtlanClient, connection: Connection +) -> Generator[QlikSpace, None, None]: + to_create = QlikSpace.creator( + name=f"space_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + qlik_id=SPACE_ID, + ) + result = _save(client, to_create, QlikSpace) + yield result + delete_asset(client, guid=result.guid, asset_type=QlikSpace) + + +@pytest.fixture(scope="module") +def stream( + client: AtlanClient, connection: Connection +) -> Generator[QlikStream, None, None]: + to_create = QlikStream.creator( + name=f"stream_{MODULE_NAME}", + connection_qualified_name=connection.qualified_name, + qlik_id=STREAM_ID, + ) + result = _save(client, to_create, QlikStream) + yield result + delete_asset(client, guid=result.guid, asset_type=QlikStream) + + +@pytest.fixture(scope="module") +def app(client: AtlanClient, space: QlikSpace) -> Generator[QlikApp, None, None]: + to_create = QlikApp.creator( + name=f"app_{MODULE_NAME}", + qlik_id=APP_ID, + space_qualified_name=space.qualified_name, + ) + result = _save(client, to_create, QlikApp) + yield result + delete_asset(client, guid=result.guid, asset_type=QlikApp) + + +@pytest.fixture(scope="module") +def stream_app( + client: AtlanClient, stream: QlikStream +) -> Generator[QlikApp, None, None]: + to_create = QlikApp.creator( + name=f"stream_app_{MODULE_NAME}", + qlik_id=STREAM_APP_ID, + space_qualified_name=stream.qualified_name, + ) + result = _save(client, to_create, QlikApp) + yield result + delete_asset(client, guid=result.guid, asset_type=QlikApp) + + +@pytest.fixture(scope="module") +def sheet( + client: AtlanClient, app: QlikApp, space: QlikSpace +) -> Generator[QlikSheet, None, None]: + to_create = QlikSheet.creator( + name=f"sheet_{MODULE_NAME}", + qlik_id=SHEET_ID, + app_qualified_name=app.qualified_name, + space_qualified_name=space.qualified_name, + ) + result = _save(client, to_create, QlikSheet) + yield result + delete_asset(client, guid=result.guid, asset_type=QlikSheet) + + +@pytest.fixture(scope="module") +def chart( + client: AtlanClient, sheet: QlikSheet, space: QlikSpace +) -> Generator[QlikChart, None, None]: + to_create = QlikChart.creator( + name=f"chart_{MODULE_NAME}", + qlik_id=CHART_ID, + sheet_qualified_name=sheet.qualified_name, + space_qualified_name=space.qualified_name, + ) + result = _save(client, to_create, QlikChart) + yield result + delete_asset(client, guid=result.guid, asset_type=QlikChart) + + +@pytest.fixture(scope="module") +def dataset( + client: AtlanClient, space: QlikSpace +) -> Generator[QlikDataset, None, None]: + to_create = QlikDataset.creator( + name=f"dataset_{MODULE_NAME}", + qlik_id=DATASET_ID, + space_qualified_name=space.qualified_name, + ) + result = _save(client, to_create, QlikDataset) + yield result + delete_asset(client, guid=result.guid, asset_type=QlikDataset) + + +def _column(client: AtlanClient, parent_qualified_name: str, space_qualified_name): + to_create = QlikColumn.creator( + name=FIELD, + parent_qualified_name=parent_qualified_name, + space_qualified_name=space_qualified_name, + ) + return to_create, _save(client, to_create, QlikColumn) + + +@pytest.fixture(scope="module") +def sheet_column(client: AtlanClient, sheet: QlikSheet, space: QlikSpace): + to_create, result = _column(client, sheet.qualified_name, space.qualified_name) + yield to_create, result + delete_asset(client, guid=result.guid, asset_type=QlikColumn) + + +@pytest.fixture(scope="module") +def chart_column(client: AtlanClient, chart: QlikChart, space: QlikSpace): + to_create, result = _column(client, chart.qualified_name, space.qualified_name) + yield to_create, result + delete_asset(client, guid=result.guid, asset_type=QlikColumn) + + +@pytest.fixture(scope="module") +def dataset_column(client: AtlanClient, dataset: QlikDataset, space: QlikSpace): + to_create, result = _column(client, dataset.qualified_name, space.qualified_name) + yield to_create, result + delete_asset(client, guid=result.guid, asset_type=QlikColumn) + + +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.QLIKSENSE + assert retrieved.connection_qualified_name == connection.qualified_name + return retrieved + + +def test_space(client, connection, space): + retrieved = _assert_retrieved( + client, space, QlikSpace, connection, f"space/{SPACE_ID}" + ) + assert retrieved.qlik_id == SPACE_ID + + +def test_stream(client, connection, stream): + retrieved = _assert_retrieved( + client, stream, QlikStream, connection, f"stream/{STREAM_ID}" + ) + assert retrieved.qlik_id == STREAM_ID + + +def test_app(client, connection, space, app): + retrieved = _assert_retrieved(client, app, QlikApp, connection, f"app/{APP_ID}") + assert retrieved.qlik_space.guid == space.guid + assert retrieved.qlik_space_id == SPACE_ID + assert retrieved.qlik_space_qualified_name == space.qualified_name + + +def test_app_in_stream(client, connection, stream, stream_app): + retrieved = _assert_retrieved( + client, stream_app, QlikApp, connection, f"app/{STREAM_APP_ID}" + ) + # The app links to the stream itself, not a space with the stream's name. + assert retrieved.qlik_space.guid == stream.guid + assert retrieved.qlik_space_qualified_name == stream.qualified_name + retrieved_stream = client.asset.get_by_guid( + stream.guid, asset_type=QlikStream, ignore_relationships=False + ) + assert stream_app.guid in {app.guid for app in retrieved_stream.qlik_apps} + + +def test_sheet(client, connection, space, app, sheet): + retrieved = _assert_retrieved( + client, sheet, QlikSheet, connection, f"app/{APP_ID}/sheet/{SHEET_ID}" + ) + assert retrieved.qlik_app.guid == app.guid + assert retrieved.qlik_app_qualified_name == app.qualified_name + assert retrieved.qlik_space_qualified_name == space.qualified_name + + +def test_chart(client, connection, app, sheet, chart): + # A chart's qualifiedName hangs off its app, not its sheet. + retrieved = _assert_retrieved( + client, chart, QlikChart, connection, f"app/{APP_ID}/chart/{CHART_ID}" + ) + assert retrieved.qlik_sheet.guid == sheet.guid + assert retrieved.qlik_app_qualified_name == app.qualified_name + + +def test_dataset(client, connection, space, dataset): + retrieved = _assert_retrieved( + client, dataset, QlikDataset, connection, f"dataset/{DATASET_ID}" + ) + assert retrieved.qlik_space.guid == space.guid + + +@pytest.mark.parametrize( + "column_fixture, parent_fixture, relationship, parent_suffix", + [ + ("sheet_column", "sheet", "qlik_sheet", f"app/{APP_ID}/sheet/{SHEET_ID}"), + ("chart_column", "chart", "qlik_chart", f"app/{APP_ID}/chart/{CHART_ID}"), + ("dataset_column", "dataset", "qlik_dataset", f"dataset/{DATASET_ID}"), + ], + ids=["sheet", "chart", "dataset"], +) +def test_column( + request, + client, + connection, + column_fixture, + parent_fixture, + relationship, + parent_suffix, +): + to_create, column = request.getfixturevalue(column_fixture) + parent = request.getfixturevalue(parent_fixture) + + retrieved = _assert_retrieved( + client, + column, + QlikColumn, + connection, + to_create.qualified_name[len(connection.qualified_name) + 1 :], + ) + assert retrieved.qualified_name.startswith( + f"{connection.qualified_name}/{parent_suffix}/column/" + ) + assert getattr(retrieved, relationship).guid == parent.guid + assert retrieved.qlik_column_name == FIELD + assert retrieved.qlik_parent_qualified_name == parent.qualified_name diff --git a/tests_v9/unit/model/qlik_app_test.py b/tests_v9/unit/model/qlik_app_test.py new file mode 100644 index 000000000..86cd725b0 --- /dev/null +++ b/tests_v9/unit/model/qlik_app_test.py @@ -0,0 +1,98 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for QlikApp.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-qlik-sense-app emits today +(app/transformers/qlik_sense_cloud.py, _transform_app), so a creator change that +would re-key existing Qlik assets fails here. +""" + +import json + +import pytest + +from pyatlan_v9.model import QlikApp + +CONNECTION_QN = "default/qlik-sense/1712345678" +SPACE_QN = "default/qlik-sense/1712345678/space/sp01" +STREAM_QN = "default/qlik-sense/1712345678/stream/st01" +NAME = "Sales" +EXPECTED_QN = "default/qlik-sense/1712345678/app/a1b2" + + +def _creator(**overrides): + args = dict(name=NAME, qlik_id="a1b2", space_qualified_name=SPACE_QN) + args.update(overrides) + return QlikApp.creator(**args) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_id == "a1b2" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert sut.qlik_space_id == "sp01" + assert sut.qlik_space_qualified_name == SPACE_QN + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_create_alias(): + sut = QlikApp.create(name=NAME, qlik_id="a1b2", space_qualified_name=SPACE_QN) + + assert sut.qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize( + "space_qn, space_type, space_id", + [(SPACE_QN, "QlikSpace", "sp01"), (STREAM_QN, "QlikStream", "st01")], +) +def test_creator_sets_space_or_stream_parent(space_qn, space_type, space_id): + sut = _creator(space_qualified_name=space_qn) + + assert sut.qualified_name == EXPECTED_QN + assert sut.qlik_space_id == space_id + assert sut.qlik_space_qualified_name == space_qn + assert sut.qlik_space.type_name == space_type + assert sut.qlik_space.unique_attributes == {"qualifiedName": space_qn} + + +def test_app_in_stream_serializes_stream_relationship(serde): + data = json.loads( + _creator(space_qualified_name=STREAM_QN).to_json(nested=True, serde=serde) + ) + + space = data["relationshipAttributes"]["qlikSpace"] + assert space["typeName"] == "QlikStream" + assert space["uniqueAttributes"] == {"qualifiedName": STREAM_QN} + + +@pytest.mark.parametrize("key", ["name", "qlik_id", "space_qualified_name"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_creator_rejects_blank_inputs(key, blank): + with pytest.raises(ValueError, match=key): + _creator(**{key: blank}) + + +@pytest.mark.parametrize("space_qn", [CONNECTION_QN, EXPECTED_QN]) +def test_creator_rejects_wrong_parent(space_qn): + with pytest.raises(ValueError, match="space_qualified_name must end in"): + _creator(space_qualified_name=space_qn) + + +def test_updater(): + sut = QlikApp.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/qlik_chart_test.py b/tests_v9/unit/model/qlik_chart_test.py new file mode 100644 index 000000000..e4bb2e7d6 --- /dev/null +++ b/tests_v9/unit/model/qlik_chart_test.py @@ -0,0 +1,97 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for QlikChart.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-qlik-sense-app emits today +(app/transformers/qlik_sense_cloud.py, _transform_chart). A chart's qualifiedName +hangs off its app, not its sheet, so a creator change that would re-key existing +Qlik assets fails here. +""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import QlikChart + +CONNECTION_QN = "default/qlik-sense/1712345678" +STREAM_QN = "default/qlik-sense/1712345678/stream/st01" +APP_QN = "default/qlik-sense/1712345678/app/a1b2" +SHEET_QN = "default/qlik-sense/1712345678/app/a1b2/sheet/sh01" +NAME = "Revenue by region" +EXPECTED_QN = "default/qlik-sense/1712345678/app/a1b2/chart/ch01" + + +def _creator(**overrides): + args = dict( + name=NAME, + qlik_id="ch01", + sheet_qualified_name=SHEET_QN, + space_qualified_name=STREAM_QN, + ) + args.update(overrides) + return QlikChart.creator(**args) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_id == "ch01" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert sut.qlik_app_id == "a1b2" + assert sut.qlik_app_qualified_name == APP_QN + assert sut.qlik_space_id == "st01" + assert sut.qlik_space_qualified_name == STREAM_QN + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_create_alias(): + sut = QlikChart.create(name=NAME, qlik_id="ch01", sheet_qualified_name=SHEET_QN) + + assert sut.qualified_name == EXPECTED_QN + + +def test_creator_sets_parent(): + assert _creator().qlik_sheet.unique_attributes == {"qualifiedName": SHEET_QN} + + +def test_space_is_optional(): + sut = _creator(space_qualified_name=None) + + assert sut.qualified_name == EXPECTED_QN + assert sut.qlik_space_id is UNSET + assert sut.qlik_space_qualified_name is UNSET + + +@pytest.mark.parametrize("key", ["name", "qlik_id", "sheet_qualified_name"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_creator_rejects_blank_inputs(key, blank): + with pytest.raises(ValueError, match=key): + _creator(**{key: blank}) + + +@pytest.mark.parametrize( + "sheet_qn", [APP_QN, EXPECTED_QN, f"{CONNECTION_QN}/dataset/ds01"] +) +def test_creator_rejects_wrong_parent(sheet_qn): + with pytest.raises(ValueError, match="sheet_qualified_name must end in"): + _creator(sheet_qualified_name=sheet_qn) + + +def test_updater(): + sut = QlikChart.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_sheet is UNSET diff --git a/tests_v9/unit/model/qlik_column_test.py b/tests_v9/unit/model/qlik_column_test.py new file mode 100644 index 000000000..d4f9b5b62 --- /dev/null +++ b/tests_v9/unit/model/qlik_column_test.py @@ -0,0 +1,149 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for QlikColumn.creator in pyatlan_v9. + +The expected qualifiedNames are the exact strings atlan-qlik-sense-app emits today: +`{connection}/{id}`, where `id` comes from generate_field_id in +app/processors/qlik_lineage.py, i.e. `{parent path}/column/{md5}` with the md5 +taken over `"{parent path}/column/{field name}"`. A creator change that would +re-key existing Qlik columns fails here. +""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import QlikColumn + +CONNECTION_QN = "default/qlik-sense/1712345678" +SPACE_QN = "default/qlik-sense/1712345678/space/sp01" +APP_QN = "default/qlik-sense/1712345678/app/a1b2" +SHEET_QN = "default/qlik-sense/1712345678/app/a1b2/sheet/sh01" +CHART_QN = "default/qlik-sense/1712345678/app/a1b2/chart/ch01" +DATASET_QN = "default/qlik-sense/1712345678/dataset/ds01" +NAME = "Region" +PARENTS = ["qlik_sheet", "qlik_chart", "qlik_dataset"] + + +@pytest.mark.parametrize( + "parent_qn, relationship, expected_qn", + [ + ( + SHEET_QN, + "qlik_sheet", + "default/qlik-sense/1712345678/app/a1b2/sheet/sh01" + "/column/72e9c09388e68cdceee575c42e2ee669", + ), + ( + CHART_QN, + "qlik_chart", + "default/qlik-sense/1712345678/app/a1b2/chart/ch01" + "/column/9361398dc447eebbec36cb448d7cc064", + ), + ( + DATASET_QN, + "qlik_dataset", + "default/qlik-sense/1712345678/dataset/ds01" + "/column/8899bf1ab296301e5c10c37600324580", + ), + ], +) +def test_creator_matches_app_qualified_name(parent_qn, relationship, expected_qn): + sut = QlikColumn.creator( + name=NAME, parent_qualified_name=parent_qn, space_qualified_name=SPACE_QN + ) + + assert sut.qualified_name == expected_qn + assert sut.name == NAME + assert sut.qlik_column_name == NAME + assert sut.qlik_parent_qualified_name == parent_qn + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert sut.qlik_space_id == "sp01" + assert sut.qlik_space_qualified_name == SPACE_QN + assert getattr(sut, relationship).unique_attributes == {"qualifiedName": parent_qn} + for other in set(PARENTS) - {relationship}: + assert getattr(sut, other) is UNSET + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + # Any one of the three parents satisfies creation validation; a dataset + # column has no app, so the app qualifiedName must not be required. + sut.validate(for_creation=True) + + +def test_create_alias(): + sut = QlikColumn.create(name=NAME, parent_qualified_name=DATASET_QN) + + assert sut.qualified_name.startswith(f"{DATASET_QN}/column/") + + +def test_space_is_optional(): + sut = QlikColumn.creator(name=NAME, parent_qualified_name=DATASET_QN) + + assert sut.qlik_space_id is UNSET + assert sut.qlik_space_qualified_name is UNSET + + +def test_same_field_under_different_parents_gets_different_hashes(): + sheet_col = QlikColumn.creator(name=NAME, parent_qualified_name=SHEET_QN) + chart_col = QlikColumn.creator(name=NAME, parent_qualified_name=CHART_QN) + + assert ( + sheet_col.qualified_name.rsplit("/", 1)[1] + != (chart_col.qualified_name.rsplit("/", 1)[1]) + ) + + +def test_column_without_parent_fails_creation_validation(): + sut = QlikColumn( + name=NAME, + qualified_name=f"{DATASET_QN}/column/abc", + connection_qualified_name=CONNECTION_QN, + qlik_space_qualified_name=SPACE_QN, + ) + + with pytest.raises(ValueError, match="one of qlik_sheet, qlik_chart, qlik_dataset"): + sut.validate(for_creation=True) + + +@pytest.mark.parametrize("key", ["name", "parent_qualified_name"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_creator_rejects_blank_inputs(key, blank): + args = dict(name=NAME, parent_qualified_name=SHEET_QN) + args[key] = blank + + with pytest.raises(ValueError, match=key): + QlikColumn.creator(**args) + + +@pytest.mark.parametrize( + "overrides, message", + [ + ({"parent_qualified_name": APP_QN}, "parent_qualified_name must end in"), + ({"parent_qualified_name": SPACE_QN}, "parent_qualified_name must end in"), + ({"parent_qualified_name": CONNECTION_QN}, "parent_qualified_name must end in"), + ({"space_qualified_name": DATASET_QN}, "space_qualified_name must end in"), + ], +) +def test_creator_rejects_wrong_parent(overrides, message): + args = dict(name=NAME, parent_qualified_name=SHEET_QN) + args.update(overrides) + + with pytest.raises(ValueError, match=message): + QlikColumn.creator(**args) + + +def test_updater(): + qn = f"{DATASET_QN}/column/8899bf1ab296301e5c10c37600324580" + sut = QlikColumn.updater(qualified_name=qn, name=NAME) + + assert sut.qualified_name == qn + assert sut.name == NAME + + +def test_trim_to_required(): + sut = QlikColumn.creator(name=NAME, parent_qualified_name=DATASET_QN) + trimmed = sut.trim_to_required() + + assert trimmed.qualified_name == sut.qualified_name + assert trimmed.name == NAME + assert trimmed.qlik_dataset is UNSET diff --git a/tests_v9/unit/model/qlik_dataset_test.py b/tests_v9/unit/model/qlik_dataset_test.py new file mode 100644 index 000000000..b849e662a --- /dev/null +++ b/tests_v9/unit/model/qlik_dataset_test.py @@ -0,0 +1,89 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for QlikDataset.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-qlik-sense-app emits today +(app/transformers/qlik_sense_cloud.py, _transform_dataset), so a creator change that +would re-key existing Qlik assets fails here. +""" + +import pytest + +from pyatlan_v9.model import QlikDataset + +CONNECTION_QN = "default/qlik-sense/1712345678" +SPACE_QN = "default/qlik-sense/1712345678/space/sp01" +STREAM_QN = "default/qlik-sense/1712345678/stream/st01" +NAME = "orders.qvd" +EXPECTED_QN = "default/qlik-sense/1712345678/dataset/ds01" + + +def _creator(**overrides): + args = dict(name=NAME, qlik_id="ds01", space_qualified_name=SPACE_QN) + args.update(overrides) + return QlikDataset.creator(**args) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_id == "ds01" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert sut.qlik_space_id == "sp01" + assert sut.qlik_space_qualified_name == SPACE_QN + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_create_alias(): + sut = QlikDataset.create(name=NAME, qlik_id="ds01", space_qualified_name=SPACE_QN) + + assert sut.qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize( + "space_qn, space_type", [(SPACE_QN, "QlikSpace"), (STREAM_QN, "QlikStream")] +) +def test_creator_sets_space_or_stream_parent(space_qn, space_type): + sut = _creator(space_qualified_name=space_qn) + + assert sut.qlik_space.type_name == space_type + assert sut.qlik_space.unique_attributes == {"qualifiedName": space_qn} + + +def test_personal_space(): + # The app uses the pseudo-space "personal" when a dataset has no spaceId. + sut = _creator(space_qualified_name=f"{CONNECTION_QN}/space/personal") + + assert sut.qualified_name == EXPECTED_QN + assert sut.qlik_space_id == "personal" + + +@pytest.mark.parametrize("key", ["name", "qlik_id", "space_qualified_name"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_creator_rejects_blank_inputs(key, blank): + with pytest.raises(ValueError, match=key): + _creator(**{key: blank}) + + +def test_creator_rejects_wrong_parent(): + with pytest.raises(ValueError, match="space_qualified_name must end in"): + _creator(space_qualified_name=CONNECTION_QN) + + +def test_updater(): + sut = QlikDataset.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/qlik_sheet_test.py b/tests_v9/unit/model/qlik_sheet_test.py new file mode 100644 index 000000000..7e46bcc3e --- /dev/null +++ b/tests_v9/unit/model/qlik_sheet_test.py @@ -0,0 +1,100 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for QlikSheet.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-qlik-sense-app emits today +(app/transformers/qlik_sense_cloud.py, _transform_sheet), so a creator change that +would re-key existing Qlik assets fails here. +""" + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import QlikSheet + +CONNECTION_QN = "default/qlik-sense/1712345678" +SPACE_QN = "default/qlik-sense/1712345678/space/sp01" +APP_QN = "default/qlik-sense/1712345678/app/a1b2" +NAME = "Overview" +EXPECTED_QN = "default/qlik-sense/1712345678/app/a1b2/sheet/sh01" + + +def _creator(**overrides): + args = dict( + name=NAME, + qlik_id="sh01", + app_qualified_name=APP_QN, + space_qualified_name=SPACE_QN, + ) + args.update(overrides) + return QlikSheet.creator(**args) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_id == "sh01" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert sut.qlik_app_id == "a1b2" + assert sut.qlik_app_qualified_name == APP_QN + assert sut.qlik_space_id == "sp01" + assert sut.qlik_space_qualified_name == SPACE_QN + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_create_alias(): + sut = QlikSheet.create(name=NAME, qlik_id="sh01", app_qualified_name=APP_QN) + + assert sut.qualified_name == EXPECTED_QN + + +def test_creator_sets_parent(): + assert _creator().qlik_app.unique_attributes == {"qualifiedName": APP_QN} + + +def test_space_is_optional(): + sut = _creator(space_qualified_name=None) + + assert sut.qualified_name == EXPECTED_QN + assert sut.qlik_space_id is UNSET + assert sut.qlik_space_qualified_name is UNSET + + +@pytest.mark.parametrize("key", ["name", "qlik_id", "app_qualified_name"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_creator_rejects_blank_inputs(key, blank): + with pytest.raises(ValueError, match=key): + _creator(**{key: blank}) + + +@pytest.mark.parametrize( + "overrides, message", + [ + ({"app_qualified_name": SPACE_QN}, "app_qualified_name must end in"), + ({"app_qualified_name": CONNECTION_QN}, "app_qualified_name must end in"), + ({"space_qualified_name": APP_QN}, "space_qualified_name must end in"), + ], +) +def test_creator_rejects_wrong_parent(overrides, message): + with pytest.raises(ValueError, match=message): + _creator(**overrides) + + +def test_updater(): + sut = QlikSheet.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_app is UNSET diff --git a/tests_v9/unit/model/qlik_space_test.py b/tests_v9/unit/model/qlik_space_test.py new file mode 100644 index 000000000..3b3ff8d11 --- /dev/null +++ b/tests_v9/unit/model/qlik_space_test.py @@ -0,0 +1,70 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for QlikSpace.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-qlik-sense-app emits today +(app/transformers/qlik_sense_cloud.py, _transform_space), so a creator change that +would re-key existing Qlik assets fails here. +""" + +import pytest + +from pyatlan_v9.model import QlikSpace + +CONNECTION_QN = "default/qlik-sense/1712345678" +NAME = "Finance" +EXPECTED_QN = "default/qlik-sense/1712345678/space/sp01" + + +def _creator(**overrides): + args = dict(name=NAME, connection_qualified_name=CONNECTION_QN, qlik_id="sp01") + args.update(overrides) + return QlikSpace.creator(**args) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_id == "sp01" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_create_alias(): + sut = QlikSpace.create( + name=NAME, connection_qualified_name=CONNECTION_QN, qlik_id="sp01" + ) + + assert sut.qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize("key", ["name", "connection_qualified_name", "qlik_id"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_creator_rejects_blank_inputs(key, blank): + with pytest.raises(ValueError, match=key): + _creator(**{key: blank}) + + +def test_stream_qualified_name_does_not_match_space_pattern(): + assert not QlikSpace._QUALIFIED_NAME_PATTERN.match( + "default/qlik-sense/1712345678/stream/st01" + ) + + +def test_updater(): + sut = QlikSpace.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME diff --git a/tests_v9/unit/model/qlik_stream_test.py b/tests_v9/unit/model/qlik_stream_test.py new file mode 100644 index 000000000..23b7c9349 --- /dev/null +++ b/tests_v9/unit/model/qlik_stream_test.py @@ -0,0 +1,64 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for QlikStream.creator in pyatlan_v9. + +EXPECTED_QN is the exact string atlan-qlik-sense-app emits today for a Qlik Sense +Enterprise on Windows stream (_transform_space with isStream), so a creator change +that would re-key existing Qlik assets fails here. +""" + +import pytest + +from pyatlan_v9.model import QlikStream + +CONNECTION_QN = "default/qlik-sense/1712345678" +NAME = "Everyone" +EXPECTED_QN = "default/qlik-sense/1712345678/stream/st01" + + +def _creator(**overrides): + args = dict(name=NAME, connection_qualified_name=CONNECTION_QN, qlik_id="st01") + args.update(overrides) + return QlikStream.creator(**args) + + +def test_creator_matches_app_qualified_name(): + sut = _creator() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + assert sut.qlik_id == "st01" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert sut._QUALIFIED_NAME_PATTERN.match(sut.qualified_name) + sut.validate(for_creation=True) + + +def test_create_alias(): + sut = QlikStream.create( + name=NAME, connection_qualified_name=CONNECTION_QN, qlik_id="st01" + ) + + assert sut.qualified_name == EXPECTED_QN + + +@pytest.mark.parametrize("key", ["name", "connection_qualified_name", "qlik_id"]) +@pytest.mark.parametrize("blank", [None, "", " "]) +def test_creator_rejects_blank_inputs(key, blank): + with pytest.raises(ValueError, match=key): + _creator(**{key: blank}) + + +def test_updater(): + sut = QlikStream.updater(qualified_name=EXPECTED_QN, name=NAME) + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME + + +def test_trim_to_required(): + sut = _creator().trim_to_required() + + assert sut.qualified_name == EXPECTED_QN + assert sut.name == NAME