diff --git a/pyatlan/generator/generate_apps.py b/pyatlan/generator/generate_apps.py index 556b09196..0a1bb7f58 100644 --- a/pyatlan/generator/generate_apps.py +++ b/pyatlan/generator/generate_apps.py @@ -44,7 +44,15 @@ # Modules with a hand-polished builder — the generator leaves these untouched. # databricks_crawler has a hand-written multi-mode asset_selection (the configmap # can't express its include/exclude × hierarchy/regex widget). -_HAND_WRITTEN = {"bigquery_crawler", "databricks_crawler", "kafka_confluent"} +# standard_lineage re-scopes an EXISTING workflow (an update against a slug, which +# no generated builder does) and must JSON-encode a list into a contract field +# declared `str`; neither is expressible from a configmap. +_HAND_WRITTEN = { + "bigquery_crawler", + "databricks_crawler", + "kafka_confluent", + "standard_lineage", +} # Apps to generate even when not currently deployed/running on the tenant # (configmaps are served per app-id, so live discovery alone misses these). diff --git a/pyatlan/model/apps/__init__.py b/pyatlan/model/apps/__init__.py index f5aab1529..998f9f3e5 100644 --- a/pyatlan/model/apps/__init__.py +++ b/pyatlan/model/apps/__init__.py @@ -7,8 +7,6 @@ """ from pyatlan.model.apps._base import AppBuilder, AppInput -from pyatlan.model.apps.bigquery_crawler import BigqueryCrawler, BigqueryCrawlerInputs -from pyatlan.model.apps.bigquery_miner import BigqueryMiner, BigqueryMinerInputs from pyatlan.model.apps.anaplan import Anaplan, AnaplanInputs from pyatlan.model.apps.atlan_athena import AtlanAthena, AtlanAthenaInputs from pyatlan.model.apps.atlan_dbt import AtlanDbt, AtlanDbtInputs @@ -27,6 +25,8 @@ from pyatlan.model.apps.atlan_sigma import AtlanSigma, AtlanSigmaInputs from pyatlan.model.apps.atlan_tableau import AtlanTableau, AtlanTableauInputs from pyatlan.model.apps.atlan_trino import AtlanTrino, AtlanTrinoInputs +from pyatlan.model.apps.bigquery_crawler import BigqueryCrawler, BigqueryCrawlerInputs +from pyatlan.model.apps.bigquery_miner import BigqueryMiner, BigqueryMinerInputs from pyatlan.model.apps.csa_uber_asset_export_basic import ( CsaUberAssetExportBasic, CsaUberAssetExportBasicInputs, @@ -39,10 +39,7 @@ from pyatlan.model.apps.hive_crawler import HiveCrawler, HiveCrawlerInputs from pyatlan.model.apps.kafka_apache import KafkaApache, KafkaApacheInputs from pyatlan.model.apps.kafka_confluent import KafkaConfluent, KafkaConfluentInputs -from pyatlan.model.apps.mongodbatlas_atlas import ( - MongodbAtlas, - MongodbAtlasInputs, -) +from pyatlan.model.apps.mongodbatlas_atlas import MongodbAtlas, MongodbAtlasInputs from pyatlan.model.apps.oracle_crawler import OracleCrawler, OracleCrawlerInputs from pyatlan.model.apps.oracle_miner import OracleMiner, OracleMinerInputs from pyatlan.model.apps.postgres_crawler import PostgresCrawler, PostgresCrawlerInputs @@ -54,6 +51,7 @@ SnowflakeCrawlerInputs, ) from pyatlan.model.apps.snowflake_miner import SnowflakeMiner, SnowflakeMinerInputs +from pyatlan.model.apps.standard_lineage import StandardLineage, StandardLineageInputs from pyatlan.model.apps.teradata_crawler import TeradataCrawler, TeradataCrawlerInputs from pyatlan.model.apps.teradata_miner import TeradataMiner, TeradataMinerInputs @@ -124,6 +122,8 @@ "SnowflakeCrawlerInputs", "SnowflakeMiner", "SnowflakeMinerInputs", + "StandardLineage", + "StandardLineageInputs", "TeradataCrawler", "TeradataCrawlerInputs", "TeradataMiner", diff --git a/pyatlan/model/apps/_base.py b/pyatlan/model/apps/_base.py index a2ca6db2c..4db21a040 100644 --- a/pyatlan/model/apps/_base.py +++ b/pyatlan/model/apps/_base.py @@ -25,6 +25,10 @@ from pyatlan.model.credential import Credential +# Handshake/runtime ids the server injects into a workflow's inputs; they must +# not be echoed back on update (stripped server-side on create). +_RUNTIME_KEYS = frozenset({"user-id", "user_id", "workflow_id", "correlation_id"}) + class AppInput(BaseModel): """A typed, configmap-derived ``inputs`` payload for an app workflow.""" @@ -109,6 +113,7 @@ def __init__(self, client: Any): self._admin_groups: List[str] = [] self._admin_roles: List[str] = [] self._metadata: Dict[str, Any] = {} + self._update_slug: Optional[str] = None # ── Step 1 · Credential ──────────────────────────────────────────────── def _stage_credential(self, field: str, credential: Credential): @@ -156,7 +161,12 @@ def connection( return self # ── assembly (no network) ────────────────────────────────────────────── - def _build_connection(self, qualified_name: str) -> Dict[str, Any]: + def _build_connection( + self, + qualified_name: str, + *, + default_credential_guid: Optional[str] = None, + ) -> Dict[str, Any]: # Derive the connector from the QN (``default/{connector}/{epoch}``) so a # referenced existing connection (e.g. for miners) reports the right # connectorName, not the builder's app-id-derived fallback. @@ -170,6 +180,8 @@ def _build_connection(self, qualified_name: str) -> Dict[str, Any]: "qualifiedName": qualified_name, "connectorName": connector, } + if default_credential_guid: + attrs["defaultCredentialGuid"] = default_credential_guid if self._connection_name: attrs["name"] = self._connection_name if self._admin_users: @@ -214,9 +226,27 @@ def _assemble( resolved_guids: Optional[Dict[str, str]] = None, ) -> AppInput: resolved_guids = resolved_guids or {} + # Reusing an already-vaulted credential on an EXISTING connection (e.g. a + # miner picking up that connection's own credential): the guid rides on the + # connection entity, the way the UI sends it, and the top-level + # credential_guid stays "". A bare top-level guid with no credential body + # makes the create endpoint rewrite that credential's shared config record + # from the (absent) body, flattening it to {"credentialSource": "direct"} + # for every workflow sharing the guid (CONNECT-843). + reuse_on_existing_connection = bool( + self._extraction_method != "agent" + and self._credential_guid + and self._connection_qualified_name + and "credential_guid" not in self._raw_creds + ) kwargs: Dict[str, Any] = dict(self._HIDDEN_DEFAULTS) kwargs.update(self._metadata) - kwargs["connection"] = self._build_connection(qualified_name) + kwargs["connection"] = self._build_connection( + qualified_name, + default_credential_guid=( + self._credential_guid if reuse_on_existing_connection else None + ), + ) kwargs["extraction_method"] = self._extraction_method if self._extraction_method == "agent": kwargs["agent_json"] = self._agent_json @@ -240,9 +270,12 @@ def _assemble( else: kwargs[field] = self._raw_credential(cred, epoch=epoch, redact=True) # credential_guid is a (non-null) string in the contract: reuse an existing - # guid if given, else "" (omitting it reads as null and is rejected). + # guid if given, else "" (omitting it reads as null and is rejected). When + # the guid rides on the connection instead, this stays "" (see above). kwargs["credential_guid"] = ( - self._credential_guid if self._credential_guid is not None else "" + "" + if reuse_on_existing_connection or self._credential_guid is None + else self._credential_guid ) return self._INPUTS_CLASS(**kwargs) @@ -267,6 +300,113 @@ def run(self, *, name: Optional[str] = None, schedule: Optional[Any] = None): """Create the workflow **and** submit a run immediately (``run=True``).""" return self._create(name=name, run=True, schedule=schedule) + # ── update (load an existing workflow, change fields, full-replace) ───── + def load(self, slug: str) -> "AppBuilder": + """Seed this builder from an existing workflow's current inputs, so + :meth:`update` can change a few fields and preserve the rest. + + Mirrors the create builders — same fluent methods, one extra step:: + + Builder(client).load(slug).include({...}).update() + + The connection and the existing credential are read from the workflow and + reused: the credential is referenced by guid and **never re-vaulted or + rotated**. Everything you don't change is carried through untouched. + """ + args = ( + self._client.app.get(slug) + .dict() + .get("dag", {}) + .get("extract", {}) + .get("inputs", {}) + .get("args", {}) + ) + self._update_slug = slug + conn = args.get("connection") + conn = json.loads(conn) if isinstance(conn, str) else conn + attrs = (conn or {}).get("attributes", {}) if isinstance(conn, dict) else {} + self._connection_qualified_name = attrs.get("qualifiedName") + self._connection_name = attrs.get("name") + self._admin_users = list(attrs.get("adminUsers") or []) + self._admin_groups = list(attrs.get("adminGroups") or []) + self._admin_roles = list(attrs.get("adminRoles") or []) + # The reused credential guid may ride on the connection entity + # (attributes.defaultCredentialGuid, the UI/CONNECT-843 shape) or, on + # older workflows, sit top-level. Read whichever is present so update() + # never drops the credential reference. + self._credential_guid = ( + args.get("credential_guid") or attrs.get("defaultCredentialGuid") or "" + ) + self._extraction_method = ( + args.get("extraction_method") or self._EXTRACTION_METHOD + ) + # Carry every non-structural current field so nothing is dropped on the + # full replace; the structural bits are rebuilt from the state above. + structural = { + "connection", + "connection_qualified_name", + "credential", + "credential_guid", + "agent_json", + "extraction_method", + } | _RUNTIME_KEYS + self._metadata = {k: v for k, v in args.items() if k not in structural} + return self + + def update(self): + """Publish a new version of the loaded workflow (full-replace) with the + builder's current fields. Call :meth:`load` first. + + Re-injects ``connection_qualified_name`` (the read-back omits it, but the + run needs it) and re-encodes string-typed contract fields returned as + objects — so a caller only sets what they want to change. + """ + if not self._update_slug: + raise ValueError("call load(slug) before update()") + # Normalize BEFORE assembly — the typed inputs model rejects a string + # field (e.g. control_config) that the read-back returned as an object. + self._metadata = self._normalize_string_inputs(dict(self._metadata)) + epoch = int(time.time()) + qn = ( + self._connection_qualified_name or f"default/{self._CONNECTOR_NAME}/{epoch}" + ) + payload = self._assemble(qualified_name=qn, epoch=epoch).to_inputs() + payload["connection_qualified_name"] = qn + + def _update_with(entrypoint: Optional[str]): + return self._client.app.update( + slug=self._update_slug, inputs=payload, entrypoint=entrypoint or None + ) + + # Same 1003 fallback as _create(): retry at the default slot if the named + # entrypoint has no registered contract. + ep = self._ENTRYPOINT or None + try: + return _update_with(ep) + except Exception as exc: # noqa: BLE001 + msg = str(exc) + if ep is not None and ("1003" in msg or "unknown entrypoint" in msg): + return _update_with(None) + raise + + def _normalize_string_inputs(self, payload: Dict[str, Any]) -> Dict[str, Any]: + """JSON-encode string-typed contract fields the read-back returned as + objects (e.g. ``control_config``), so the full-replace matches the contract.""" + try: + contract = self._client.app.get_input_contract( + self._APP_ID, self._ENTRYPOINT or None + ) + except Exception: # noqa: BLE001 - best-effort; skip if the contract can't load + return payload + for key, spec in (contract.properties or {}).items(): + if ( + isinstance(spec, dict) + and spec.get("type") == "string" + and isinstance(payload.get(key), (dict, list)) + ): + payload[key] = json.dumps(payload[key]) + return payload + def _vault_credential(self, cred: Credential) -> str: """Vault a raw credential into Atlan's secret store and return its guid. diff --git a/pyatlan/model/apps/bigquery_crawler.py b/pyatlan/model/apps/bigquery_crawler.py index 035e9d8b6..e58bf4207 100644 --- a/pyatlan/model/apps/bigquery_crawler.py +++ b/pyatlan/model/apps/bigquery_crawler.py @@ -132,12 +132,41 @@ def workload_identity_federation( self, *, project_id: str, + service_account_email: str, + wif_pool_provider_id: str, + atlan_oauth_id: str, + atlan_oauth_secret: str, connectivity: str = "public", host: Optional[str] = None, port: Optional[int] = None, **extra: Any, ) -> "BigqueryCrawler": - """Direct extraction with Workload Identity Federation auth.""" + """Direct extraction with Workload Identity Federation (keyless) auth. + + Atlan mints a token from its own OAuth client and exchanges it through the + WIF pool/provider to impersonate ``service_account_email`` — so no Service + Account key is stored. All five values are required by the ``gcp-wif`` + credential; a missing one is a ``TypeError`` rather than a silently broken + credential. Any additional keyword rides through into ``extra`` unchanged + (forward-compatible with fields newer than this signature). + + :param project_id: GCP project ID. + :param service_account_email: Service Account that Atlan impersonates via WIF. + :param wif_pool_provider_id: full Workload Identity Pool *provider* resource id. + :param atlan_oauth_id: Atlan OAuth client id used for the token exchange. + :param atlan_oauth_secret: Atlan OAuth client secret (vaulted, never logged). + :param connectivity: ``public`` (Google's endpoint) or ``private`` (PSC). + :param host: Private Service Connect host (required for ``private``). + """ + wif: Dict[str, Any] = { + "project_id": project_id, + "connect_type": connectivity, + "service_account_email": service_account_email, + "wif_pool_provider_id": wif_pool_provider_id, + "atlan_oauth_id": atlan_oauth_id, + "atlan_oauth_secret": atlan_oauth_secret, + } + wif.update(extra) return self._stage_credential( "credential_guid", Credential( @@ -146,7 +175,7 @@ def workload_identity_federation( auth_type="gcp-wif", host=host or self._DEFAULT_HOST, port=port or self._DEFAULT_PORT, - extra={"project_id": project_id, "connect_type": connectivity, **extra}, + extra=wif, ), ) diff --git a/pyatlan/model/apps/standard_lineage.py b/pyatlan/model/apps/standard_lineage.py new file mode 100644 index 000000000..2c26abc64 --- /dev/null +++ b/pyatlan/model/apps/standard_lineage.py @@ -0,0 +1,287 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +"""Standard Lineage app — typed inputs + fluent builder. + +Hand-written rather than generated, for two reasons the configmap cannot express: + +* ``cross_connection_qualified_names`` is declared ``str`` in the app's input + contract but *means* a list of connection qualified names. The generator would + emit a bare ``str`` field and every caller would have to remember to + ``json.dumps`` it — sending a native list fails contract validation server-side. + :meth:`StandardLineage.connections` takes a ``List[str]`` and encodes it. +* Standard Lineage's defining operation is **re-scoping an existing workflow** + (adding a connection as it is onboarded), not creating a new one. That is an + update against a slug, and it has to preserve the workflow's own connection + entity — see :meth:`StandardLineage.add_connections`. + +Sourced from the app's UI configmap and input contract: + * inputs form : /api/service/configmaps/atlan-standard-lineage?entrypoint=standard-lineage + * input contract : /api/service/v1/apps/atlan-standard-lineage/inputs?entrypoint=standard-lineage + +There is no credential: the app reads query history already extracted by each +in-scope connection's own miner, so the contract's ``credential_ref`` and +``agent_json`` are optional and stay null. +""" + +from __future__ import annotations + +import json +import re +from typing import Any, ClassVar, Dict, Iterable, List, Optional, Sequence, Union + +from ._base import AppBuilder, AppInput + +#: A connection qualified name is ``default/{connector}/{epoch}``. +_CONNECTION_QN = re.compile(r"^default/([^/]+)/\d+$") + +#: The connector of the workflow's *own* connection — distinct from the +#: ``connector`` input, which names the connector of the connections in scope. +_OWN_CONNECTOR = "standard-lineage" + + +def _connector_of(qualified_name: str) -> str: + """Return the connector segment of a connection qualified name.""" + match = _CONNECTION_QN.match(qualified_name) + if not match: + raise ValueError( + f"{qualified_name!r} is not a connection qualified name " + "(expected 'default/{connector}/{epoch}')" + ) + return match.group(1) + + +def _validate_scope(qualified_names: Sequence[str]) -> str: + """Validate a scope list and return the single connector it covers. + + The app requires a non-empty, same-connector scope and fails the run when that + does not hold, so this is checked client-side where the error is actionable. + """ + if not qualified_names: + raise ValueError( + "Standard Lineage needs at least one connection in scope; " + "to stop processing a connection, remove it and leave the rest, " + "or delete the workflow" + ) + connectors = {_connector_of(qn) for qn in qualified_names} + if len(connectors) > 1: + raise ValueError( + "every connection in scope must belong to the same connector, got " + f"{sorted(connectors)}; use one Standard Lineage workflow per connector" + ) + connector = connectors.pop() + if connector == _OWN_CONNECTOR: + raise ValueError( + "the scope must list the SOURCE connections to build lineage across " + f"(e.g. 'default/bigquery/1700000000'), not the workflow's own " + f"{_OWN_CONNECTOR!r} connection" + ) + return connector + + +def _parse_scope(value: Any) -> List[str]: + """Read a scope back out of a persisted workflow. + + The value is a JSON-encoded string on the wire but a native list once the + Automation Engine has rendered it into the DAG, so both shapes occur. + """ + if value is None: + return [] + if isinstance(value, str): + if not value.strip(): + return [] + try: + parsed = json.loads(value) + except json.JSONDecodeError: + # A bare single qualified name, tolerated rather than crashed on. + return [value.strip()] + return [str(v) for v in parsed] if isinstance(parsed, list) else [str(parsed)] + if isinstance(value, Iterable): + return [str(v) for v in value] + return [str(value)] + + +class StandardLineageInputs(AppInput): + """Typed inputs for the ``atlan-standard-lineage`` / ``standard-lineage`` app.""" + + _APP_ID: ClassVar[str] = "atlan-standard-lineage" + _ENTRYPOINT: ClassVar[Optional[str]] = "standard-lineage" + + # Step 1 · Connection — the workflow's OWN connection, under the + # ``standard-lineage`` connector. Created by the workflow on first run. + connection: Optional[Any] = None + + # Step 3 · Metadata + connector: str = "" + """Connector of the connections in scope (e.g. ``bigquery``).""" + cross_connection_qualified_names: str = "" + """JSON-encoded list of source connection qualified names. Prefer + :meth:`StandardLineage.connections`, which encodes a ``List[str]`` for you.""" + run_role: str = "standard-lineage" + """Fixed — this is what makes the lineage app take its cross-connection path.""" + + +class StandardLineage(AppBuilder): + """Fluent builder for the Standard Lineage (cross-connection lineage) app. + + Standard Lineage builds lineage *across* several connections of one connector, + using query history their own miners already extracted. One workflow owns a set + of connections; onboarding a new connection means adding it to that set. + + Create a workflow:: + + resp = ( + StandardLineage(client) + .connection(name="bigquery-cross-connection") + .connections([ + "default/bigquery/1700000000", + "default/bigquery/1700000001", + ]) + .run() + ) + + Add a connection to an existing workflow — the common case, and idempotent:: + + resp = StandardLineage(client).add_connections( + resp.slug, ["default/bigquery/1700000002"] + ) + + Also available: :meth:`remove_connections`, :meth:`set_connections` and the + read-only :meth:`get_connections`. + """ + + _APP_ID: ClassVar[str] = "atlan-standard-lineage" + _ENTRYPOINT: ClassVar[Optional[str]] = "standard-lineage" + #: The workflow's own connection is minted under this connector; the connector + #: of the connections in scope is the separate ``connector`` input. + _CONNECTOR_NAME: ClassVar[str] = _OWN_CONNECTOR + _CONNECTOR_CONFIG: ClassVar[str] = "" + _INPUTS_CLASS = StandardLineageInputs + _HIDDEN_DEFAULTS: ClassVar[Dict[str, Any]] = {"run_role": "standard-lineage"} + #: No credential — nothing is extracted from a source system directly. + _EXTRACTION_METHOD: ClassVar[str] = "" + + # ── Step 3 · Metadata ────────────────────────────────────────────────── + def connections( + self, + qualified_names: Union[Sequence[str], str], + *, + connector: Optional[str] = None, + ): + """Set the connections to build lineage across (create-time scope). + + :param qualified_names: source connection qualified names, e.g. + ``["default/bigquery/1700000000", ...]``. All must belong to the same + connector. A pre-encoded JSON string is accepted as-is. + :param connector: the scope's connector; derived from + ``qualified_names`` when omitted. + :raises ValueError: on an empty scope, a malformed qualified name, or a + scope spanning more than one connector. + + To change the scope of a workflow that already exists, use + :meth:`add_connections` / :meth:`remove_connections` / :meth:`set_connections` + — those preserve the workflow's own connection, which this does not know about. + """ + scope = _parse_scope(qualified_names) + derived = _validate_scope(scope) + self._metadata["connector"] = connector or derived + # The contract declares this field as `str`, so it goes over JSON-encoded; + # Heracles parses it back into a list for the manifest placeholder. + self._metadata["cross_connection_qualified_names"] = json.dumps(scope) + self._metadata["run_role"] = "standard-lineage" + return self + + # ── Re-scoping an existing workflow (network) ─────────────────────────── + def get_connections(self, slug: str) -> List[str]: + """Return the connections currently in scope for ``slug``. Read-only.""" + return _parse_scope( + self._persisted_args(slug).get("cross_connection_qualified_names") + ) + + def add_connections(self, slug: str, qualified_names: Union[Sequence[str], str]): + """Add connections to an existing workflow's scope, keeping the rest. + + Idempotent: connections already in scope are ignored, and when nothing + would change no version is published and ``None`` is returned. This is the + onboarding call — adding each new connection as it is created. + """ + current = self.get_connections(slug) + additions = [qn for qn in _parse_scope(qualified_names) if qn not in current] + if not additions: + return None + return self.set_connections(slug, current + additions) + + def remove_connections(self, slug: str, qualified_names: Union[Sequence[str], str]): + """Remove connections from an existing workflow's scope, keeping the rest. + + Idempotent in the same way as :meth:`add_connections`. Removing every + connection raises — the app cannot run on an empty scope. + + The next run hands each removed connection back to its own miner and + crawler, by flipping the per-connection standard-lineage marker off. + """ + current = self.get_connections(slug) + removals = set(_parse_scope(qualified_names)) + remaining = [qn for qn in current if qn not in removals] + if len(remaining) == len(current): + return None + return self.set_connections(slug, remaining) + + def set_connections( + self, + slug: str, + qualified_names: Union[Sequence[str], str], + *, + connector: Optional[str] = None, + ): + """Replace an existing workflow's scope with exactly ``qualified_names``. + + Preserves the workflow's own connection, its ``run_role`` and its identity + by reading them back from the persisted workflow — ``client.app.update`` is + a full replace, so anything omitted from the payload would be dropped from + the new version, and the workflow's connection entity is republished on + every run (a partial one would strip its admins). + """ + scope = _parse_scope(qualified_names) + derived = _validate_scope(scope) + args = self._persisted_args(slug) + own_connection = args.get("connection") + if not own_connection: + raise ValueError( + f"workflow {slug!r} has no connection on its extract node; refusing " + "to update, because a partial connection would be republished over " + "the real one" + ) + inputs = self._INPUTS_CLASS( + connection=own_connection, + connector=connector or args.get("connector") or derived, + cross_connection_qualified_names=json.dumps(scope), + run_role=args.get("run_role") or "standard-lineage", + ) + return self._client.app.update( + slug=slug, inputs=inputs, entrypoint=self._ENTRYPOINT + ) + + # ── internals ────────────────────────────────────────────────────────── + def _persisted_args(self, slug: str) -> Dict[str, Any]: + """Read the persisted workflow's extract-node args. + + ``GET /v1/app/{slug}`` returns the rendered DAG; ``AppSummary`` tolerates + unmodelled fields, so it arrives as an extra attribute. + """ + summary = self._client.app.get(slug) + dag = getattr(summary, "dag", None) + if not isinstance(dag, dict): + raise ValueError( + f"workflow {slug!r} returned no DAG to read the scope from" + ) + node = dag.get("extract") + args = ((node or {}).get("inputs") or {}).get("args") if node else None + if not isinstance(args, dict): + raise ValueError( + f"workflow {slug!r} has no extract node args; is it a Standard " + "Lineage workflow?" + ) + return args + + +__all__ = ["StandardLineage", "StandardLineageInputs"] diff --git a/tests/unit/apps/test_bigquery_crawler.py b/tests/unit/apps/test_bigquery_crawler.py new file mode 100644 index 000000000..dfd9927f2 --- /dev/null +++ b/tests/unit/apps/test_bigquery_crawler.py @@ -0,0 +1,93 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +# Hand-written: bigquery_crawler is a hand-authored builder (see generate_apps +# _HAND_WRITTEN), so this test is preserved across regeneration. +from unittest.mock import Mock + +import pytest + +from pyatlan.model.apps import BigqueryCrawler, BigqueryCrawlerInputs + +WIF = dict( + project_id="my-project", + service_account_email="svc@my-project.iam.gserviceaccount.com", + wif_pool_provider_id="projects/1/locations/global/workloadIdentityPools/p/providers/pr", + atlan_oauth_id="oauth-client-id", + atlan_oauth_secret="oauth-secret", +) + + +def test_bigquery_crawler_inputs_defaults(): + i = BigqueryCrawlerInputs() + assert BigqueryCrawlerInputs._APP_ID == "bigquery-crawler" + assert BigqueryCrawlerInputs._ENTRYPOINT == "crawler" + assert i.include_filter == "{}" + assert i.exclude_filter == "{}" + assert i.enable_nested_columns is True + assert i.filter_sharded_tables is True + + +def test_bigquery_crawler_builder_payload(): + out = ( + BigqueryCrawler(Mock()) + .service_account( + email="svc@my-project.iam.gserviceaccount.com", + service_account_json="{}", + project_id="my-project", + ) + .connection(name="conn", admin_users=["u"]) + .include({"my-project": ["analytics"]}) + .preview() + ) + assert out["connection"]["attributes"]["connectorName"] == "bigquery" + assert out["extraction_method"] == "direct" + assert out["include_filter"] == '{"^my-project$": ["^analytics$"]}' + + +def test_wif_stages_gcp_wif_credential_with_the_five_typed_keys(): + b = BigqueryCrawler(Mock()).workload_identity_federation(**WIF) + cred = b._raw_creds["credential_guid"] + assert cred.auth_type == "gcp-wif" + assert cred.connector_config_name == "atlan-connectors-bigquery" + # WIF carries no username/password — auth is the token exchange. + assert cred.username is None and cred.password is None + assert cred.extras == { + "project_id": "my-project", + "connect_type": "public", + "service_account_email": "svc@my-project.iam.gserviceaccount.com", + "wif_pool_provider_id": "projects/1/locations/global/workloadIdentityPools/p/providers/pr", + "atlan_oauth_id": "oauth-client-id", + "atlan_oauth_secret": "oauth-secret", + } + + +def test_wif_keys_land_in_the_previewed_credential_payload(): + out = ( + BigqueryCrawler(Mock()) + .workload_identity_federation(**WIF) + .connection(name="conn") + .preview() + ) + cred = out["credential"] + assert cred["authType"] == "gcp-wif" + assert set(cred["extra"]) == { + "project_id", + "connect_type", + "service_account_email", + "wif_pool_provider_id", + "atlan_oauth_id", + "atlan_oauth_secret", + } + + +def test_wif_requires_every_credential_field(): + # Each WIF field is required; a missing one is a TypeError, not a silently + # incomplete credential. A misspelled required field trips the same check. + with pytest.raises(TypeError): + BigqueryCrawler(Mock()).workload_identity_federation(project_id="p") + + +def test_wif_forwards_unknown_extra_keys(): + # Forward-compatible: keys newer than this signature ride through **extra. + b = BigqueryCrawler(Mock()).workload_identity_federation(**WIF, future_flag="x") + assert b._raw_creds["credential_guid"].extras["future_flag"] == "x" diff --git a/tests/unit/apps/test_standard_lineage.py b/tests/unit/apps/test_standard_lineage.py new file mode 100644 index 000000000..c35b6d590 --- /dev/null +++ b/tests/unit/apps/test_standard_lineage.py @@ -0,0 +1,234 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 Atlan Pte. Ltd. +"""HAND-WRITTEN tests for the Standard Lineage builder (CONNECT-182). + +The generator's regen only clears files carrying the AUTO-GENERATED banner, so this +file survives regeneration alongside the builder it covers. + +Most cases pin the re-scope path rather than create: the builder's job is to change an +existing workflow's set of connections without disturbing the workflow's own connection +entity, and ``client.app.update`` is a full replace — so what is *absent* from the +payload matters as much as what is present. +""" + +import json +from unittest.mock import Mock + +import pytest + +from pyatlan.model.apps import StandardLineage, StandardLineageInputs + +SLUG = "atlan-standard-lineage-1700000000-Abcd1234" +BQ1 = "default/bigquery/1700000001" +BQ2 = "default/bigquery/1700000002" +BQ3 = "default/bigquery/1700000003" + +# The workflow's own connection, as it comes back on the persisted DAG. Every +# attribute matters: the create-connection node republishes this entity, so a +# partial copy would strip the connection's name and admins in Atlan. +OWN_CONNECTION = { + "typeName": "Connection", + "attributes": { + "qualifiedName": "default/standard-lineage/1700000000", + "name": "bq-cross-connection", + "connectorName": "standard-lineage", + "category": "lineage", + "adminUsers": ["someone"], + "adminRoles": ["role-guid"], + "adminGroups": [], + "rowLimit": 10000, + }, +} + + +def _client(scope, *, connection=OWN_CONNECTION, connector="bigquery"): + """A mock client whose app.get() returns a persisted Standard Lineage DAG.""" + args = {"connector": connector, "run_role": "standard-lineage"} + if connection is not None: + args["connection"] = connection + if scope is not None: + args["cross_connection_qualified_names"] = scope + client = Mock() + client.app.get.return_value = Mock(dag={"extract": {"inputs": {"args": args}}}) + client.app.update.return_value = Mock(slug=SLUG, version=1700000009) + return client + + +def _sent_inputs(client): + """The inputs dict actually handed to client.app.update().""" + return client.app.update.call_args.kwargs["inputs"].to_inputs() + + +# ── inputs model ──────────────────────────────────────────────────────────── +def test_inputs_defaults(): + i = StandardLineageInputs() + assert StandardLineageInputs._APP_ID == "atlan-standard-lineage" + assert StandardLineageInputs._ENTRYPOINT == "standard-lineage" + assert i.connector == "" + assert i.cross_connection_qualified_names == "" + assert i.run_role == "standard-lineage" + + +def test_builder_class_vars(): + assert StandardLineage._APP_ID == "atlan-standard-lineage" + assert StandardLineage._ENTRYPOINT == "standard-lineage" + # The workflow's OWN connection lives under standard-lineage; the connections + # in scope are a different connector entirely. + assert StandardLineage._CONNECTOR_NAME == "standard-lineage" + + +# ── create-time scope ─────────────────────────────────────────────────────── +def test_connections_json_encodes_and_derives_connector(): + out = ( + StandardLineage(Mock()) + .connection(name="bq-cross-connection") + .connections([BQ1, BQ2]) + .preview() + ) + # Declared `str` in the contract — a native list fails validation server-side. + assert isinstance(out["cross_connection_qualified_names"], str) + assert json.loads(out["cross_connection_qualified_names"]) == [BQ1, BQ2] + assert out["connector"] == "bigquery" + assert out["run_role"] == "standard-lineage" + assert out["connection"]["attributes"]["connectorName"] == "standard-lineage" + + +def test_connections_explicit_connector_wins(): + out = ( + StandardLineage(Mock()) + .connections([BQ1], connector="bigquery-custom") + .preview() + ) + assert out["connector"] == "bigquery-custom" + + +def test_connections_accepts_a_preencoded_json_string(): + out = StandardLineage(Mock()).connections(json.dumps([BQ1, BQ2])).preview() + assert json.loads(out["cross_connection_qualified_names"]) == [BQ1, BQ2] + + +# ── validation ────────────────────────────────────────────────────────────── +def test_empty_scope_is_rejected(): + with pytest.raises(ValueError, match="at least one connection"): + StandardLineage(Mock()).connections([]) + + +def test_mixed_connector_scope_is_rejected(): + with pytest.raises(ValueError, match="same connector"): + StandardLineage(Mock()).connections([BQ1, "default/snowflake/1700000004"]) + + +def test_own_connection_as_scope_is_rejected(): + """A natural mistake: passing the workflow's own connection as its scope.""" + with pytest.raises(ValueError, match="not the workflow's own"): + StandardLineage(Mock()).connections(["default/standard-lineage/1700000000"]) + + +def test_malformed_qualified_name_is_rejected(): + with pytest.raises(ValueError, match="not a connection qualified name"): + StandardLineage(Mock()).connections(["bigquery/1700000001"]) + + +# ── reading an existing workflow ──────────────────────────────────────────── +def test_get_connections_reads_a_native_list(): + """Once the Automation Engine renders the DAG the value is a real list.""" + assert StandardLineage(_client([BQ1, BQ2])).get_connections(SLUG) == [BQ1, BQ2] + + +def test_get_connections_reads_a_json_string(): + """On the wire it is JSON-encoded, so both shapes have to be readable.""" + assert StandardLineage(_client(json.dumps([BQ1, BQ2]))).get_connections(SLUG) == [ + BQ1, + BQ2, + ] + + +def test_get_connections_on_empty_scope(): + assert StandardLineage(_client("")).get_connections(SLUG) == [] + + +def test_missing_extract_args_raises(): + client = Mock() + client.app.get.return_value = Mock(dag={"publish": {}}) + with pytest.raises(ValueError, match="no extract node args"): + StandardLineage(client).get_connections(SLUG) + + +# ── re-scoping ────────────────────────────────────────────────────────────── +def test_set_connections_preserves_the_persisted_connection_verbatim(): + """The whole reason set_connections reads before it writes. + + ``client.app.update`` is a full replace, and the connection entity is + republished on every run — so sending a rebuilt or partial connection would + overwrite the real one in Atlan. + """ + client = _client([BQ1, BQ2]) + StandardLineage(client).set_connections(SLUG, [BQ1, BQ2, BQ3]) + sent = _sent_inputs(client) + assert sent["connection"] == OWN_CONNECTION + assert sent["run_role"] == "standard-lineage" + assert json.loads(sent["cross_connection_qualified_names"]) == [BQ1, BQ2, BQ3] + assert client.app.update.call_args.kwargs["entrypoint"] == "standard-lineage" + assert client.app.update.call_args.kwargs["slug"] == SLUG + + +def test_set_connections_refuses_when_the_workflow_has_no_connection(): + client = _client([BQ1], connection=None) + with pytest.raises(ValueError, match="no connection on its extract node"): + StandardLineage(client).set_connections(SLUG, [BQ1, BQ2]) + client.app.update.assert_not_called() + + +def test_set_connections_refuses_to_empty_the_scope(): + client = _client([BQ1]) + with pytest.raises(ValueError, match="at least one connection"): + StandardLineage(client).set_connections(SLUG, []) + client.app.update.assert_not_called() + + +def test_add_connections_appends_and_keeps_the_rest(): + client = _client([BQ1, BQ2]) + StandardLineage(client).add_connections(SLUG, [BQ3]) + assert json.loads(_sent_inputs(client)["cross_connection_qualified_names"]) == [ + BQ1, + BQ2, + BQ3, + ] + + +def test_add_connections_is_idempotent(): + """The onboarding portal may replay; a no-op must not publish a version.""" + client = _client([BQ1, BQ2]) + assert StandardLineage(client).add_connections(SLUG, [BQ2]) is None + client.app.update.assert_not_called() + + +def test_add_connections_adds_only_the_new_ones(): + client = _client([BQ1]) + StandardLineage(client).add_connections(SLUG, [BQ1, BQ2]) + assert json.loads(_sent_inputs(client)["cross_connection_qualified_names"]) == [ + BQ1, + BQ2, + ] + + +def test_remove_connections_keeps_the_rest(): + client = _client([BQ1, BQ2, BQ3]) + StandardLineage(client).remove_connections(SLUG, [BQ2]) + assert json.loads(_sent_inputs(client)["cross_connection_qualified_names"]) == [ + BQ1, + BQ3, + ] + + +def test_remove_connections_not_in_scope_is_a_noop(): + client = _client([BQ1, BQ2]) + assert StandardLineage(client).remove_connections(SLUG, [BQ3]) is None + client.app.update.assert_not_called() + + +def test_remove_last_connection_is_refused(): + client = _client([BQ1]) + with pytest.raises(ValueError, match="at least one connection"): + StandardLineage(client).remove_connections(SLUG, [BQ1]) + client.app.update.assert_not_called() diff --git a/tests/unit/test_app_builders.py b/tests/unit/test_app_builders.py index 3d90ec7a4..723e8fc99 100644 --- a/tests/unit/test_app_builders.py +++ b/tests/unit/test_app_builders.py @@ -158,7 +158,13 @@ def test_service_account_credential_shape(): def test_workload_identity_federation_auth_type(): - b = BigqueryCrawler(Mock()).workload_identity_federation(project_id="proj") + b = BigqueryCrawler(Mock()).workload_identity_federation( + project_id="proj", + service_account_email="svc@proj.iam.gserviceaccount.com", + wif_pool_provider_id="pool/provider", + atlan_oauth_id="oauth-id", + atlan_oauth_secret="oauth-secret", + ) cred = b._raw_creds["credential_guid"] assert cred.auth_type == "gcp-wif" assert cred.extras["project_id"] == "proj" @@ -243,7 +249,126 @@ def test_miner_auto_resolves_connection_credential(client): SnowflakeMiner(client).connection(qualified_name="default/snowflake/123").create() assert client.asset.search.called # connection was looked up out = client.app.create.call_args.kwargs["inputs"].to_inputs() - assert out["credential_guid"] == "conn-cred-guid" # its credential reused + # CONNECT-843: the reused guid rides on the connection entity (the UI's wire + # shape), never as a bare top-level credential_guid. A top-level guid with no + # credential body makes the create endpoint rewrite that credential's shared + # config record. Do not "fix" this back to out["credential_guid"]. + attrs = out["connection"]["attributes"] + assert attrs["defaultCredentialGuid"] == "conn-cred-guid" # its credential reused + assert out["credential_guid"] == "" # and not duplicated at the top level + + +# --------------------------------------------------------------------------- # +# CONNECT-843: where a REUSED credential guid is allowed to ride +# +# Reusing an already-vaulted guid on an existing connection must send the guid +# on the connection entity (the UI's wire shape) and leave top-level +# credential_guid "". A bare top-level guid with no credential body makes the +# create endpoint rewrite that credential's shared config record down to +# {"credentialSource": "direct"}, breaking every workflow sharing the guid. +# --------------------------------------------------------------------------- # +def _resolve_to(client, guid): + """Make the connection lookup in _create() resolve to ``guid``.""" + client.asset.search.return_value = iter([Mock(default_credential_guid=guid)]) + + +@pytest.mark.parametrize( + "cls, connector", + [(apps.BigqueryMiner, "bigquery"), (apps.SnowflakeMiner, "snowflake")], + ids=["bigquery", "snowflake"], +) +def test_auto_resolved_guid_rides_on_connection_not_top_level(client, cls, connector): + # Generic across connectors: the base builder owns this, no connector + # overrides _build_connection/_assemble. + _resolve_to(client, "resolved-guid") + cls(client).connection(qualified_name=f"default/{connector}/123").create() + out = client.app.create.call_args.kwargs["inputs"].to_inputs() + attrs = out["connection"]["attributes"] + # exactly the UI's reuse shape: identity + the connection's own credential + assert attrs["defaultCredentialGuid"] == "resolved-guid" + assert attrs["qualifiedName"] == f"default/{connector}/123" + assert attrs["connectorName"] == connector + # the guid is NOT echoed at the top level, and no credential body is invented + assert out["credential_guid"] == "" + assert "credential" not in out + + +def test_explicit_guid_on_existing_connection_rides_on_connection(): + # Same routing when the caller supplies the guid itself instead of letting + # _create() resolve it. The trigger is "guid + existing connection", not + # "guid came from a lookup". + out = ( + apps.BigqueryMiner(Mock()) + .connection(qualified_name="default/bigquery/1700000000") + .credential_guid("caller-supplied-guid") + .preview() + ) + assert ( + out["connection"]["attributes"]["defaultCredentialGuid"] + == "caller-supplied-guid" + ) + assert out["credential_guid"] == "" + + +def test_explicit_guid_on_new_connection_stays_top_level(): + # Deliberately unchanged, and NOT safe. This shape is KNOWN to still trigger + # the CONNECT-843 server bug: the credential config record is keyed by the guid + # alone, so minting a new connection protects nothing -- a bare top-level guid + # with no credential body still flattens that guid's shared config record to + # {"credentialSource": "direct"} for every workflow using it. It is left as-is + # because rerouting it here would newly affect crawler-shaped apps, whose + # connection-attribute fallback is unverified (crawler forms carry an explicit + # credential-guid widget, so they may legitimately read the top-level field), + # and because the real fix for this branch is the server-side guard (see the + # heracles handoff on CONNECT-843). This test pins today's behaviour so any + # future reroute is a deliberate, evidenced change and not a drive-by. + out = ( + BigqueryCrawler(Mock()) + .connection(name="prod-bq") + .credential_guid("existing-guid") + .preview() + ) + assert out["credential_guid"] == "existing-guid" + assert "defaultCredentialGuid" not in out["connection"]["attributes"] + + +def test_staged_credential_on_existing_connection_keeps_vaulting_shape(client): + # A staged raw credential is still vaulted by the create endpoint from the + # `credential` key, even on an existing connection: nothing moves onto the + # connection and credential_guid stays "" for the server to fill in. + ( + BigqueryCrawler(client) + .workload_identity_federation( + project_id="proj", + service_account_email="svc@proj.iam.gserviceaccount.com", + wif_pool_provider_id="projects/1/locations/global/workloadIdentityPools/p/providers/pr", + atlan_oauth_id="oauth-id", + atlan_oauth_secret="oauth-secret", + ) + .connection(qualified_name="default/bigquery/1700000000") + .include({"proj": ["ds"]}) + .create() + ) + out = client.app.create.call_args.kwargs["inputs"].to_inputs() + assert out["credential"]["authType"] == "gcp-wif" + assert out["credential_guid"] == "" + assert "defaultCredentialGuid" not in out["connection"]["attributes"] + client.asset.search.assert_not_called() # a staged cred needs no lookup + + +def test_agent_mode_on_existing_connection_ignores_credential_guid(): + # The agent/SDR path returns before any credential routing, so neither field + # appears. Unchanged by CONNECT-843. + out = ( + apps.SnowflakeMiner(Mock()) + .agent({"name": "my-agent"}) + .connection(qualified_name="default/snowflake/123") + .credential_guid("should-be-ignored") + .preview() + ) + assert out["agent_json"] == {"name": "my-agent"} + assert "credential_guid" not in out + assert "defaultCredentialGuid" not in out["connection"]["attributes"] def test_agent_mode_uses_agent_json_not_credential(client): @@ -258,3 +383,198 @@ def test_agent_mode_uses_agent_json_not_credential(client): assert out["agent_json"] == {"name": "my-agent"} assert "credential" not in out assert "credential_guid" not in out + + +# --------------------------------------------------------------------------- # +# update() — load an existing workflow, change a field, full-replace +# --------------------------------------------------------------------------- # +def test_load_update_preserves_reinjects_and_references_credential(): + """load(slug)..update() re-injects connection_qualified_name, keeps + the existing credential (referenced, not rotated), normalizes string-typed + fields, drops runtime keys, and preserves everything else.""" + from types import SimpleNamespace + + client = Mock() + current = { + "connection": { + "typeName": "Connection", + "attributes": { + "qualifiedName": "default/bigquery/123", + "connectorName": "bigquery", + "name": "prod", + "adminRoles": ["role-1"], + }, + }, + "credential_guid": "cred-1", + "extraction_method": "direct", + "include_filter": {"^proj$": ["^old$"]}, + "control_config": {}, # object on read-back -> must normalize to "{}" + "user-id": "u1", + "workflow_id": "w1", # runtime keys -> must be dropped + "atlas_auth_type": "internal", # non-structural -> preserved + } + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace( + properties={"control_config": {"type": "string"}} + ) + client.app.update.return_value = Mock(version=2) + + BigqueryCrawler(client).load("slug-1").include({"proj": ["new"]}).update() + + kw = client.app.update.call_args.kwargs + inp = kw["inputs"] + assert kw["slug"] == "slug-1" and kw["entrypoint"] == "crawler" + assert inp["connection_qualified_name"] == "default/bigquery/123" # re-injected + # CONNECT-843: reused credential rides on the connection, never top-level, and + # is never re-vaulted (no rotation). + assert inp["connection"]["attributes"]["defaultCredentialGuid"] == "cred-1" + assert inp["credential_guid"] == "" and "credential" not in inp + assert inp["include_filter"] == '{"^proj$": ["^new$"]}' # changed + anchored + assert inp["control_config"] == "{}" # normalized object -> string + assert "user-id" not in inp and "workflow_id" not in inp # runtime keys dropped + assert inp["atlas_auth_type"] == "internal" # preserved + assert inp["connection"]["attributes"]["adminRoles"] == [ + "role-1" + ] # connection kept + + +def test_update_without_load_raises(): + with pytest.raises(ValueError): + BigqueryCrawler(Mock()).update() + + +@pytest.mark.parametrize("cls", BUILDERS, ids=BUILDER_IDS) +def test_load_update_is_generic_across_builders(cls): + """load()/update() lives on the base, so every app builder can update: it + re-injects connection_qualified_name and references the existing credential + (no rotation), targeting the right slug/entrypoint. The builder's own + preview() is used as the 'current inputs' so the seed is valid for its + typed inputs model.""" + from types import SimpleNamespace + + qn = f"default/{cls._CONNECTOR_NAME}/1700000000" + current = ( + cls(Mock()).connection(qualified_name=qn).credential_guid("cred-x").preview() + ) + client = Mock() + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace(properties={}) + client.app.update.return_value = Mock(version=2) + + cls(client).load("slug-x").update() + + kw = client.app.update.call_args.kwargs + inp = kw["inputs"] + assert kw["slug"] == "slug-x" + assert kw["entrypoint"] == (cls._ENTRYPOINT or None) + assert inp["connection_qualified_name"] == qn # re-injected + # CONNECT-843: the guid rides on the connection (also proves load() reads it + # back from attributes.defaultCredentialGuid, not just top-level). + assert inp["connection"]["attributes"]["defaultCredentialGuid"] == "cred-x" + assert inp["credential_guid"] == "" # referenced, not rotated, not top-level + assert "credential" not in inp # no raw credential re-sent + + +def test_update_retries_without_entrypoint_on_1003(): + """update() falls back to the default entrypoint when the named one has no + registered contract (server 1003) — mirroring _create().""" + from types import SimpleNamespace + + current = ( + BigqueryCrawler(Mock()) + .connection(qualified_name="default/bigquery/1") + .credential_guid("g") + .preview() + ) + client = Mock() + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace(properties={}) + client.app.update.side_effect = [ + Exception("Server responded 1003: unknown entrypoint"), + Mock(version=3), + ] + + BigqueryCrawler(client).load("slug-x").update() + + assert client.app.update.call_count == 2 + assert client.app.update.call_args_list[0].kwargs["entrypoint"] == "crawler" + assert client.app.update.call_args_list[1].kwargs["entrypoint"] is None + + +def test_load_update_applies_multiple_field_changes(): + """load(slug)..update() applies every change and + preserves + re-injects the rest (connection_qualified_name, credential ref, + hidden defaults), dropping runtime keys.""" + from types import SimpleNamespace + + current = { + "connection": { + "typeName": "Connection", + "attributes": { + "qualifiedName": "default/bigquery/1", + "connectorName": "bigquery", + "name": "prod", + "adminRoles": ["role-1"], + }, + }, + "credential_guid": "cred-1", + "extraction_method": "direct", + "include_filter": {"^p$": ["^old$"]}, + "exclude_filter": {}, + "temp_table_regex": "", + "enable_nested_columns": True, + "enable_bigquery_tag_sync": False, + "filter_sharded_tables": True, + "hidden_datasets": False, + "control_config": {}, + "control_config_strategy": "default", + "atlas_auth_type": "internal", + "user-id": "u1", + } + client = Mock() + client.app.get.return_value.dict.return_value = { + "dag": {"extract": {"inputs": {"args": current}}} + } + client.app.get_input_contract.return_value = SimpleNamespace( + properties={"control_config": {"type": "string"}} + ) + client.app.update.return_value = Mock(version=2) + + ( + BigqueryCrawler(client) + .load("slug-1") + .include({"p": ["a", "b"]}) + .exclude({"p": ["tmp"]}) + .exclude_regex(".*_bak$") + .import_nested_columns(False) + .import_tags(True) + .combine_sharded_tables(False) + .hidden_assets(True) + .custom_config('{"flag": 1}') + .update() + ) + + inp = client.app.update.call_args.kwargs["inputs"] + # every requested change applied + assert inp["include_filter"] == '{"^p$": ["^a$", "^b$"]}' + assert inp["exclude_filter"] == '{"^p$": ["^tmp$"]}' + assert inp["temp_table_regex"] == ".*_bak$" + assert inp["enable_nested_columns"] is False + assert inp["enable_bigquery_tag_sync"] is True + assert inp["filter_sharded_tables"] is False + assert inp["hidden_datasets"] is True + assert inp["control_config_strategy"] == "custom" + assert inp["control_config"] == '{"flag": 1}' + # rest preserved / re-injected / stripped + assert inp["connection_qualified_name"] == "default/bigquery/1" + # CONNECT-843: reused credential on the connection, top-level empty, no rotation + assert inp["connection"]["attributes"]["defaultCredentialGuid"] == "cred-1" + assert inp["credential_guid"] == "" and "credential" not in inp + assert inp["atlas_auth_type"] == "internal" + assert "user-id" not in inp diff --git a/tests/unit/test_app_generated_inputs.py b/tests/unit/test_app_generated_inputs.py index 06888672c..ed6d5757c 100644 --- a/tests/unit/test_app_generated_inputs.py +++ b/tests/unit/test_app_generated_inputs.py @@ -88,6 +88,9 @@ def test_no_internal_fields_leak(cls): "BigqueryCrawlerInputs", "DatabricksCrawlerInputs", "KafkaConfluentInputs", + # Standard Lineage re-scopes an EXISTING workflow and JSON-encodes a list into a + # contract field declared `str`; neither is expressible from a configmap. + "StandardLineageInputs", }