From 2d0256c1c6a10e1a27e5cec1264b2a8dd43ef2fc Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Wed, 14 Oct 2026 16:09:32 +0100 Subject: [PATCH 1/3] FND-3857: add pyatlan_v9 creators for the 7 Qlik Sense types Adds hand-written creator overlays for QlikSpace, QlikStream, QlikApp, QlikDataset, QlikSheet, QlikChart and QlikColumn, and regenerates those classes from the paired atlanhq/models change that adds a qualifiedNameFormula per Qlik type. Each creator emits the exact qualifiedName atlan-qlik-sense-app emits today. Child creators take the parent's qualifiedName and derive the connection, app id and space id from it: - App / Dataset take space_qualified_name. Its /space/ or /stream/ segment decides whether qlik_space points at a QlikSpace or a QlikStream, so the creator does not guess from a deployment type. - Sheet takes app_qualified_name; Chart takes sheet_qualified_name and builds {app}/chart/{id} (the chart qn hangs off the app, not the sheet). Both take an optional space_qualified_name for the denormalised space. - Column takes parent_qualified_name (sheet, chart or dataset) and builds {parent}/column/{md5("{parent path}/column/{name}")}, matching the app's generate_field_id. The parent type is read from the qualifiedName. Blank inputs and parent qualifiedNames of the wrong shape raise ValueError. tests_v9/unit/model/qlik_test.py asserts each literal app qualifiedName. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WG31Pc155G4jqZUzC2FEet --- pyatlan_v9/model/assets/_overlays/qlik_app.py | 52 ++++ .../model/assets/_overlays/qlik_chart.py | 58 ++++ .../model/assets/_overlays/qlik_column.py | 70 +++++ .../model/assets/_overlays/qlik_dataset.py | 52 ++++ .../model/assets/_overlays/qlik_sheet.py | 55 ++++ .../model/assets/_overlays/qlik_space.py | 33 +++ .../model/assets/_overlays/qlik_stream.py | 33 +++ pyatlan_v9/model/assets/qlik_app.py | 54 +++- pyatlan_v9/model/assets/qlik_chart.py | 66 ++++- pyatlan_v9/model/assets/qlik_column.py | 74 ++++- pyatlan_v9/model/assets/qlik_dataset.py | 54 +++- pyatlan_v9/model/assets/qlik_sheet.py | 59 +++- pyatlan_v9/model/assets/qlik_space.py | 41 +++ pyatlan_v9/model/assets/qlik_stream.py | 57 +++- tests_v9/unit/model/qlik_test.py | 264 ++++++++++++++++++ 15 files changed, 1003 insertions(+), 19 deletions(-) create mode 100644 pyatlan_v9/model/assets/_overlays/qlik_app.py create mode 100644 pyatlan_v9/model/assets/_overlays/qlik_chart.py create mode 100644 pyatlan_v9/model/assets/_overlays/qlik_column.py create mode 100644 pyatlan_v9/model/assets/_overlays/qlik_dataset.py create mode 100644 pyatlan_v9/model/assets/_overlays/qlik_sheet.py create mode 100644 pyatlan_v9/model/assets/_overlays/qlik_space.py create mode 100644 pyatlan_v9/model/assets/_overlays/qlik_stream.py create mode 100644 tests_v9/unit/model/qlik_test.py 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..20691051b --- /dev/null +++ b/pyatlan_v9/model/assets/_overlays/qlik_column.py @@ -0,0 +1,70 @@ +# 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, + ) -> "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. + """ + 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 + ) + 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..9508da2fb 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: """ @@ -365,6 +365,74 @@ 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, + ) -> "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. + """ + 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 + ) + 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/unit/model/qlik_test.py b/tests_v9/unit/model/qlik_test.py new file mode 100644 index 000000000..da312abc5 --- /dev/null +++ b/tests_v9/unit/model/qlik_test.py @@ -0,0 +1,264 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. + +"""Unit tests for the Qlik Sense creators in pyatlan_v9. + +The expected qualifiedNames are the exact strings atlan-qlik-sense-app emits today +(app/transformers/qlik_sense_cloud.py and the generate_field_id column ids in +app/processors/qlik_lineage.py), so a creator change that would re-key existing +Qlik assets fails here. +""" + +import json + +import pytest +from msgspec import UNSET + +from pyatlan_v9.model import ( + QlikApp, + QlikChart, + QlikColumn, + QlikDataset, + QlikSheet, + QlikSpace, + QlikStream, +) + +CONNECTION_QN = "default/qlik-sense/1712345678" +SPACE_QN = "default/qlik-sense/1712345678/space/sp01" +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" +CHART_QN = "default/qlik-sense/1712345678/app/a1b2/chart/ch01" +DATASET_QN = "default/qlik-sense/1712345678/dataset/ds01" +NAME = "Quarterly Sales" +FIELD = "Region" + + +@pytest.mark.parametrize( + "asset_type, seg", [(QlikSpace, "space"), (QlikStream, "stream")] +) +def test_space_and_stream_creator_matches_app_qualified_name(asset_type, seg): + sut = asset_type.creator( + name=NAME, connection_qualified_name=CONNECTION_QN, qlik_id="sp01" + ) + + assert sut.qualified_name == f"default/qlik-sense/1712345678/{seg}/sp01" + assert sut.name == NAME + assert sut.qlik_id == "sp01" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + sut.validate() + + +@pytest.mark.parametrize( + "asset_type, expected_qn", + [ + (QlikApp, "default/qlik-sense/1712345678/app/a1b2"), + (QlikDataset, "default/qlik-sense/1712345678/dataset/a1b2"), + ], +) +@pytest.mark.parametrize( + "space_qn, space_type, space_id", + [(SPACE_QN, "QlikSpace", "sp01"), (STREAM_QN, "QlikStream", "st01")], +) +def test_app_and_dataset_creator_matches_app_qualified_name( + asset_type, expected_qn, space_qn, space_type, space_id +): + sut = asset_type.creator(name=NAME, qlik_id="a1b2", space_qualified_name=space_qn) + + assert sut.qualified_name == expected_qn + assert sut.qlik_id == "a1b2" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + 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} + sut.validate(for_creation=True) + + +def test_app_in_stream_serializes_stream_relationship(serde): + sut = QlikApp.creator(name=NAME, qlik_id="a1b2", space_qualified_name=STREAM_QN) + + data = json.loads(sut.to_json(nested=True, serde=serde)) + + space = data["relationshipAttributes"]["qlikSpace"] + assert space["typeName"] == "QlikStream" + assert space["uniqueAttributes"] == {"qualifiedName": STREAM_QN} + + +def test_sheet_creator_matches_app_qualified_name(): + sut = QlikSheet.creator( + name=NAME, + qlik_id="sh01", + app_qualified_name=APP_QN, + space_qualified_name=SPACE_QN, + ) + + assert sut.qualified_name == "default/qlik-sense/1712345678/app/a1b2/sheet/sh01" + 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_app.unique_attributes == {"qualifiedName": APP_QN} + assert sut.qlik_space_id == "sp01" + assert sut.qlik_space_qualified_name == SPACE_QN + sut.validate(for_creation=True) + + +def test_chart_creator_matches_app_qualified_name(): + sut = QlikChart.creator( + name=NAME, + qlik_id="ch01", + sheet_qualified_name=SHEET_QN, + space_qualified_name=STREAM_QN, + ) + + # The chart hangs off the app, not the sheet. + assert sut.qualified_name == "default/qlik-sense/1712345678/app/a1b2/chart/ch01" + assert sut.qlik_id == "ch01" + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.qlik_app_id == "a1b2" + assert sut.qlik_app_qualified_name == APP_QN + assert sut.qlik_sheet.unique_attributes == {"qualifiedName": SHEET_QN} + assert sut.qlik_space_id == "st01" + assert sut.qlik_space_qualified_name == STREAM_QN + sut.validate(for_creation=True) + + +@pytest.mark.parametrize( + "asset_type, parent", + [ + (QlikSheet, {"app_qualified_name": APP_QN}), + (QlikChart, {"sheet_qualified_name": SHEET_QN}), + ], +) +def test_sheet_and_chart_space_is_optional(asset_type, parent): + sut = asset_type.creator(name=NAME, qlik_id="x1", **parent) + + assert sut.qlik_space_id is UNSET + assert sut.qlik_space_qualified_name is UNSET + + +@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_column_creator_matches_app_qualified_name( + parent_qn, relationship, expected_qn +): + sut = QlikColumn.creator(name=FIELD, parent_qualified_name=parent_qn) + + assert sut.qualified_name == expected_qn + assert sut.name == FIELD + assert sut.qlik_column_name == FIELD + assert sut.qlik_parent_qualified_name == parent_qn + assert sut.connection_qualified_name == CONNECTION_QN + assert sut.connector_name == "qlik-sense" + assert getattr(sut, relationship).unique_attributes == {"qualifiedName": parent_qn} + for other in {"qlik_sheet", "qlik_chart", "qlik_dataset"} - {relationship}: + assert not getattr(sut, other) + sut.validate() + + +@pytest.mark.parametrize( + "creator, kwargs, message", + [ + ( + QlikSpace.creator, + {"name": "", "connection_qualified_name": CONNECTION_QN, "qlik_id": "x"}, + "name", + ), + ( + QlikStream.creator, + {"name": NAME, "connection_qualified_name": CONNECTION_QN, "qlik_id": ""}, + "qlik_id", + ), + ( + QlikApp.creator, + {"name": NAME, "qlik_id": "x", "space_qualified_name": None}, + "space_qualified_name", + ), + ( + QlikDataset.creator, + {"name": NAME, "qlik_id": " ", "space_qualified_name": SPACE_QN}, + "qlik_id", + ), + ( + QlikSheet.creator, + {"name": NAME, "qlik_id": "x", "app_qualified_name": ""}, + "app_qualified_name", + ), + ( + QlikChart.creator, + {"name": NAME, "qlik_id": "x", "sheet_qualified_name": None}, + "sheet_qualified_name", + ), + ( + QlikColumn.creator, + {"name": "", "parent_qualified_name": SHEET_QN}, + "name", + ), + ], +) +def test_creator_with_blank_input_raises(creator, kwargs, message): + with pytest.raises(ValueError, match=message): + creator(**kwargs) + + +@pytest.mark.parametrize( + "creator, kwargs", + [ + ( + QlikApp.creator, + {"name": NAME, "qlik_id": "x", "space_qualified_name": APP_QN}, + ), + ( + QlikDataset.creator, + {"name": NAME, "qlik_id": "x", "space_qualified_name": CONNECTION_QN}, + ), + ( + QlikSheet.creator, + {"name": NAME, "qlik_id": "x", "app_qualified_name": SPACE_QN}, + ), + ( + QlikSheet.creator, + { + "name": NAME, + "qlik_id": "x", + "app_qualified_name": APP_QN, + "space_qualified_name": DATASET_QN, + }, + ), + ( + QlikChart.creator, + {"name": NAME, "qlik_id": "x", "sheet_qualified_name": CHART_QN}, + ), + (QlikColumn.creator, {"name": FIELD, "parent_qualified_name": APP_QN}), + (QlikColumn.creator, {"name": FIELD, "parent_qualified_name": SPACE_QN}), + ], +) +def test_creator_with_wrong_parent_qualified_name_raises(creator, kwargs): + with pytest.raises(ValueError, match="must end in"): + creator(**kwargs) From 78c1b75d7ae651e46a5078ccf97ed687dfb95451 Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Thu, 15 Oct 2026 09:17:30 +0100 Subject: [PATCH 2/3] FND-3857: fix QlikColumn creation validation for chart and dataset parents The generated QlikColumn.validate(for_creation=True) always required qlik_sheet and qlik_app_qualified_name, so it rejected every chart- or dataset-parented column. Regenerated from the paired models change (sdk.parentTypes on Qlik Column + renderer common-ancestor fix), it now requires one of qlik_sheet / qlik_chart / qlik_dataset plus qlik_space_qualified_name, the one ancestor all three parents share. QlikColumn.creator gains an optional space_qualified_name, like the sheet and chart creators, to fill that denormalised space. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WG31Pc155G4jqZUzC2FEet --- .../model/assets/_overlays/qlik_column.py | 11 +++++++++ pyatlan_v9/model/assets/qlik_column.py | 23 +++++++++++++++---- tests_v9/unit/model/qlik_test.py | 22 ++++++++++++++++-- 3 files changed, 50 insertions(+), 6 deletions(-) diff --git a/pyatlan_v9/model/assets/_overlays/qlik_column.py b/pyatlan_v9/model/assets/_overlays/qlik_column.py index 20691051b..a68cc7073 100644 --- a/pyatlan_v9/model/assets/_overlays/qlik_column.py +++ b/pyatlan_v9/model/assets/_overlays/qlik_column.py @@ -8,6 +8,7 @@ def creator( *, name: str, parent_qualified_name: str, + space_qualified_name: Union[str, None] = None, ) -> "QlikColumn": """ Build a Qlik column under a sheet, chart or dataset. @@ -17,6 +18,8 @@ def creator( 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] @@ -59,6 +62,14 @@ def creator( 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 diff --git a/pyatlan_v9/model/assets/qlik_column.py b/pyatlan_v9/model/assets/qlik_column.py index 9508da2fb..89443065b 100644 --- a/pyatlan_v9/model/assets/qlik_column.py +++ b/pyatlan_v9/model/assets/qlik_column.py @@ -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}") @@ -372,6 +376,7 @@ def creator( *, name: str, parent_qualified_name: str, + space_qualified_name: Union[str, None] = None, ) -> "QlikColumn": """ Build a Qlik column under a sheet, chart or dataset. @@ -381,6 +386,8 @@ def creator( 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] @@ -423,6 +430,14 @@ def creator( 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 diff --git a/tests_v9/unit/model/qlik_test.py b/tests_v9/unit/model/qlik_test.py index da312abc5..9177c197b 100644 --- a/tests_v9/unit/model/qlik_test.py +++ b/tests_v9/unit/model/qlik_test.py @@ -168,7 +168,9 @@ def test_sheet_and_chart_space_is_optional(asset_type, parent): def test_column_creator_matches_app_qualified_name( parent_qn, relationship, expected_qn ): - sut = QlikColumn.creator(name=FIELD, parent_qualified_name=parent_qn) + sut = QlikColumn.creator( + name=FIELD, parent_qualified_name=parent_qn, space_qualified_name=SPACE_QN + ) assert sut.qualified_name == expected_qn assert sut.name == FIELD @@ -176,10 +178,26 @@ def test_column_creator_matches_app_qualified_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 {"qlik_sheet", "qlik_chart", "qlik_dataset"} - {relationship}: assert not getattr(sut, other) - sut.validate() + # 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_column_without_parent_fails_creation_validation(): + sut = QlikColumn( + name=FIELD, + 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( From b0e3eb886efec293f34f540e8705c0d7caf3eeb2 Mon Sep 17 00:00:00 2001 From: Christopher Grote Date: Mon, 19 Oct 2026 16:42:14 +0100 Subject: [PATCH 3/3] FND-3857: per-type Qlik unit tests and a Qlik integration test Splits tests_v9/unit/model/qlik_test.py into one file per type, in the layout the merged Fabric creators use: literal app qualifiedName, create alias, None / "" / " " for every required input, wrong-shape parents, updater and trim_to_required. Adds a personal-space dataset case and a check that the same field under a sheet and a chart hashes differently. tests_v9/integration/qlik_asset_test.py creates a Qlik Sense connection, a space and a stream, an app in each, a sheet and a chart under the space's app, a dataset, and a column under each of the sheet, chart and dataset, all through the creators. It retrieves each by guid and asserts the exact qualifiedName and parent relationship, then purges. 10 passed against a tenant. Also merges origin/main; regenerating the Qlik classes from the models branch on top of it is a no-op. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WG31Pc155G4jqZUzC2FEet --- tests_v9/integration/qlik_asset_test.py | 288 +++++++++++++++++++++++ tests_v9/unit/model/qlik_app_test.py | 98 ++++++++ tests_v9/unit/model/qlik_chart_test.py | 97 ++++++++ tests_v9/unit/model/qlik_column_test.py | 149 ++++++++++++ tests_v9/unit/model/qlik_dataset_test.py | 89 +++++++ tests_v9/unit/model/qlik_sheet_test.py | 100 ++++++++ tests_v9/unit/model/qlik_space_test.py | 70 ++++++ tests_v9/unit/model/qlik_stream_test.py | 64 +++++ tests_v9/unit/model/qlik_test.py | 282 ---------------------- 9 files changed, 955 insertions(+), 282 deletions(-) create mode 100644 tests_v9/integration/qlik_asset_test.py create mode 100644 tests_v9/unit/model/qlik_app_test.py create mode 100644 tests_v9/unit/model/qlik_chart_test.py create mode 100644 tests_v9/unit/model/qlik_column_test.py create mode 100644 tests_v9/unit/model/qlik_dataset_test.py create mode 100644 tests_v9/unit/model/qlik_sheet_test.py create mode 100644 tests_v9/unit/model/qlik_space_test.py create mode 100644 tests_v9/unit/model/qlik_stream_test.py delete mode 100644 tests_v9/unit/model/qlik_test.py 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 diff --git a/tests_v9/unit/model/qlik_test.py b/tests_v9/unit/model/qlik_test.py deleted file mode 100644 index 9177c197b..000000000 --- a/tests_v9/unit/model/qlik_test.py +++ /dev/null @@ -1,282 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# Copyright 2026 Atlan Pte. Ltd. - -"""Unit tests for the Qlik Sense creators in pyatlan_v9. - -The expected qualifiedNames are the exact strings atlan-qlik-sense-app emits today -(app/transformers/qlik_sense_cloud.py and the generate_field_id column ids in -app/processors/qlik_lineage.py), so a creator change that would re-key existing -Qlik assets fails here. -""" - -import json - -import pytest -from msgspec import UNSET - -from pyatlan_v9.model import ( - QlikApp, - QlikChart, - QlikColumn, - QlikDataset, - QlikSheet, - QlikSpace, - QlikStream, -) - -CONNECTION_QN = "default/qlik-sense/1712345678" -SPACE_QN = "default/qlik-sense/1712345678/space/sp01" -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" -CHART_QN = "default/qlik-sense/1712345678/app/a1b2/chart/ch01" -DATASET_QN = "default/qlik-sense/1712345678/dataset/ds01" -NAME = "Quarterly Sales" -FIELD = "Region" - - -@pytest.mark.parametrize( - "asset_type, seg", [(QlikSpace, "space"), (QlikStream, "stream")] -) -def test_space_and_stream_creator_matches_app_qualified_name(asset_type, seg): - sut = asset_type.creator( - name=NAME, connection_qualified_name=CONNECTION_QN, qlik_id="sp01" - ) - - assert sut.qualified_name == f"default/qlik-sense/1712345678/{seg}/sp01" - assert sut.name == NAME - assert sut.qlik_id == "sp01" - assert sut.connection_qualified_name == CONNECTION_QN - assert sut.connector_name == "qlik-sense" - sut.validate() - - -@pytest.mark.parametrize( - "asset_type, expected_qn", - [ - (QlikApp, "default/qlik-sense/1712345678/app/a1b2"), - (QlikDataset, "default/qlik-sense/1712345678/dataset/a1b2"), - ], -) -@pytest.mark.parametrize( - "space_qn, space_type, space_id", - [(SPACE_QN, "QlikSpace", "sp01"), (STREAM_QN, "QlikStream", "st01")], -) -def test_app_and_dataset_creator_matches_app_qualified_name( - asset_type, expected_qn, space_qn, space_type, space_id -): - sut = asset_type.creator(name=NAME, qlik_id="a1b2", space_qualified_name=space_qn) - - assert sut.qualified_name == expected_qn - assert sut.qlik_id == "a1b2" - assert sut.connection_qualified_name == CONNECTION_QN - assert sut.connector_name == "qlik-sense" - 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} - sut.validate(for_creation=True) - - -def test_app_in_stream_serializes_stream_relationship(serde): - sut = QlikApp.creator(name=NAME, qlik_id="a1b2", space_qualified_name=STREAM_QN) - - data = json.loads(sut.to_json(nested=True, serde=serde)) - - space = data["relationshipAttributes"]["qlikSpace"] - assert space["typeName"] == "QlikStream" - assert space["uniqueAttributes"] == {"qualifiedName": STREAM_QN} - - -def test_sheet_creator_matches_app_qualified_name(): - sut = QlikSheet.creator( - name=NAME, - qlik_id="sh01", - app_qualified_name=APP_QN, - space_qualified_name=SPACE_QN, - ) - - assert sut.qualified_name == "default/qlik-sense/1712345678/app/a1b2/sheet/sh01" - 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_app.unique_attributes == {"qualifiedName": APP_QN} - assert sut.qlik_space_id == "sp01" - assert sut.qlik_space_qualified_name == SPACE_QN - sut.validate(for_creation=True) - - -def test_chart_creator_matches_app_qualified_name(): - sut = QlikChart.creator( - name=NAME, - qlik_id="ch01", - sheet_qualified_name=SHEET_QN, - space_qualified_name=STREAM_QN, - ) - - # The chart hangs off the app, not the sheet. - assert sut.qualified_name == "default/qlik-sense/1712345678/app/a1b2/chart/ch01" - assert sut.qlik_id == "ch01" - assert sut.connection_qualified_name == CONNECTION_QN - assert sut.qlik_app_id == "a1b2" - assert sut.qlik_app_qualified_name == APP_QN - assert sut.qlik_sheet.unique_attributes == {"qualifiedName": SHEET_QN} - assert sut.qlik_space_id == "st01" - assert sut.qlik_space_qualified_name == STREAM_QN - sut.validate(for_creation=True) - - -@pytest.mark.parametrize( - "asset_type, parent", - [ - (QlikSheet, {"app_qualified_name": APP_QN}), - (QlikChart, {"sheet_qualified_name": SHEET_QN}), - ], -) -def test_sheet_and_chart_space_is_optional(asset_type, parent): - sut = asset_type.creator(name=NAME, qlik_id="x1", **parent) - - assert sut.qlik_space_id is UNSET - assert sut.qlik_space_qualified_name is UNSET - - -@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_column_creator_matches_app_qualified_name( - parent_qn, relationship, expected_qn -): - sut = QlikColumn.creator( - name=FIELD, parent_qualified_name=parent_qn, space_qualified_name=SPACE_QN - ) - - assert sut.qualified_name == expected_qn - assert sut.name == FIELD - assert sut.qlik_column_name == FIELD - 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 {"qlik_sheet", "qlik_chart", "qlik_dataset"} - {relationship}: - assert not getattr(sut, other) - # 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_column_without_parent_fails_creation_validation(): - sut = QlikColumn( - name=FIELD, - 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( - "creator, kwargs, message", - [ - ( - QlikSpace.creator, - {"name": "", "connection_qualified_name": CONNECTION_QN, "qlik_id": "x"}, - "name", - ), - ( - QlikStream.creator, - {"name": NAME, "connection_qualified_name": CONNECTION_QN, "qlik_id": ""}, - "qlik_id", - ), - ( - QlikApp.creator, - {"name": NAME, "qlik_id": "x", "space_qualified_name": None}, - "space_qualified_name", - ), - ( - QlikDataset.creator, - {"name": NAME, "qlik_id": " ", "space_qualified_name": SPACE_QN}, - "qlik_id", - ), - ( - QlikSheet.creator, - {"name": NAME, "qlik_id": "x", "app_qualified_name": ""}, - "app_qualified_name", - ), - ( - QlikChart.creator, - {"name": NAME, "qlik_id": "x", "sheet_qualified_name": None}, - "sheet_qualified_name", - ), - ( - QlikColumn.creator, - {"name": "", "parent_qualified_name": SHEET_QN}, - "name", - ), - ], -) -def test_creator_with_blank_input_raises(creator, kwargs, message): - with pytest.raises(ValueError, match=message): - creator(**kwargs) - - -@pytest.mark.parametrize( - "creator, kwargs", - [ - ( - QlikApp.creator, - {"name": NAME, "qlik_id": "x", "space_qualified_name": APP_QN}, - ), - ( - QlikDataset.creator, - {"name": NAME, "qlik_id": "x", "space_qualified_name": CONNECTION_QN}, - ), - ( - QlikSheet.creator, - {"name": NAME, "qlik_id": "x", "app_qualified_name": SPACE_QN}, - ), - ( - QlikSheet.creator, - { - "name": NAME, - "qlik_id": "x", - "app_qualified_name": APP_QN, - "space_qualified_name": DATASET_QN, - }, - ), - ( - QlikChart.creator, - {"name": NAME, "qlik_id": "x", "sheet_qualified_name": CHART_QN}, - ), - (QlikColumn.creator, {"name": FIELD, "parent_qualified_name": APP_QN}), - (QlikColumn.creator, {"name": FIELD, "parent_qualified_name": SPACE_QN}), - ], -) -def test_creator_with_wrong_parent_qualified_name_raises(creator, kwargs): - with pytest.raises(ValueError, match="must end in"): - creator(**kwargs)