From 13a85ef167cc05616fae4fb6ae01a921c3e0105e Mon Sep 17 00:00:00 2001 From: Volv G Date: Thu, 24 Sep 2026 19:19:44 -0700 Subject: [PATCH] Read config values from the environment: _env directive, EnvField tier, and _select/_env in pipeline cfg (v0.1.20) *(AI-assisted)* --- README.md | 61 +- .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../src/tangle_cli/args_container.py | 517 ++++++++++++++-- .../src/tangle_cli/pipeline_compiler.py | 109 +++- .../src/tangle_cli/python_pipeline/cfg.py | 110 +++- pyproject.toml | 2 +- tests/test_args_container.py | 4 +- tests/test_args_container_env.py | 562 ++++++++++++++++++ tests/test_packaging.py | 2 +- tests/test_pipeline_cfg_select_env.py | 463 +++++++++++++++ uv.lock | 2 +- 11 files changed, 1728 insertions(+), 106 deletions(-) create mode 100644 tests/test_args_container_env.py create mode 100644 tests/test_pipeline_cfg_select_env.py diff --git a/README.md b/README.md index bf638d2..63889ab 100644 --- a/README.md +++ b/README.md @@ -57,7 +57,7 @@ API-backed commands commonly accept these options. Explicit CLI options win over | `--token`, `TANGLE_API_TOKEN` | Bearer token shorthand. | | `--auth-header`, `TANGLE_API_AUTH_HEADER`, `TANGLE_AUTH_HEADER` | Full `Authorization` value such as `Bearer ...` or `Basic ...`. | | `-H`, `--header`, `TANGLE_API_HEADERS` | Extra headers. Repeatable as CLI flags; env accepts a JSON object or newline-separated `Name: value` entries. | -| `--config` | YAML/JSON defaults. Many commands accept a single object, a list of objects, or `_defaults` + `configs`, optionally wrapped in a top-level `_select` environment selector. | +| `--config` | YAML/JSON defaults. Many commands accept a single object, a list of objects, or `_defaults` + `configs`, optionally wrapped in a top-level `_select` environment selector. Values may be read from the environment with `{_env: NAME}`. | | `--log-type` | SDK progress logs: `console`, `none`, or `file`. Logs go to stderr or a temp log file so structured stdout stays parseable. | | `TANGLE_VERBOSE=1` | Redacted HTTP request/response diagnostics only. This is separate from normal progress logging. | | `--ca-bundle` | Global CLI flag: path to a PEM CA bundle used as the TLS trust store for every transport. Overrides `TANGLE_API_CA_BUNDLE`. Place before the subcommand. | @@ -232,7 +232,7 @@ For generated `tangle api` commands, config keys use generated CLI parameter nam ### Environment-selected configs (`_select`) -Any command that accepts `--config` can pick one of several config documents from an environment variable by making `_select` the top-level node: +Any command that accepts `--config`, and any Python pipeline `config.yaml` (see [Environment-selected pipeline config](#environment-selected-pipeline-config-_select-_env)), can pick one of several config documents from an environment variable by making `_select` the top-level node: ```yaml _shared: &shared @@ -271,7 +271,7 @@ With `default`, an unset variable or a value matching no case resolves to that b Rules: -- `_select` is the only newly reserved key, and it is an exact key name rather than a prefix. Configs without `_select` are unchanged byte-for-byte and semantically. +- `_select` is an exact reserved key name rather than a prefix (the only other reserved key is the [`_env`](#environment-variable-values-_env) value directive). Configs without `_select` are unchanged byte-for-byte and semantically. - At a `_select` node, only `_select` and other underscore-prefixed helper keys (YAML anchor holders such as `_shared`) may appear; ordinary sibling keys are rejected. - `_select` accepts only `env`, `cases`, and the optional `default`. There are no aliases: `else`, `fallback`, and `defaults` are rejected. - The selector shape, the `env` name, every `cases` key, and every case and `default` branch are validated before the environment is read, so a malformed selector fails identically in every environment. Branches are checked as complete config documents with the same rules the loader applies to a whole file, and nested selectors are validated recursively. @@ -280,6 +280,41 @@ Rules: - Matching uses `os.environ[NAME]` exactly: case sensitive, with no trimming, case folding, or interpolation. - Fallback exists only where it is authored. Without `default`, an unset variable or an unmatched value is an error; there is never an implicit default or implicit production branch. The raw environment value is never echoed — diagnostics list only the configured case names. +### Environment-variable values (`_env`) + +Any config value — in `--config` files and in [Python pipeline configs](#environment-selected-pipeline-config-_select-_env) — can be read from an environment variable with the `_env` value directive, so a checked-in config can reference a secret or per-machine value without containing it: + +```yaml +base_url: https://api.prod +token: {_env: TANGLE_PROD_TOKEN} +header: + - {_env: GATEWAY_HEADER} + - "X-Team: search" +limit: {_env: RUN_LIMIT, default: 10} +``` + +- The directive is exactly `{_env: NAME}` or `{_env: NAME, default: }`. Any other key beside `_env` — including underscore-prefixed keys and aliases such as `fallback` — is rejected, and `NAME` follows the same rule as `_select.env` (`[A-Za-z_][A-Za-z0-9_]*`). +- It may appear anywhere a value may appear: under a config key, in `_defaults`, in `configs` entries, and inside nested maps and lists. A document or config-entry mapping is never itself a directive, so an `_env:` helper/anchor key at the top level of a config object is unaffected. +- A missing variable without `default` fails closed with the variable name, the key path (for example `configs[1].token`), and the config file. An empty string counts as set and is used as-is. +- The resolved value is always a string, and a `default` is stringified the same way: numbers and booleans use their JSON spelling (`10`, `1.5`, `true`), dates use ISO format, and `null`/maps/lists are rejected (quote `''` for an empty default). A field therefore has one type whether or not the variable is set, and the command's usual conversion (JSON fields, repeatable options, enums, typed converters) applies downstream in both cases. +- Diagnostics name the variable, never its value, and no directive value is logged. There is no `${VAR}` string interpolation. + +With `_select`, selection happens first, and `_env` applies to the selected document: + +```yaml +_select: + env: TANGLE_ENV + cases: + prod: + token: {_env: TANGLE_PROD_TOKEN} + dev: + token: {_env: TANGLE_DEV_TOKEN, default: dev-token} +``` + +Every `_env` directive — in every case and `default` branch, and in helper sections — is shape-checked before any variable is read, so a malformed directive fails identically in every environment. Only the directives in the selected document's config entries and `_defaults` are looked up; a dormant branch never requires its variables. `_select` itself is unchanged. + +Precedence per field is **CLI > config > environment > default**: an explicit CLI value wins, then the config key (including a value read through `_env`), then an environment variable the command has opted that field into (see [`EnvField`](#shared-cli-helpers-and-logging)), then the default. As before, a CLI value equal to the option default is indistinguishable from an omitted one. + ## API schema cache and dynamic commands Refresh the local schema cache for a live backend with: @@ -509,6 +544,24 @@ def greeting_pipeline(who: In[str], cfg) -> Out[str]: `In[T]` parameters become runtime graph inputs. A single `-> Out[T]` return exposes one graph output; use `@pipeline(output_name=...)` to name that output. For multiple outputs, define a frozen dataclass subclass of `Outputs` with `Out[T]` fields and return an instance. A pipeline that accepts a `cfg` parameter reads `config.yaml` (or the path passed via `@pipeline(config="...")`) at compile time, with `--override key=value` values overlaid by the compile command. +##### Environment-selected pipeline config (`_select`, `_env`) + +A pipeline `config.yaml` — root or child, including one broadcast with `propagate_config=True` — is resolved by the same loader as `--config` files (see [`_select`](#environment-selected-configs-_select) and [`_env`](#environment-variable-values-_env)), with one shape rule: the selected document, and every case and `default` branch, must be a mapping. `_defaults`/`configs` have no special meaning in a pipeline config; they are ordinary keys. + +```yaml +_select: + env: TANGLE_ENV + cases: + prod: {dataset: prod_ds, batch_size: 500, token: {_env: PROD_TOKEN}} + dev: {dataset: dev_ds, batch_size: 10} + default: {dataset: local_ds, batch_size: 1} +``` + +- Resolution happens first; overrides are layered on the result. Precedence is `--override` / `.override_config` / `propagate_config` broadcast > selected branch. An `.override_config` key must exist in the child's *selected* branch, and a broadcast carries resolved values, never `_select`/`_env` nodes. Each child resolves its own config. +- `_select` branch values keep their native YAML types (`batch_size` is an `int`). `_env` values are strings (a `default` is stringified) and are **not** YAML-coerced the way raw `--override` strings are: an environment value is opaque (`007` stays `"007"`, `no` stays `"no"`). Convert explicitly in pipeline code (`int(cfg.limit)`), or put typed values in `_select` branches. +- Every branch is structure-checked in every environment, including the `template_file:` rejection, and a dormant branch never requires its variables. An unset or unmatched selector without `default` fails closed without echoing the value. +- Compile identity follows the resolved values. For a config that uses `_select` or `_env`, the child sidecar name (`-.yaml`) includes a digest of the resolved config, so different selections that produce different values never share a sidecar and identical values keep the same name. Configs without `_select`/`_env` keep their existing names. + Task IDs default from the left-hand variable name at the call site, converted to title case. If there is no simple left-hand variable, or if you want a stable explicit label, call `.named("Task Id")` before invoking the task. Use `.bind(...)` to pre-fill task arguments and `.with_annotations({...})` to add per-task annotations. ##### Root pipeline annotations @@ -968,6 +1021,8 @@ Use these for generic downstream behavior such as alternate storage, extra annot `cli_options.py` centralizes shared Cyclopts annotations such as `BaseUrlOption`, `TokenOption`, `AuthHeaderOption`, `HeaderOption`, `ConfigOption`, and `LogTypeOption`. `cli_helpers.py` centralizes config loading, JSON printing, credential-isolation helpers, and the native-safe `LazyTangleApiClient` proxy. `logger.py` provides `ConsoleLogger`, `NullLogger`, `CaptureLogger`, `logger_for_log_type(...)`, and `run_with_logging(...)`. +`ArgsContainer.load(...)` field specs are tuples (see `ArgsContainer._resolve`). To give one field an environment tier, wrap its unchanged spec: `token=EnvField("TANGLE_PROD_TOKEN", (token, None))`. The field then resolves CLI > config > `os.environ["TANGLE_PROD_TOKEN"]` > default; the raw string (empty counts as set) goes through the spec's usual converter, and a conversion error names the variable without echoing its value. Nothing is mapped automatically: fields that are not wrapped never read the environment. `args.origin(name)` reports where each field came from — `cli`, `config`, `env:NAME`, or `default` — without the value. + Use these helpers for new SDK commands so top-level imports remain native-free, `--config` behavior stays consistent, credentials from config do not accidentally mix with ambient environment auth, and progress logs stay off structured stdout. ## Development checks diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index 4eb9746..375f8b8 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.19" + __version__ = "0.1.20" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/args_container.py b/packages/tangle-cli/src/tangle_cli/args_container.py index e4a83e0..b169832 100644 --- a/packages/tangle-cli/src/tangle_cli/args_container.py +++ b/packages/tangle-cli/src/tangle_cli/args_container.py @@ -3,14 +3,20 @@ This module provides generic config-file behavior shared by Tangle CLI commands: load one or more config objects, merge each with parsed CLI arguments, and keep explicit CLI values higher precedence than config values. + +Precedence per field is CLI > config > environment (only for fields wrapped +in :class:`EnvField`) > default. Config values may themselves be read from +the environment with the ``{_env: NAME}`` value directive. """ from __future__ import annotations +import datetime import json import os import re from collections.abc import Callable +from dataclasses import dataclass from enum import Enum from pathlib import Path from typing import Any, cast @@ -24,6 +30,15 @@ #: not a prefix: other ``_``-prefixed keys stay available for YAML anchors. SELECT_KEY = "_select" +#: Key of the value directive ``{_env: NAME}`` / ``{_env: NAME, default: V}``. +#: Recognized only in value positions, never on a document or config-entry +#: mapping itself, so a top-level ``_env:`` helper/anchor key is unaffected. +ENV_KEY = "_env" +_ENV_DIRECTIVE_KEYS = (ENV_KEY, "default") +# Bounds recursion while rewriting a document that actually uses ``_env``. +_MAX_ENV_VALUE_DEPTH = 256 +_MAX_RENDERED_PATH_LENGTH = 160 + # Guards against recursive YAML aliases producing an endless selector chain. _MAX_SELECT_DEPTH = 32 _ENV_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") @@ -35,12 +50,37 @@ #: ``id(node) -> (greatest_depth_validated, node_pin, summary)``. Built per #: document load and thrown away with it; there is no global/cross-load cache. _ValidationMemo = dict[int, tuple[int, Any, _SelectorSummary]] +#: Raw key/index segments from a document root to a value, rendered lazily. +_ConfigPath = tuple[Any, ...] class ConfigFileError(Exception): """Raised when there is an error loading or resolving a config file.""" +def _is_env_name(name: Any) -> bool: + """Return whether *name* is a valid environment variable name. + + The single name rule shared by ``_select.env``, ``_env``, and + :class:`EnvField`. + """ + + return isinstance(name, str) and bool(_ENV_NAME_PATTERN.match(name)) + + +_ENV_NAME_RULE = "must be a valid environment variable name matching [A-Za-z_][A-Za-z0-9_]*" + + +def _scrub_config_key(key: Any) -> str: + """Scrub and length-cap a config key without quoting it.""" + + text = key if isinstance(key, str) else str(key) + rendered = "".join(char if char.isprintable() else "?" for char in text) + if len(rendered) > _MAX_RENDERED_CASE_KEY_LENGTH: + rendered = rendered[: _MAX_RENDERED_CASE_KEY_LENGTH - 3] + "..." + return rendered + + def _render_config_key(key: Any) -> str: """Render a config-document key safely for a diagnostic message. @@ -50,18 +90,201 @@ def _render_config_key(key: Any) -> str: callers can report a bad key without first proving it is a string. """ - text = key if isinstance(key, str) else str(key) - rendered = "".join(char if char.isprintable() else "?" for char in text) - if len(rendered) > _MAX_RENDERED_CASE_KEY_LENGTH: - rendered = rendered[: _MAX_RENDERED_CASE_KEY_LENGTH - 3] + "..." - return repr(rendered) + return repr(_scrub_config_key(key)) + + +def _render_config_path(path: _ConfigPath) -> str: + """Render a key path such as ``configs[0].token`` for a diagnostic. + + Every key is scrubbed like :func:`_render_config_key`; an overlong path + keeps its tail, which is the part that locates the value. + """ + + rendered = "" + for segment in path: + if isinstance(segment, int) and not isinstance(segment, bool): + rendered += f"[{segment}]" + else: + rendered += ("." if rendered else "") + _scrub_config_key(segment) + rendered = rendered or "" + if len(rendered) > _MAX_RENDERED_PATH_LENGTH: + rendered = "..." + rendered[-(_MAX_RENDERED_PATH_LENGTH - 3) :] + return rendered + + +def _stringify_env_default(value: Any, location: str) -> str: + """Return the string form of an ``_env`` ``default`` scalar. + + A default is stringified so a field has one type whether or not the + variable is set; downstream converters then type both paths identically. + Numbers and booleans use their JSON spelling (``true``, ``5``, ``1.5``), + so JSON-typed fields parse a default back to the authored value; dates use + ISO format. ``null`` has no string form and is rejected rather than + guessed at (quote ``''`` for an empty string). + """ + + if isinstance(value, str): + return value + if isinstance(value, (bool, int, float)): + return json.dumps(value) + if isinstance(value, datetime.date): + return value.isoformat() + type_name = "null" if value is None else type(value).__name__ + raise ConfigFileError( + f"{ENV_KEY} at {location}: default must be a string, number, boolean, " + f"or date scalar, got {type_name}" + ) + + +def _is_container(node: Any) -> bool: + return isinstance(node, (dict, list)) + + +def _parse_env_directive(node: dict[Any, Any], path: _ConfigPath) -> tuple[str, str | None]: + """Validate one ``_env`` directive node; return ``(name, default)``. + + ``default`` is the stringified default, or ``None`` when none was + authored. Reads no environment variable. + """ + + location = _render_config_path(path) + unexpected = sorted( + _render_config_key(key) for key in node if key not in _ENV_DIRECTIVE_KEYS + ) + if unexpected: + raise ConfigFileError( + f"{ENV_KEY} at {location} allows only an optional 'default' beside it, " + f"got unexpected keys: {', '.join(unexpected)}" + ) + name = node[ENV_KEY] + if not _is_env_name(name): + raise ConfigFileError(f"{ENV_KEY} at {location} {_ENV_NAME_RULE}") + default = _stringify_env_default(node["default"], location) if "default" in node else None + return name, default + + +class _EnvValueResolver: + """Replace ``_env`` directives in config values with environment strings. + + Copy-on-write: a container without directives is returned as the same + object, and each shared (YAML alias) node is resolved once. A directive + reached through a recursive alias cannot be represented and is rejected. + Diagnostics name the variable and key path, never a value. + """ + + def __init__(self, context: str) -> None: + self._context = context + self._memo: dict[int, Any] = {} + self._in_progress: set[int] = set() + self._back_refs: set[int] = set() + + def resolve_config_object(self, obj: dict[str, Any], path: _ConfigPath) -> dict[str, Any]: + """Resolve every value of one config entry; the entry itself is not a value.""" + + resolved: dict[str, Any] = {} + changed = False + for key, value in obj.items(): + new_value = self._resolve_value(value, (*path, key), 1) + resolved[key] = new_value + changed = changed or new_value is not value + return resolved if changed else obj + + def _resolve_value(self, node: Any, path: _ConfigPath, depth: int) -> Any: + if not _is_container(node): + return node + node_id = id(node) + if node_id in self._memo: + return self._memo[node_id] + if node_id in self._in_progress: + self._back_refs.add(node_id) + return node + if depth > _MAX_ENV_VALUE_DEPTH: + raise ConfigFileError( + f"Config nesting at {_render_config_path(path)} exceeds " + f"{_MAX_ENV_VALUE_DEPTH} levels in a document that uses {ENV_KEY}" + ) + + result: Any + if isinstance(node, dict) and ENV_KEY in node: + result = self._lookup(cast(dict[Any, Any], node), path) + else: + self._in_progress.add(node_id) + try: + result = self._resolve_container(node, path, depth) + finally: + self._in_progress.discard(node_id) + if node_id in self._back_refs and result is not node: + raise ConfigFileError( + f"{ENV_KEY} cannot be used inside a recursive YAML alias " + f"(at {_render_config_path(path)})" + ) + self._memo[node_id] = result + return result + + def _resolve_container(self, node: Any, path: _ConfigPath, depth: int) -> Any: + changed = False + if isinstance(node, dict): + mapping = cast(dict[Any, Any], node) + resolved_dict: dict[Any, Any] = {} + for key, value in mapping.items(): + new_value = self._resolve_value(value, (*path, key), depth + 1) + resolved_dict[key] = new_value + changed = changed or new_value is not value + return resolved_dict if changed else mapping + items = cast(list[Any], node) + resolved_list: list[Any] = [] + for index, value in enumerate(items): + new_value = self._resolve_value(value, (*path, index), depth + 1) + resolved_list.append(new_value) + changed = changed or new_value is not value + return resolved_list if changed else items + + def _lookup(self, node: dict[Any, Any], path: _ConfigPath) -> str: + name, default = _parse_env_directive(node, path) + # An empty string is a set variable. + if name in os.environ: + return os.environ[name] + if default is not None: + return default + raise ConfigFileError( + f"Environment variable {name} is required by {ENV_KEY} at " + f"{_render_config_path(path)}{self._context} but is not set" + ) + + +@dataclass(frozen=True) +class EnvField: + """Opt one :meth:`ArgsContainer.load` field spec into the environment tier. + + ``token=EnvField("TANGLE_TOKEN", (cli_token, None))`` wraps any ordinary + tuple *spec* unchanged; the field then resolves CLI > config > + ``os.environ[env]`` > default. The raw string (empty counts as set) goes + through the spec's usual converter. No field reads the environment unless + wrapped, and no name is derived automatically. + """ + + env: str + spec: tuple[Any, ...] + + def __post_init__(self) -> None: + if not _is_env_name(self.env): + raise ValueError(f"EnvField.env {_ENV_NAME_RULE}") + spec: Any = self.spec + if not isinstance(spec, tuple) or not 1 <= len(cast(tuple[Any, ...], spec)) <= 6: + raise ValueError("EnvField.spec must be an ArgsContainer field-spec tuple") class ArgsContainer: """Container for resolved CLI arguments with config-file defaults.""" - def __init__(self, resolved: dict[str, Any], raw_config: dict[str, Any]): + def __init__( + self, + resolved: dict[str, Any], + raw_config: dict[str, Any], + origins: dict[str, str] | None = None, + ): self._config = raw_config + self._origins = dict(origins or {}) for key, value in resolved.items(): setattr(self, key, value) @@ -80,7 +303,21 @@ def get(self, key: str, cli_value: Any = None, cli_default: Any = None) -> Any: def to_dict(self) -> dict[str, Any]: """Return resolved public values as a dictionary.""" - return {key: value for key, value in vars(self).items() if key != "_config"} + return { + key: value + for key, value in vars(self).items() + if key not in ("_config", "_origins") + } + + def origin(self, name: str) -> str | None: + """Return where field *name* was resolved from, never its value. + + One of ``"cli"``, ``"config"``, ``"env:NAME"`` (the :class:`EnvField` + tier), or ``"default"``; ``None`` for an unknown field. A config value + read through an ``_env`` directive reports ``"config"``. + """ + + return self._origins.get(name) @staticmethod def _validate_branch_document( @@ -88,6 +325,8 @@ def _validate_branch_document( where: str, depth: int = 0, memo: _ValidationMemo | None = None, + *, + mapping_only: bool = False, ) -> None: """Validate a selector branch as a complete config document root. @@ -101,9 +340,13 @@ def _validate_branch_document( the business of the selected branch, later. *memo* makes YAML alias DAGs linear instead of exponential; see - :meth:`_validate_selector_node`. + :meth:`_validate_selector_node`. With *mapping_only* (pipeline ``cfg`` + documents) every branch must be a mapping and ``_defaults``/``configs`` + carry no meaning. """ + if mapping_only and not isinstance(branch, dict): + raise ConfigFileError(f"{where} must be a mapping, got {type(branch).__name__}") if isinstance(branch, dict): branch_dict = cast(dict[Any, Any], branch) if SELECT_KEY in branch_dict: @@ -112,9 +355,11 @@ def _validate_branch_document( f"{SELECT_KEY} nesting exceeded the maximum depth " f"of {_MAX_SELECT_DEPTH}" ) - ArgsContainer._validate_selector_node(branch_dict, depth + 1, memo) + ArgsContainer._validate_selector_node( + branch_dict, depth + 1, memo, mapping_only=mapping_only + ) return - if "configs" in branch_dict: + if "configs" in branch_dict and not mapping_only: defaults = branch_dict.get("_defaults", {}) if not isinstance(defaults, dict): raise ConfigFileError( @@ -150,6 +395,8 @@ def _validate_selector_node( node: dict[str, Any], depth: int = 0, memo: _ValidationMemo | None = None, + *, + mapping_only: bool = False, ) -> _SelectorSummary: """Structurally validate one ``_select`` node without reading the env. @@ -217,11 +464,8 @@ def _validate_selector_node( raise ConfigFileError(f"{SELECT_KEY} requires a 'cases' object") env_name = selector_dict["env"] - if not isinstance(env_name, str) or not _ENV_NAME_PATTERN.match(env_name): - raise ConfigFileError( - f"{SELECT_KEY}.env must be a valid environment variable name " - "matching [A-Za-z_][A-Za-z0-9_]*" - ) + if not _is_env_name(env_name): + raise ConfigFileError(f"{SELECT_KEY}.env {_ENV_NAME_RULE}") cases = selector_dict["cases"] if not isinstance(cases, dict): @@ -246,14 +490,18 @@ def _validate_selector_node( # Every branch document is shape-checked, not just the selected one, # so a malformed selector fails identically in every environment. ArgsContainer._validate_branch_document( - case_value, f"{SELECT_KEY} case {_render_config_key(case_key)}", depth, memo + case_value, + f"{SELECT_KEY} case {_render_config_key(case_key)}", + depth, + memo, + mapping_only=mapping_only, ) default_branch: Any = None if "default" in selector_dict: default_branch = selector_dict["default"] ArgsContainer._validate_branch_document( - default_branch, f"{SELECT_KEY}.default", depth, memo + default_branch, f"{SELECT_KEY}.default", depth, memo, mapping_only=mapping_only ) allowed = ", ".join(_render_config_key(key) for key in sorted(cases_dict)) @@ -265,7 +513,12 @@ def _validate_selector_node( return summary @staticmethod - def _select_branch(node: dict[str, Any], memo: _ValidationMemo | None = None) -> Any: + def _select_branch( + node: dict[str, Any], + memo: _ValidationMemo | None = None, + *, + mapping_only: bool = False, + ) -> Any: """Validate one ``_select`` node and return the branch chosen by the env. The whole selector shape, the environment variable name, every @@ -284,7 +537,7 @@ def _select_branch(node: dict[str, Any], memo: _ValidationMemo | None = None) -> """ env_name, cases_dict, default_branch, allowed = ArgsContainer._validate_selector_node( - node, 0, memo + node, 0, memo, mapping_only=mapping_only ) # Environment is read only after the selector itself is known to be valid. @@ -307,7 +560,12 @@ def _select_branch(node: dict[str, Any], memo: _ValidationMemo | None = None) -> return cases_dict[env_value] @staticmethod - def _resolve_select(parsed: Any) -> Any: + def _resolve_select( + parsed: Any, + memo: _ValidationMemo | None = None, + *, + mapping_only: bool = False, + ) -> Any: """Replace a root ``_select`` node with its selected config branch. A selected branch is a complete config document in its own right, so it @@ -317,10 +575,11 @@ def _resolve_select(parsed: Any) -> Any: The whole selector tree is structurally validated up front by :meth:`_select_branch`; this loop only walks the chain the environment actually selects, and keeps its own bound as a belt-and-braces guard. - The validation memo lives for exactly this one document resolution. + The validation memo lives for exactly this one document resolution; a + caller may pass the memo it already used to validate this document. """ - memo: _ValidationMemo = {} + memo = {} if memo is None else memo node = parsed selections = 0 while isinstance(node, dict) and SELECT_KEY in cast(dict[Any, Any], node): @@ -328,10 +587,77 @@ def _resolve_select(parsed: Any) -> Any: raise ConfigFileError( f"{SELECT_KEY} nesting exceeded the maximum depth of {_MAX_SELECT_DEPTH}" ) - node = ArgsContainer._select_branch(cast(dict[str, Any], node), memo) + node = ArgsContainer._select_branch( + cast(dict[str, Any], node), memo, mapping_only=mapping_only + ) selections += 1 return node + @staticmethod + def _validate_env_directives( + document: Any, *, mapping_only: bool = False + ) -> tuple[bool, list[Any]]: + """Structurally validate every ``_env`` directive; read no variable. + + Walks the whole raw document -- every ``_select`` case and ``default`` + branch (dormant ones included) and helper sections -- so a malformed + directive fails identically in every environment. Selector structure + must already be validated. Shared nodes are visited once per role. + Returns whether any directive exists, and every candidate document + root (each non-selector branch, dormant ones included). + """ + + found = False + branches: list[Any] = [] + seen: set[tuple[str, int]] = set() + stack: list[tuple[str, Any, _ConfigPath]] = [("document", document, ())] + + def push(role: str, items: list[tuple[Any, _ConfigPath]]) -> None: + # Reversed so diagnostics follow document order. + stack.extend((role, node, path) for node, path in reversed(items)) + + while stack: + role, node, path = stack.pop() + if not _is_container(node) or (role, id(node)) in seen: + continue + seen.add((role, id(node))) + if isinstance(node, list): + child_role = "value" if role == "value" else "entry" + if role != "entry": + items = cast(list[Any], node) + push(child_role, [(item, (*path, i)) for i, item in enumerate(items)]) + continue + + node_dict = cast(dict[Any, Any], node) + if role == "document" and SELECT_KEY not in node_dict: + branches.append(node_dict) + if role == "value" and ENV_KEY in node_dict: + _parse_env_directive(node_dict, path) + found = True + elif role == "document" and SELECT_KEY in node_dict: + selector = cast(dict[Any, Any], node_dict[SELECT_KEY]) + helpers = [(v, (*path, k)) for k, v in node_dict.items() if k != SELECT_KEY] + candidates = [ + (branch, (*path, SELECT_KEY, "cases", case)) + for case, branch in cast(dict[Any, Any], selector["cases"]).items() + ] + if "default" in selector: + candidates.append((selector["default"], (*path, SELECT_KEY, "default"))) + push("document", candidates) + push("value", helpers) + elif role == "document" and "configs" in node_dict and not mapping_only: + for key, value in reversed(list(node_dict.items())): + if key == "_defaults": + push("entry", [(value, (*path, key))]) + elif key == "configs" and isinstance(value, list): + push("document", [(value, (*path, key))]) + else: + push("value", [(value, (*path, key))]) + else: + # A config entry (or a plain-object document): its values are values. + push("value", [(v, (*path, k)) for k, v in node_dict.items()]) + return found, branches + @staticmethod def _load_config_file( config_path: str | Path | None, @@ -346,6 +672,11 @@ def _load_config_file( A document may instead select one of those shapes at load time with a top-level ``_select`` node (see :meth:`_select_branch`). + + ``{_env: NAME}`` / ``{_env: NAME, default: V}`` value directives are + then replaced by the variable's string value. Every directive is shape + checked first, but only those in the selected document's config + entries (and ``_defaults``) are looked up. """ log = logger or get_default_logger() @@ -367,7 +698,9 @@ def _load_config_file( except (OSError, json.JSONDecodeError, yaml.YAMLError) as exc: raise ConfigFileError(f"Error loading config file: {exc}") from exc - parsed = ArgsContainer._resolve_select(parsed) + resolution = resolve_config_document(parsed, path) + parsed = resolution.document + resolve_entry = resolution.resolve_entry if isinstance(parsed, dict): parsed_dict = cast(dict[str, Any], parsed) @@ -388,12 +721,17 @@ def _load_config_file( "configs entry " f"{index} must be an object, got {type(item).__name__}" ) + defaults = resolve_entry(cast(dict[str, Any], defaults), ("_defaults",)) + configs_list = [ + resolve_entry(item, ("configs", index)) + for index, item in enumerate(cast(list[dict[str, Any]], configs_list)) + ] merged = apply_defaults(configs_list, defaults) assert isinstance(merged, list) log.info(f"Loaded config: {path} ({len(merged)} configs with defaults)") return merged log.info(f"Loaded config: {path} (1 config)") - return [parsed_dict] + return [resolve_entry(parsed_dict, ())] if isinstance(parsed, list): for index, item in enumerate(cast(list[Any], parsed)): @@ -402,7 +740,10 @@ def _load_config_file( "Config file entry " f"{index} must be an object, got {type(item).__name__}" ) - configs = cast(list[dict[str, Any]], parsed) + configs = [ + resolve_entry(item, (index,)) + for index, item in enumerate(cast(list[dict[str, Any]], parsed)) + ] log.info(f"Loaded config: {path} ({len(configs)} configs)") return configs @@ -436,18 +777,20 @@ def convert(value: Any) -> Any: @staticmethod def _make_enum_converter(field_name: str, enum_type: type[Enum]) -> Callable[[Any], Any]: - """Create a converter that accepts enum values by string.""" + """Create a converter that accepts enum values by string. + + The rejected value is not echoed (it may come from the environment). + """ def convert(value: Any) -> Any: if isinstance(value, str): try: return enum_type(value) - except ValueError as exc: + except ValueError: valid_values = [member.value for member in enum_type] raise ConfigFileError( - f"Invalid value '{value}' for {field_name}. " - f"Valid values: {valid_values}" - ) from exc + f"Invalid value for {field_name}. Valid values: {valid_values}" + ) from None return value return convert @@ -462,13 +805,25 @@ def _resolve(config: dict[str, Any], **kwargs: Any) -> ArgsContainer: - ``(cli_value, default, converter)``: optional with converter; - ``(config_key, cli_value, default, is_json)``: explicit key; - ``(config_key, cli_value, default, is_json, required)``; - - ``(config_key, cli_value, default, is_json, required, converter)``. + - ``(config_key, cli_value, default, is_json, required, converter)``; + - :class:`EnvField` wrapping any of the above to add the env tier. + + Precedence: an explicit CLI value (one differing from the spec + default), then the config key, then the ``EnvField`` variable, then + the CLI/default value. """ resolved: dict[str, Any] = {} + origins: dict[str, str] = {} required_fields: list[str] = [] + env_names: dict[str, str] = {} for param_name, spec in kwargs.items(): + env_name: str | None = None + if isinstance(spec, EnvField): + env_name = spec.env + env_names[param_name] = env_name + spec = spec.spec converter = None default_value = None if len(spec) == 1: @@ -502,21 +857,41 @@ def _resolve(config: dict[str, Any], **kwargs: Any) -> ArgsContainer: converter = ArgsContainer._make_enum_converter(param_name, type(default_value)) if cli_value is not None and cli_value != default_value: - value = cli_value + value, origin = cli_value, "cli" elif config_key in config: - value = config[config_key] + value, origin = config[config_key], "config" + elif env_name is not None and env_name in os.environ: + value, origin = os.environ[env_name], f"env:{env_name}" else: - value = cli_value - - resolved[param_name] = converter(value) if converter and value is not None else value + value, origin = cli_value, "default" + + if converter and value is not None: + if env_name is not None and origin == f"env:{env_name}": + try: + value = converter(value) + except (ConfigFileError, ValueError, TypeError): + # A converter message may quote the raw value. + raise ConfigFileError( + f"Invalid value for {param_name} from environment " + f"variable {env_name}" + ) from None + else: + value = converter(value) + resolved[param_name] = value + origins[param_name] = origin for field_name in required_fields: if resolved.get(field_name) is None: + if field_name in env_names: + raise ConfigFileError( + f"{field_name} is required (via CLI argument, config file, " + f"or environment variable {env_names[field_name]})" + ) raise ConfigFileError( f"{field_name} is required (via CLI argument or config file)" ) - return ArgsContainer(resolved, config) + return ArgsContainer(resolved, config, origins) @staticmethod def load( @@ -530,4 +905,68 @@ def load( return [ArgsContainer._resolve(config, **kwargs) for config in configs] -__all__ = ["SELECT_KEY", "ArgsContainer", "ConfigFileError"] +@dataclass(frozen=True) +class ResolvedConfigDocument: + """A config document after ``_select``, before its ``_env`` lookups. + + ``document`` is the selected document root; callers apply their own shape + rules and then :meth:`resolve_entry` to each config mapping they keep, so + only directives in used entries are looked up. ``branches`` holds every + candidate document root, dormant ones included, for fail-closed shape + checks. ``env_dependent`` is true when the result can depend on the + environment (a root ``_select`` or any ``_env``). + """ + + document: Any + branches: tuple[Any, ...] + env_dependent: bool + _resolver: _EnvValueResolver | None + + def resolve_entry(self, entry: dict[str, Any], path: _ConfigPath = ()) -> dict[str, Any]: + """Replace the ``_env`` directives in one config mapping's values.""" + + if self._resolver is None: + return entry + return self._resolver.resolve_config_object(entry, path) + + +def resolve_config_document( + parsed: Any, source: str | Path, *, mapping_only: bool = False +) -> ResolvedConfigDocument: + """Resolve a parsed config document's root ``_select`` chain and ``_env`` plan. + + The single document-level resolution shared by ``--config`` files and + pipeline ``cfg`` files: the selector tree and every ``_env`` directive are + structurally validated first (dormant branches included, no variable + read), then the selector chain is resolved against the environment. + *mapping_only* makes every branch a plain mapping (pipeline ``cfg``); + otherwise branches take the ``--config`` shapes. *source* labels + ``_env`` diagnostics. Raises :class:`ConfigFileError`. + """ + + memo: _ValidationMemo = {} + is_select = isinstance(parsed, dict) and SELECT_KEY in cast(dict[Any, Any], parsed) + if is_select: + ArgsContainer._validate_selector_node( + cast(dict[str, Any], parsed), 0, memo, mapping_only=mapping_only + ) + uses_env, branches = ArgsContainer._validate_env_directives( + parsed, mapping_only=mapping_only + ) + selected = ArgsContainer._resolve_select(parsed, memo, mapping_only=mapping_only) + resolver: _EnvValueResolver | None = None + if uses_env: + within = " (within the selected _select branch)" if is_select else "" + resolver = _EnvValueResolver(f" in {source}{within}") + return ResolvedConfigDocument(selected, tuple(branches), is_select or uses_env, resolver) + + +__all__ = [ + "ENV_KEY", + "SELECT_KEY", + "ArgsContainer", + "ConfigFileError", + "EnvField", + "ResolvedConfigDocument", + "resolve_config_document", +] diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py index 8b7040e..dd4cb6b 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py @@ -55,7 +55,14 @@ from .authenticated_identity import ME from .component_from_func import build_unwrapped_inputs_schema from .handler import TangleCliHandler -from .python_pipeline.cfg import Cfg, _coerce_override, load_cfg +from .python_pipeline.cfg import ( + Cfg, + CfgDocument, + _coerce_override, + build_cfg, + read_cfg_document, + reject_template_file, +) from .python_pipeline.compiler_context import ( BroadcastLayer, CompileContext, @@ -350,10 +357,12 @@ def _resolve_cfg_for_pipeline( *, is_root: bool, output_path: Path, -) -> tuple[Cfg, dict[str, Any], Path]: +) -> tuple[Cfg, dict[str, Any], Path, str | None]: """Resolve + load a pipeline's cfg relative to its own source dir. - Returns ``(cfg, raw_cfg, cfg_path)``. The config file is loaded when the + Returns ``(cfg, raw_cfg, cfg_path, config_identity)`` -- ``raw_cfg`` is + the ``_select``/``_env``-resolved mapping before overrides, and + ``config_identity`` is :func:`_config_identity` of it. The config file is loaded when the function takes a ``cfg`` parameter OR the pipeline is flagged ``propagate_config`` (which must read its own config to broadcast it), and otherwise skipped — a pipeline with neither tolerates a missing config.yaml. @@ -371,7 +380,7 @@ def _resolve_cfg_for_pipeline( pipeline_fn=pipeline_fn, is_root=is_root, ) - loaded_cfg, raw_cfg = _load_cfg_and_raw( + loaded_cfg, raw_cfg, config_identity = _load_cfg_and_raw( cfg_path, overrides, coerce=is_root, @@ -387,7 +396,7 @@ def _resolve_cfg_for_pipeline( "subpipelines. Add a `cfg` parameter if this pipeline also needs to " "read the config directly." ) - return cfg, raw_cfg, cfg_path + return cfg, raw_cfg, cfg_path, config_identity if is_root and overrides: keys = sorted(overrides) @@ -405,7 +414,7 @@ def _resolve_cfg_for_pipeline( "not loaded. Remove `config=` when no compile-time config is needed, " "or add a `cfg` parameter to use it." ) - return Cfg({}), {}, cfg_path + return Cfg({}), {}, cfg_path, None def _compile_pipeline_fn( @@ -457,7 +466,7 @@ def _compile_pipeline_fn( # The ROOT coerces its CLI ``--override`` strings (yaml.safe_load); a # CHILD receives already-typed native ``effective`` overrides (broadcast # + explicit ``.override_config``) that must pass through unchanged. - cfg, raw_cfg, cfg_path = _resolve_cfg_for_pipeline( + cfg, raw_cfg, cfg_path, config_identity = _resolve_cfg_for_pipeline( pipeline_fn, base_dir, overrides, @@ -490,11 +499,12 @@ def _compile_pipeline_fn( # the key the parent tests with ``child_key in active_stack`` — a real # cycle under ``propagate_config`` would then escape precise detection # and degrade to the max-depth guard. The root has no parent (and no - # ambient), so it builds its own key from its own overrides. + # ambient), so it builds its own key from its own overrides and its + # resolved config identity. if precomputed_key is not None: key = precomputed_key else: - key = _compile_key_for(pipeline_fn, cfg_path, overrides) + key = _compile_key_for(pipeline_fn, cfg_path, overrides, config_identity=config_identity) # 4. @task local_from_python sidecar for THIS artifact (built, not # written). Each @task ref is rewritten to a pure @@ -699,6 +709,15 @@ def _process_subpipeline_children( # sidecars) whenever the difference can affect its descendants — even # when the child's OWN effective overrides are identical. ambient_passthrough: dict[str, Any] = {} + # A child that will load its config folds the resolved config's + # identity into its key, so two environment selections of the same + # file never share a sidecar (None, and no key change, without + # ``_select``/``_env``). + child_doc: CfgDocument | None = None + child_identity: str | None = None + if (_pipeline_accepts_cfg(child_fn) or child_fn.propagate_config) and child_cfg_path.exists(): + child_doc = read_cfg_document(child_cfg_path) + child_identity = _config_identity(child_doc) if ctx.broadcast_stack or sub_ref.config_overrides: # A config-less child (no ``cfg`` param, no ``config=``) has no # ``config.yaml`` on disk. Under an active broadcast (or an @@ -711,7 +730,10 @@ def _process_subpipeline_children( # broadcasts. A config-DECLARING child whose file is genuinely # missing is still caught, with guidance, later in # ``_load_cfg_and_raw`` (via the child's own ``_compile_pipeline_fn``). - child_raw = _read_raw_cfg(child_cfg_path) if child_cfg_path.exists() else {} + if child_doc is not None: + child_raw = child_doc.data + else: + child_raw = _read_raw_cfg(child_cfg_path) if child_cfg_path.exists() else {} effective = _effective_overrides_for_child( child_raw=child_raw, broadcast_stack=ctx.broadcast_stack, @@ -735,6 +757,7 @@ def _process_subpipeline_children( effective, fingerprint_context=f"on subpipeline task {task_id!r} -> child pipeline {child_fn.name!r}", ambient_context=ambient_passthrough or None, + config_identity=child_identity, ) # Cycle detection (covers self-reference). active_stack holds the @@ -983,6 +1006,7 @@ def _compile_key_for( *, fingerprint_context: str | None = None, ambient_context: Mapping[str, Any] | None = None, + config_identity: str | None = None, ) -> PipelineCompileKey: """Build the canonical :class:`PipelineCompileKey` for ``pipeline_fn``. @@ -1013,20 +1037,26 @@ def _compile_key_for( config keys, routed through :func:`overrides_fingerprint` so its non-serializable-value -> :class:`CompileError` detection (which recurses into dicts) still fires for BOTH effective and ambient values. + * **``config_identity``** (only for a config using ``_select``/``_env``, + see :func:`_config_identity`) -> the same envelope plus + ``"\x00config"``, so different environment selections get distinct + keys while identical resolved configs keep identical keys. """ src = _pipeline_source_path(pipeline_fn) if src is not None: source_canon = canonical_repo_path(src) else: source_canon = f"" - if ambient_context: + if ambient_context or config_identity is not None: # Reserved keys (NUL-prefixed) can never be real config keys, so an # envelope never collides with an effective-only fingerprint over the # same keys (e.g. effective={x:1},ambient={} vs effective={x:1},ambient={y:2}). - fingerprint_input: Mapping[str, Any] = { - "\x00effective": dict(sorted(overrides.items())), - "\x00ambient": dict(sorted(ambient_context.items())), - } + envelope: dict[str, Any] = {"\x00effective": dict(sorted(overrides.items()))} + if ambient_context: + envelope["\x00ambient"] = dict(sorted(ambient_context.items())) + if config_identity is not None: + envelope["\x00config"] = config_identity + fingerprint_input: Mapping[str, Any] = envelope else: # Byte-identical to today's key: the default-isolation guarantee. fingerprint_input = overrides @@ -2618,18 +2648,19 @@ def _load_cfg_and_raw( pipeline_fn: PipelineFn, usage: str = "cfg", ): - """Load cfg via ``cfg.load_cfg`` AND read the raw YAML dict. + """Load cfg AND return the resolved config dict and its identity. - The raw dict (config.yaml WITHOUT overrides applied) is returned so the - caller can build a broadcast layer from this pipeline's OWN config. + The config file is read once through :func:`read_cfg_document`, so + ``_select`` / ``_env`` are resolved before overrides overlay it. The + resolved dict (WITHOUT overrides applied) is returned so the caller can + build a broadcast layer from this pipeline's OWN config; it never holds a + ``_select`` or ``_env`` node. ``_config_identity`` of it is returned too. ``coerce`` is threaded to :func:`load_cfg`: the ROOT passes ``coerce=True`` so its CLI ``--override`` strings keep YAML coercion; a CHILD passes ``coerce=False`` so its already-typed native ``effective`` overrides pass through unchanged. Overrides are applied to the :class:`Cfg` object only — the compiled YAML carries no config keys. """ - import yaml - if not cfg_path.exists(): source = ( f"@pipeline(config={pipeline_fn.config_path!r})" @@ -2653,11 +2684,29 @@ def _load_cfg_and_raw( raise CompileError( f"config file not found: {cfg_path}. Pipeline {pipeline_fn.name!r} {reason}" ) - raw = yaml.safe_load(cfg_path.read_text()) or {} - if not isinstance(raw, dict): - raise CompileError(f"config at {cfg_path} must be a YAML mapping, got {type(raw).__name__}") - cfg = load_cfg(cfg_path, overrides=dict(overrides), coerce=coerce) - return cfg, raw + document = read_cfg_document(cfg_path) + reject_template_file(document, cfg_path) + cfg = build_cfg(document.data, overrides=dict(overrides), coerce=coerce) + return cfg, document.data, _config_identity(document) + + +def _config_identity(document: CfgDocument) -> str | None: + """Digest of a resolved config for compile keys, or ``None``. + + ``None`` unless the file uses ``_select``/``_env`` -- then the compile key + (and sidecar name) is byte-identical to a build without them. Otherwise a + SHA-256 of the resolved content: environment selections yielding different + values differ, identical values stay identical. The digest is never + printed; only :meth:`PipelineCompileKey.hash8` of the whole key reaches a + sidecar file name. + """ + if not document.env_dependent: + return None + try: + canonical = yaml.safe_dump(document.data, sort_keys=True, allow_unicode=True) + except TypeError: # keys of mixed, unorderable types + canonical = yaml.safe_dump(document.data, sort_keys=False, allow_unicode=True) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() def _read_raw_cfg(cfg_path: Path) -> dict[str, Any]: @@ -2667,16 +2716,12 @@ def _read_raw_cfg(cfg_path: Path) -> dict[str, Any]: :func:`_effective_overrides_for_child` — both the broadcast same-name overlay and the explicit ``.override_config`` typo check key off the keys the child actually declares. Mirrors the error style of - :func:`_load_cfg_and_raw` (missing file / non-mapping config). + :func:`_load_cfg_and_raw` (missing file / non-mapping config). Keys are + those of the resolved (``_select``-selected) mapping. """ - import yaml - if not cfg_path.exists(): raise CompileError(f"config file not found: {cfg_path}") - raw = yaml.safe_load(cfg_path.read_text()) or {} - if not isinstance(raw, dict): - raise CompileError(f"config at {cfg_path} must be a YAML mapping, got {type(raw).__name__}") - return raw + return read_cfg_document(cfg_path).data def _resolve_broadcast_stack( diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/cfg.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/cfg.py index b44a334..76fdda4 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/cfg.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/cfg.py @@ -27,11 +27,17 @@ version='"1.0"'``. - ``template_file:`` in the source is rejected — the Python authoring layer IS the template authoring layer. +- The file is resolved like a ``--config`` document first: a root + ``_select`` picks one mapping branch (native YAML types) and + ``{_env: NAME}`` values become strings, which are NOT YAML-coerced (an + environment value is opaque; only raw CLI ``--override`` strings are). + Overrides then overlay the resolved mapping. """ from __future__ import annotations from collections.abc import Mapping +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -99,9 +105,84 @@ def __getattr__(self, key: str) -> Any: raise UnknownCfgKeyError(f"unknown config key: {key!r}. Available keys: " f"{sorted(data.keys())!r}") +@dataclass(frozen=True) +class CfgDocument: + """A pipeline config file resolved to one mapping, before overrides. + + ``data`` has ``_select`` resolved and ``_env`` values substituted. + ``branches`` are every candidate root mapping (dormant ``_select`` + branches included); ``env_dependent`` is true when ``data`` can differ + between environments. + """ + + data: dict[str, Any] + branches: tuple[dict[str, Any], ...] + env_dependent: bool + + +def read_cfg_document(path: Path) -> CfgDocument: + """Read and resolve the config file at ``path`` into one mapping. + + Shares document resolution with ``--config`` files + (:func:`tangle_cli.args_container.resolve_config_document`), with the + ``cfg`` shape rule that every branch is a mapping. Applies no overrides + and no ``template_file`` check (see :func:`reject_template_file`). + """ + from tangle_cli.args_container import ConfigFileError, resolve_config_document + + parsed = yaml.safe_load(Path(path).read_text()) or {} + try: + resolution = resolve_config_document(parsed, path, mapping_only=True) + document = resolution.document + if not isinstance(document, dict): + raise CompileError(f"config at {path} must be a YAML mapping, got {type(document).__name__}") + data = resolution.resolve_entry(document) + except ConfigFileError as exc: + raise CompileError(f"config at {path}: {exc}") from exc + return CfgDocument( + data=data, + branches=tuple(resolution.branches), + env_dependent=resolution.env_dependent, + ) + + +def reject_template_file(document: CfgDocument, path: Path) -> None: + """Reject a top-level ``template_file:`` in any branch, dormant included.""" + if any("template_file" in branch for branch in document.branches): + raise CompileError( + f"config at {path} has a top-level `template_file:` key, but " + "the Python authoring layer emits the pipeline directly. Remove " + "this key from your config.yaml — the framework is your " + "templating layer." + ) + + +def build_cfg(data: Mapping[str, Any], overrides: Mapping[str, Any] | None = None, *, coerce: bool = True) -> Cfg: + """Overlay ``overrides`` on a resolved config mapping; see :func:`load_cfg`.""" + merged = dict(data) + if overrides: + if coerce: + # Coerce each raw CLI override string to its YAML type so an + # override behaves identically to the same key written in the + # config file (which is already typed via yaml.safe_load). + # Coercion is internal to cfg only: the override dict stays + # dict[str, str] everywhere upstream and the compile-cache + # fingerprint keeps hashing the raw CLI strings. + merged.update({k: _coerce_override(v) for k, v in overrides.items()}) + else: + # Native pass-through: values are already-typed Python natives + # (from .override_config / broadcast). Overlay verbatim — no + # yaml.safe_load re-coercion (it would mangle e.g. "no" -> False). + merged.update(dict(overrides)) + return Cfg(merged) + + def load_cfg(path: Path, overrides: Mapping[str, Any] | None = None, *, coerce: bool = True) -> Cfg: """Load YAML at ``path`` and overlay ``overrides`` on top. + The file is first resolved by :func:`read_cfg_document` (``_select`` / + ``_env``); overrides always win over the SELECTED values. + Two overlay paths, selected by ``coerce``: * ``coerce=True`` (the default, the CLI ``--override`` path): override @@ -141,29 +222,6 @@ def load_cfg(path: Path, overrides: Mapping[str, Any] | None = None, *, coerce: config must not point at a separate Jinja template). FileNotFoundError: if ``path`` doesn't exist. """ - raw = yaml.safe_load(Path(path).read_text()) or {} - if not isinstance(raw, dict): - raise CompileError(f"config at {path} must be a YAML mapping, got {type(raw).__name__}") - if "template_file" in raw: - raise CompileError( - f"config at {path} has a top-level `template_file:` key, but " - "the Python authoring layer emits the pipeline directly. Remove " - "this key from your config.yaml — the framework is your " - "templating layer." - ) - merged = dict(raw) - if overrides: - if coerce: - # Coerce each raw CLI override string to its YAML type so an - # override behaves identically to the same key written in the - # config file (which is already typed via yaml.safe_load above). - # Coercion is internal to load_cfg only: the override dict stays - # dict[str, str] everywhere upstream and the compile-cache - # fingerprint keeps hashing the raw CLI strings. - merged.update({k: _coerce_override(v) for k, v in overrides.items()}) - else: - # Native pass-through: values are already-typed Python natives - # (from .override_config / broadcast). Overlay verbatim — no - # yaml.safe_load re-coercion (it would mangle e.g. "no" -> False). - merged.update(dict(overrides)) - return Cfg(merged) + document = read_cfg_document(path) + reject_template_file(document, path) + return build_cfg(document.data, overrides, coerce=coerce) diff --git a/pyproject.toml b/pyproject.toml index 41e2f02..1426452 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.19" +version = "0.1.20" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_args_container.py b/tests/test_args_container.py index 659c9f0..d1ca236 100644 --- a/tests/test_args_container.py +++ b/tests/test_args_container.py @@ -1305,13 +1305,13 @@ def selector_visit_budget(monkeypatch): original = ArgsContainer._validate_selector_node state = {"calls": 0, "cap": 10**9} - def counting(node, depth=0, memo=None): + def counting(node, depth=0, memo=None, **kwargs): state["calls"] += 1 if state["calls"] > state["cap"]: raise _ValidationBudgetExceeded( f"structural validation exceeded {state['cap']} selector visits" ) - return original(node, depth, memo) + return original(node, depth, memo, **kwargs) monkeypatch.setattr(ArgsContainer, "_validate_selector_node", staticmethod(counting)) return state diff --git a/tests/test_args_container_env.py b/tests/test_args_container_env.py new file mode 100644 index 0000000..ca74792 --- /dev/null +++ b/tests/test_args_container_env.py @@ -0,0 +1,562 @@ +"""Environment-sourced values: the ``_env`` config directive and ``EnvField``.""" + +from __future__ import annotations + +import json +import traceback +from enum import Enum + +import pytest + +import tangle_cli.args_container as args_container_module +from tangle_cli.api_transport import _header_entries +from tangle_cli.args_container import ArgsContainer, ConfigFileError, EnvField + +SECRET = "FAKE-SECRET-do-not-echo-7f3a9c2e1b" + + +def _write(tmp_path, text: str, name: str = "config.yaml"): + path = tmp_path / name + path.write_text(text, encoding="utf-8") + return path + + +def _load(tmp_path, text: str, name: str = "config.yaml"): + return ArgsContainer._load_config_file(_write(tmp_path, text, name)) + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + for name in ("TOKEN", "OTHER", "SEL", "HDR", "PAYLOAD", "MODE", "MISSING", "LIMIT"): + monkeypatch.delenv(name, raising=False) + + +# --- _env directive: set / unset / empty / default ------------------------------------- + + +def test_env_directive_reads_set_variable(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "abc") + + assert _load(tmp_path, "token: {_env: TOKEN}\nname: plain\n") == [ + {"token": "abc", "name": "plain"} + ] + + +def test_env_directive_unset_without_default_fails_closed(tmp_path) -> None: + config = _write(tmp_path, "limit: 3\nauth:\n token: {_env: TOKEN}\n") + + with pytest.raises(ConfigFileError) as excinfo: + ArgsContainer._load_config_file(config) + + message = str(excinfo.value) + assert "Environment variable TOKEN is required by _env at auth.token" in message + assert str(config) in message + assert "not set" in message + + +def test_env_directive_empty_string_counts_as_set(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "") + + assert _load(tmp_path, "token: {_env: TOKEN, default: fallback}\n") == [{"token": ""}] + + +def test_env_directive_default_used_only_when_unset(tmp_path, monkeypatch) -> None: + text = "token: {_env: TOKEN, default: fallback}\n" + assert _load(tmp_path, text) == [{"token": "fallback"}] + + monkeypatch.setenv("TOKEN", "real") + assert _load(tmp_path, text) == [{"token": "real"}] + + +@pytest.mark.parametrize( + ("default", "expected"), + [ + ("5", "5"), + ("1.5", "1.5"), + ("true", "true"), + ("no", "false"), + ("'007'", "007"), + ("''", ""), + ("2024-01-02", "2024-01-02"), + ], +) +def test_env_directive_default_is_stringified(tmp_path, default, expected) -> None: + [config] = _load(tmp_path, f"value: {{_env: TOKEN, default: {default}}}\n") + + assert config["value"] == expected + assert type(config["value"]) is str + + +def test_env_directive_value_is_string_even_when_numeric(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("LIMIT", "10") + + [config] = _load(tmp_path, "limit: {_env: LIMIT, default: 5}\n") + + assert config["limit"] == "10" + + +@pytest.mark.parametrize("default", ["null", "~", "[a]", "{a: 1}"]) +def test_env_directive_non_scalar_or_null_default_rejected(tmp_path, monkeypatch, default) -> None: + monkeypatch.setenv("TOKEN", "set") # rejected structurally, even when set + + with pytest.raises(ConfigFileError, match=r"_env at token: default must be a string"): + _load(tmp_path, f"token: {{_env: TOKEN, default: {default}}}\n") + + +# --- _env directive: placement ----------------------------------------------------------- + + +def test_env_directive_in_nested_maps_and_lists(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "t") + monkeypatch.setenv("OTHER", "o") + text = ( + "body:\n" + " items:\n" + " - {_env: TOKEN}\n" + " - plain\n" + " - nested: {deep: {_env: OTHER}}\n" + " keep: 3\n" + ) + + assert _load(tmp_path, text) == [ + {"body": {"items": ["t", "plain", {"nested": {"deep": "o"}}], "keep": 3}} + ] + + +def test_env_directive_nested_error_names_list_path(tmp_path) -> None: + with pytest.raises(ConfigFileError, match=r"_env at body\.items\[1\]\.deep in "): + _load(tmp_path, "body:\n items:\n - x\n - deep: {_env: MISSING}\n") + + +def test_env_directive_in_defaults_and_configs(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "t") + monkeypatch.setenv("OTHER", "o") + text = ( + "_defaults:\n" + " token: {_env: TOKEN}\n" + "configs:\n" + " - name: a\n" + " - name: {_env: OTHER}\n" + ) + + assert _load(tmp_path, text) == [ + {"token": "t", "name": "a"}, + {"token": "t", "name": "o"}, + ] + + +@pytest.mark.parametrize( + ("text", "location"), + [ + ("_defaults:\n token: {_env: MISSING}\nconfigs:\n - {}\n", "_defaults.token"), + ("configs:\n - {}\n - token: {_env: MISSING}\n", r"configs\[1\]\.token"), + ("- {}\n- token: {_env: MISSING}\n", r"\[1\]\.token"), + ], +) +def test_env_directive_missing_location_per_document_shape(tmp_path, text, location) -> None: + with pytest.raises(ConfigFileError, match=rf"MISSING is required by _env at {location} "): + _load(tmp_path, text) + + +def test_env_directive_in_json_config(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "t") + text = json.dumps({"token": {"_env": "TOKEN"}, "n": {"_env": "OTHER", "default": 2}}) + + assert _load(tmp_path, text, name="config.json") == [{"token": "t", "n": "2"}] + + +def test_env_directive_shared_alias_resolved_consistently(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "t") + text = "a: &tok {_env: TOKEN}\nb: *tok\nc: [*tok]\n" + + assert _load(tmp_path, text) == [{"a": "t", "b": "t", "c": ["t"]}] + + +def test_top_level_env_helper_key_is_not_a_directive(tmp_path) -> None: + # A document/config-entry mapping is never itself a directive, even when + # the document uses _env elsewhere. + text = "_env: &e {region: us}\nname: x\ntoken: {_env: TOKEN, default: d}\n" + assert _load(tmp_path, text) == [{"_env": {"region": "us"}, "name": "x", "token": "d"}] + + listed = "- _env: TOKEN\n token: {_env: TOKEN, default: d}\n" + assert _load(tmp_path, listed, name="list.yaml") == [{"_env": "TOKEN", "token": "d"}] + + +def test_env_directive_in_recursive_alias_rejected(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "t") + + with pytest.raises(ConfigFileError, match="recursive YAML alias"): + _load(tmp_path, "a: &a\n tok: {_env: TOKEN}\n self: *a\n") + + +def test_recursive_alias_without_directive_still_loads(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "t") + # The document uses _env elsewhere, so the resolver walks the cycle. + [config] = _load(tmp_path, "tok: {_env: TOKEN}\na: &a\n self: *a\n") + + assert config["tok"] == "t" + assert config["a"]["self"] is config["a"] + + +# --- _env directive: interplay with _select ---------------------------------------------- + +SELECT_WITH_ENV = ( + "_select:\n" + " env: SEL\n" + " cases:\n" + " a:\n" + " token: {_env: TOKEN}\n" + " b:\n" + " token: {_env: OTHER}\n" + " default:\n" + " token: {_env: MISSING, default: dflt}\n" +) + + +def test_env_directive_in_selected_branch_is_resolved(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("SEL", "a") + monkeypatch.setenv("TOKEN", "t") + + # OTHER (dormant branch b) is unset and not required. + assert _load(tmp_path, SELECT_WITH_ENV) == [{"token": "t"}] + + +def test_env_directive_in_default_branch_is_resolved(tmp_path) -> None: + assert _load(tmp_path, SELECT_WITH_ENV) == [{"token": "dflt"}] + + +def test_env_directive_missing_in_selected_branch_names_selection(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("SEL", "b") + + with pytest.raises(ConfigFileError) as excinfo: + _load(tmp_path, SELECT_WITH_ENV) + + message = str(excinfo.value) + assert "Environment variable OTHER is required by _env at token" in message + assert "within the selected _select branch" in message + assert "'b'" not in message + + +def test_env_directive_in_unused_select_helper_is_not_looked_up(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("SEL", "a") + text = "_unused: {token: {_env: MISSING}}\n" + SELECT_WITH_ENV.replace( + "{_env: TOKEN}", "plain" + ) + + assert _load(tmp_path, text) == [{"token": "plain"}] + + +def test_env_directive_in_ignored_top_level_key_is_not_looked_up(tmp_path) -> None: + text = "_shared: {token: {_env: MISSING}}\nconfigs:\n - name: a\n" + + assert _load(tmp_path, text) == [{"name": "a"}] + + +@pytest.mark.parametrize("selection", ["a", "b", None]) +def test_malformed_directive_in_dormant_branch_fails_in_every_env( + tmp_path, monkeypatch, selection +) -> None: + if selection is not None: + monkeypatch.setenv("SEL", selection) + monkeypatch.setenv("TOKEN", "t") + monkeypatch.setenv("OTHER", "o") + text = SELECT_WITH_ENV.replace( + "token: {_env: OTHER}", "token: {_env: OTHER, fallback: x}" + ) + + with pytest.raises( + ConfigFileError, + match=r"_env at _select\.cases\.b\.token allows only an optional 'default'", + ): + _load(tmp_path, text) + + +def test_malformed_directive_in_nested_dormant_selector_rejected(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("SEL", "a") + text = ( + "_select:\n" + " env: SEL\n" + " cases:\n" + " a: {name: x}\n" + " b:\n" + " _select:\n" + " env: OTHER\n" + " cases:\n" + " c:\n" + " configs:\n" + " - token: {_env: 1BAD}\n" + ) + + with pytest.raises( + ConfigFileError, + match=r"_env at _select\.cases\.b\._select\.cases\.c\.configs\[0\]\.token must be", + ): + _load(tmp_path, text) + + +def test_select_behavior_unchanged_for_directive_free_selectors(tmp_path, monkeypatch) -> None: + text = "_select:\n env: SEL\n cases:\n a: {name: x}\n" + + with pytest.raises(ConfigFileError, match="Environment variable SEL is required by _select"): + _load(tmp_path, text) + monkeypatch.setenv("SEL", "a") + assert _load(tmp_path, text) == [{"name": "x"}] + + +# --- _env directive: structural errors ---------------------------------------------------- + + +@pytest.mark.parametrize("name", ["1BAD", "has-dash", "''", "5", "null", "[A]", "'A B'"]) +def test_env_directive_invalid_name_rejected(tmp_path, monkeypatch, name) -> None: + monkeypatch.setenv("TOKEN", "t") + + with pytest.raises( + ConfigFileError, + match=r"_env at token must be a valid environment variable name matching", + ): + _load(tmp_path, f"token: {{_env: {name}}}\n") + + +def test_env_directive_invalid_name_is_not_echoed(tmp_path) -> None: + with pytest.raises(ConfigFileError) as excinfo: + _load(tmp_path, f"token: {{_env: '{SECRET}-x'}}\n") + + assert SECRET not in str(excinfo.value) + + +@pytest.mark.parametrize("sibling", ["fallback: x", "_note: x", "env: X", "defaults: x"]) +def test_env_directive_stray_sibling_rejected(tmp_path, monkeypatch, sibling) -> None: + monkeypatch.setenv("TOKEN", "t") + + with pytest.raises(ConfigFileError, match=r"_env at token allows only an optional 'default'"): + _load(tmp_path, f"token: {{_env: TOKEN, {sibling}}}\n") + + +# --- no value echo ------------------------------------------------------------------------- + + +class _Mode(Enum): + FAST = "fast" + SLOW = "slow" + + +def test_env_values_never_echoed_on_conversion_errors(tmp_path, monkeypatch, capsys) -> None: + monkeypatch.setenv("TOKEN", SECRET) + config = _write(tmp_path, "mode: {_env: TOKEN}\npayload: {_env: TOKEN}\n") + + with pytest.raises(ConfigFileError) as enum_error: + ArgsContainer.load(config, mode=(None, _Mode.FAST)) + with pytest.raises(ConfigFileError) as json_error: + ArgsContainer.load(config, payload=("payload", None, None, True)) + + assert "Invalid value for mode" in str(enum_error.value) + assert "Invalid JSON for payload" in str(json_error.value) + for excinfo in (enum_error, json_error): + # The rendered traceback (message plus any displayed cause/context). + assert SECRET not in "".join(traceback.format_exception(excinfo.value)) + captured = capsys.readouterr() + assert SECRET not in captured.out + captured.err + + +def test_env_field_conversion_error_does_not_echo_value(monkeypatch) -> None: + monkeypatch.setenv("LIMIT", SECRET) + + with pytest.raises(ConfigFileError) as excinfo: + ArgsContainer.load(None, limit=EnvField("LIMIT", (None, None, int))) + + assert str(excinfo.value) == "Invalid value for limit from environment variable LIMIT" + assert SECRET not in "".join(traceback.format_exception(excinfo.value)) + + +def test_origin_reports_env_name_not_value(monkeypatch) -> None: + monkeypatch.setenv("TOKEN", SECRET) + + [args] = ArgsContainer.load(None, token=EnvField("TOKEN", (None, None))) + + assert args.token == SECRET + assert args.origin("token") == "env:TOKEN" + assert "_origins" not in args.to_dict() + + +# --- EnvField tier --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("cli", "config", "env", "expected", "origin"), + [ + (True, True, True, "cli", "cli"), + (True, True, False, "cli", "cli"), + (True, False, True, "cli", "cli"), + (True, False, False, "cli", "cli"), + (False, True, True, "config", "config"), + (False, True, False, "config", "config"), + (False, False, True, "env", "env:TOKEN"), + (False, False, False, "dflt", "default"), + ], +) +def test_env_field_precedence_matrix( + tmp_path, monkeypatch, cli, config, env, expected, origin +) -> None: + config_path = _write(tmp_path, "token: config\n" if config else "other: 1\n") + if env: + monkeypatch.setenv("TOKEN", "env") + + [args] = ArgsContainer.load( + config_path, token=EnvField("TOKEN", ("cli" if cli else "dflt", "dflt")) + ) + + assert args.token == expected + assert args.origin("token") == origin + + +def test_config_env_directive_beats_env_field_tier(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "from-directive") + monkeypatch.setenv("OTHER", "from-tier") + config = _write(tmp_path, "token: {_env: TOKEN}\n") + + [args] = ArgsContainer.load(config, token=EnvField("OTHER", (None, None))) + + assert args.token == "from-directive" + assert args.origin("token") == "config" + + +def test_env_field_empty_string_counts_as_set(monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "") + + [args] = ArgsContainer.load(None, token=EnvField("TOKEN", ("dflt", "dflt"))) + + assert args.token == "" + assert args.origin("token") == "env:TOKEN" + + +def test_env_field_satisfies_required_and_names_var_when_missing(monkeypatch) -> None: + with pytest.raises( + ConfigFileError, + match=( + r"^token is required \(via CLI argument, config file, " + r"or environment variable TOKEN\)$" + ), + ): + ArgsContainer.load(None, token=EnvField("TOKEN", (None,))) + + monkeypatch.setenv("TOKEN", "t") + [args] = ArgsContainer.load(None, token=EnvField("TOKEN", (None,))) + assert args.token == "t" + + +def test_env_field_wraps_every_tuple_shape(monkeypatch) -> None: + monkeypatch.setenv("TOKEN", "7") + + [args] = ArgsContainer.load( + None, + one=EnvField("TOKEN", (None,)), + two=EnvField("TOKEN", (None, None)), + three=EnvField("TOKEN", (None, None, int)), + four=EnvField("TOKEN", ("four_key", None, None, True)), + five=EnvField("TOKEN", ("five_key", None, None, False, True)), + six=EnvField("TOKEN", ("six_key", None, None, False, False, int)), + ) + + assert args.to_dict() == { + "one": "7", "two": "7", "three": 7, "four": 7, "five": "7", "six": 7, + } + + +def test_env_field_json_conversion(monkeypatch) -> None: + monkeypatch.setenv("PAYLOAD", '{"a": [1, 2]}') + + [args] = ArgsContainer.load( + None, payload=EnvField("PAYLOAD", ("payload", None, None, True)) + ) + + assert args.payload == {"a": [1, 2]} + + +def test_env_directive_json_field_conversion(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("PAYLOAD", '{"a": 1}') + config = _write(tmp_path, "payload: {_env: PAYLOAD}\nflag: {_env: MISSING, default: true}\n") + + [args] = ArgsContainer.load( + config, + payload=("payload", None, None, True), + flag=("flag", None, None, True), + ) + + assert args.payload == {"a": 1} + assert args.flag is True + + +def test_env_values_feed_repeatable_header_fields(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("HDR", "X-A: 1") + config = _write(tmp_path, "header:\n - {_env: HDR}\n - 'X-B: 2'\n") + + [from_config] = ArgsContainer.load(config, header=(None, None)) + [from_tier] = ArgsContainer.load(None, header=EnvField("HDR", (None, None))) + + assert _header_entries(from_config.header) == ["X-A: 1", "X-B: 2"] + assert from_tier.header == "X-A: 1" + assert _header_entries(from_tier.header) == ["X-A: 1"] + + +def test_env_field_enum_conversion(monkeypatch) -> None: + monkeypatch.setenv("MODE", "slow") + + [args] = ArgsContainer.load(None, mode=EnvField("MODE", (_Mode.FAST, _Mode.FAST))) + + assert args.mode is _Mode.SLOW + + +@pytest.mark.parametrize("env", ["", "1BAD", "A-B", None]) +def test_env_field_rejects_invalid_names(env) -> None: + with pytest.raises(ValueError, match="EnvField.env must be a valid environment variable"): + EnvField(env, (None, None)) # type: ignore[arg-type] + + +@pytest.mark.parametrize("spec", [(), ("a",) * 7, [None, None]]) +def test_env_field_rejects_invalid_specs(spec) -> None: + with pytest.raises(ValueError, match="EnvField.spec"): + EnvField("TOKEN", spec) # type: ignore[arg-type] + + +# --- unchanged behavior without directives or env tier ------------------------------------- + + +def test_plain_specs_never_read_same_named_env(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("token", "env") + monkeypatch.setenv("TOKEN", "env") + + [args] = ArgsContainer.load(_write(tmp_path, "other: 1\n"), token=(None, None)) + + assert args.token is None + assert args.origin("token") == "default" + + +def test_plain_specs_origins(tmp_path) -> None: + [args] = ArgsContainer.load( + _write(tmp_path, "b: config\n"), a=("cli", None), b=(None, None), c=(None, None) + ) + + assert (args.origin("a"), args.origin("b"), args.origin("c")) == ("cli", "config", "default") + assert args.origin("unknown") is None + assert args.to_dict() == {"a": "cli", "b": "config", "c": None} + + +def test_directive_free_documents_skip_env_resolution(tmp_path, monkeypatch) -> None: + def forbidden(*_args, **_kwargs): + raise AssertionError("resolver must not run without an _env directive") + + monkeypatch.setattr(args_container_module, "_EnvValueResolver", forbidden) + monkeypatch.setenv("SEL", "a") + text = ( + "_shared: &s {env: TOKEN, default: x}\n" + "_select:\n" + " env: SEL\n" + " cases:\n" + " a:\n" + " _defaults: {opts: *s}\n" + " configs: [{name: a, list: [1, {k: v}]}]\n" + ) + + assert _load(tmp_path, text) == [ + {"opts": {"env": "TOKEN", "default": "x"}, "name": "a", "list": [1, {"k": "v"}]} + ] diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 9c6965e..9a5bcee 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -183,7 +183,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.19" in metadata + assert "Version: 0.1.20" in metadata assert "Requires-Dist: tangle-api==0.1.1" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_pipeline_cfg_select_env.py b/tests/test_pipeline_cfg_select_env.py new file mode 100644 index 0000000..03be1f1 --- /dev/null +++ b/tests/test_pipeline_cfg_select_env.py @@ -0,0 +1,463 @@ +"""Pipeline ``cfg`` files resolve ``_select`` and ``_env`` like ``--config`` files.""" + +from __future__ import annotations + +import traceback +from pathlib import Path + +import pytest +import yaml + +from tangle_cli.pipeline_compiler import compile_pipeline +from tangle_cli.python_pipeline.cfg import load_cfg +from tangle_cli.python_pipeline.compiler_context import ( + PipelineCompileKey, + canonical_repo_path, + overrides_fingerprint, +) +from tangle_cli.python_pipeline.errors import CompileError + +SECRET = "FAKE-SECRET-do-not-echo-7f3a9c2e1b" +ENV_VARS = ("CFG_ENV", "CFG_TOKEN", "CFG_LIMIT", "CFG_CHILD_ENV") + +_TASK = ( + "from tangle_cli.python_pipeline import In, Out, pipeline, subpipeline, task\n" + "\n" + "@task(image='python:3.12')\n" + "def echo(value: str = 'task-default', kind: str = 'task-default'):\n" + " '''Echo.\n" + "\n" + " Metadata:\n" + " Name: Echo\n" + " '''\n" + " print(value, kind)\n" + "\n" +) + +ROOT_ONLY = _TASK + ( + "@pipeline('Root', config='root_config.yaml')\n" + "def root(seed: In[str], cfg) -> Out[str]:\n" + " run = echo(value=cfg.value, kind=type(cfg.limit).__name__, wait_for=seed)\n" + " return run\n" +) + +WITH_CHILD = _TASK + ( + "@pipeline('Child', config='child_config.yaml')\n" + "def child(seed: In[str], cfg) -> Out[str]:\n" + " run = echo(value=cfg.value, kind=type(cfg.limit).__name__, wait_for=seed)\n" + " return run\n" + "\n" + "@pipeline('Root', config='root_config.yaml', propagate_config={propagate})\n" + "def root(seed: In[str], cfg) -> Out[str]:\n" + " return subpipeline(child){edge}.named('Run Child')(seed=seed)\n" +) + +SELECT_ROOT = ( + "_shared: &shared\n" + " limit: 5\n" + "_select:\n" + " env: CFG_ENV\n" + " cases:\n" + " prod:\n" + " <<: *shared\n" + " value: prod-value\n" + " dev:\n" + " value: dev-value\n" + " limit: 1.5\n" + " default:\n" + " value: default-value\n" + " limit: 0\n" +) + +SELECT_CHILD = ( + "_select:\n" + " env: CFG_ENV\n" + " cases:\n" + " prod: {value: child-prod, limit: 1, extra: prod-only}\n" + " dev: {value: child-dev, limit: 2}\n" + " also-prod: {value: child-prod, limit: 1, extra: prod-only}\n" +) + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + for name in ENV_VARS: + monkeypatch.delenv(name, raising=False) + + +def _project(tmp_path: Path, source: str, root_config: str, child_config: str | None = None): + project = tmp_path / "project" + project.mkdir(exist_ok=True) + (project / "pipeline.py").write_text(source) + (project / "root_config.yaml").write_text(root_config) + if child_config is not None: + (project / "child_config.yaml").write_text(child_config) + return project / "pipeline.py" + + +def _args(path: Path) -> dict: + data = yaml.safe_load(path.read_text()) + (task,) = data["implementation"]["graph"]["tasks"].values() + return {key: task["arguments"][key] for key in ("value", "kind")} + + +def _compile(src: Path, tmp_path: Path, name: str = "compiled", **kwargs): + out = tmp_path / "out" / f"{name}.yaml" + result = compile_pipeline(src, out, pipeline_name="Root", **kwargs) + return out, result + + +def _child_sidecar(result) -> Path: + (path,) = [p for p in result.subgraph_paths if p.name.startswith("child-")] + return path + + +# --- root config.yaml ------------------------------------------------------------------ + + +@pytest.mark.parametrize( + ("selection", "value", "kind"), + [("prod", "prod-value", "int"), ("dev", "dev-value", "float")], +) +def test_root_select_picks_branch_with_native_types( + tmp_path, monkeypatch, selection, value, kind +) -> None: + monkeypatch.setenv("CFG_ENV", selection) + src = _project(tmp_path, ROOT_ONLY, SELECT_ROOT) + + out, _ = _compile(src, tmp_path) + + assert _args(out) == {"value": value, "kind": kind} + + +@pytest.mark.parametrize("selection", [None, "staging"]) +def test_root_select_default_fallback(tmp_path, monkeypatch, selection) -> None: + if selection is not None: + monkeypatch.setenv("CFG_ENV", selection) + src = _project(tmp_path, ROOT_ONLY, SELECT_ROOT) + + out, _ = _compile(src, tmp_path) + + assert _args(out) == {"value": "default-value", "kind": "int"} + + +def test_root_select_unmatched_without_default_fails_closed_without_echo( + tmp_path, monkeypatch +) -> None: + monkeypatch.setenv("CFG_ENV", SECRET) + config = SELECT_ROOT.split(" default:")[0] + src = _project(tmp_path, ROOT_ONLY, config) + + with pytest.raises(CompileError) as excinfo: + _compile(src, tmp_path) + + message = str(excinfo.value) + assert "Environment variable CFG_ENV does not match any configured _select case" in message + assert "'dev', 'prod'" in message + assert "root_config.yaml" in message + assert SECRET not in "".join(traceback.format_exception(excinfo.value)) + assert not (tmp_path / "out").exists() + + +def test_root_select_unset_without_default_fails_closed(tmp_path) -> None: + src = _project(tmp_path, ROOT_ONLY, SELECT_ROOT.split(" default:")[0]) + + with pytest.raises(CompileError, match="CFG_ENV is required by _select but is not set"): + _compile(src, tmp_path) + + +def test_select_root_with_regular_sibling_rejected(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + src = _project(tmp_path, ROOT_ONLY, "value: stray\n" + SELECT_ROOT) + + with pytest.raises(CompileError, match="_select must be the only regular key"): + _compile(src, tmp_path) + + +# --- _env in cfg ------------------------------------------------------------------------ + + +def test_env_directive_in_cfg_stays_string(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_TOKEN", "from-env") + monkeypatch.setenv("CFG_LIMIT", "7") + src = _project( + tmp_path, ROOT_ONLY, "value: {_env: CFG_TOKEN}\nlimit: {_env: CFG_LIMIT, default: 3}\n" + ) + + out, _ = _compile(src, tmp_path) + + # `7` is NOT YAML-coerced to int, unlike a raw CLI --override string. + assert _args(out) == {"value": "from-env", "kind": "str"} + + +def test_env_directive_default_in_cfg_is_stringified(tmp_path) -> None: + src = _project( + tmp_path, ROOT_ONLY, "value: {_env: CFG_TOKEN, default: d}\nlimit: {_env: CFG_LIMIT, default: 3}\n" + ) + + out, _ = _compile(src, tmp_path) + + assert _args(out) == {"value": "d", "kind": "str"} + + +def test_env_directive_missing_in_cfg_names_var_and_path(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + config = SELECT_ROOT.replace("value: prod-value", "value: {_env: CFG_TOKEN}") + src = _project(tmp_path, ROOT_ONLY, config) + + with pytest.raises(CompileError) as excinfo: + _compile(src, tmp_path) + + message = str(excinfo.value) + assert "Environment variable CFG_TOKEN is required by _env at value" in message + assert "within the selected _select branch" in message + + +def test_env_directive_inside_selected_branch(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + monkeypatch.setenv("CFG_TOKEN", "tok") + config = SELECT_ROOT.replace("value: prod-value", "value: {_env: CFG_TOKEN}").replace( + "value: dev-value", "value: {_env: CFG_UNSET_IN_DORMANT}" + ) + src = _project(tmp_path, ROOT_ONLY, config) + + out, _ = _compile(src, tmp_path) + + assert _args(out) == {"value": "tok", "kind": "int"} + + +@pytest.mark.parametrize("selection", ["prod", "dev", None]) +def test_dormant_branch_structure_is_checked_in_every_env( + tmp_path, monkeypatch, selection +) -> None: + if selection is not None: + monkeypatch.setenv("CFG_ENV", selection) + config = SELECT_ROOT.replace("value: dev-value", "value: {_env: CFG_TOKEN, fallback: x}") + src = _project(tmp_path, ROOT_ONLY, config) + + with pytest.raises(CompileError, match=r"_env at _select\.cases\.dev\.value allows only"): + _compile(src, tmp_path) + + +@pytest.mark.parametrize("selection", ["prod", "dev", None]) +def test_template_file_rejected_in_any_branch(tmp_path, monkeypatch, selection) -> None: + if selection is not None: + monkeypatch.setenv("CFG_ENV", selection) + config = SELECT_ROOT.replace("value: dev-value", "value: dev-value\n template_file: x.j2") + src = _project(tmp_path, ROOT_ONLY, config) + + with pytest.raises(CompileError, match="top-level `template_file:` key"): + _compile(src, tmp_path) + + +@pytest.mark.parametrize("branch", ["[a, b]", "just-a-string"]) +def test_non_mapping_branch_rejected_in_every_env(tmp_path, monkeypatch, branch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + config = SELECT_ROOT.replace(" default:\n value: default-value\n limit: 0\n", "") + config += f" broken: {branch}\n" + src = _project(tmp_path, ROOT_ONLY, config) + + with pytest.raises(CompileError, match=r"_select case 'broken' must be a mapping"): + _compile(src, tmp_path) + + +def test_configs_key_is_ordinary_cfg_data(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + monkeypatch.setenv("CFG_TOKEN", "t") + config = ( + "_select:\n" + " env: CFG_ENV\n" + " cases:\n" + " prod:\n" + " configs: [1, {_env: CFG_TOKEN}]\n" + " _defaults: plain\n" + ) + path = tmp_path / "config.yaml" + path.write_text(config) + + cfg = load_cfg(path) + + assert cfg.configs == [1, "t"] + assert cfg._defaults == "plain" + + +# --- overrides -------------------------------------------------------------------------- + + +def test_cli_override_wins_over_selected_branch(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + src = _project(tmp_path, ROOT_ONLY, SELECT_ROOT) + + out, _ = _compile(src, tmp_path, overrides={"value": "from-cli", "limit": "true"}) + + assert _args(out) == {"value": "from-cli", "kind": "bool"} + + +def test_cli_override_wins_over_env_directive(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_TOKEN", "from-env") + src = _project(tmp_path, ROOT_ONLY, "value: {_env: CFG_TOKEN}\nlimit: 1\n") + + out, _ = _compile(src, tmp_path, overrides={"value": "from-cli"}) + + assert _args(out)["value"] == "from-cli" + + +def _child_project(tmp_path, *, propagate=False, edge="", root_config=None, child=SELECT_CHILD): + source = WITH_CHILD.replace("{propagate}", str(propagate)).replace("{edge}", edge) + return _project(tmp_path, source, root_config or "unused: 1\n", child) + + +@pytest.mark.parametrize( + ("selection", "value", "kind"), [("prod", "child-prod", "int"), ("dev", "child-dev", "int")] +) +def test_child_config_select(tmp_path, monkeypatch, selection, value, kind) -> None: + monkeypatch.setenv("CFG_ENV", selection) + src = _child_project(tmp_path) + + _, result = _compile(src, tmp_path) + + assert _args(_child_sidecar(result)) == {"value": value, "kind": kind} + + +def test_override_config_key_checked_against_selected_branch(tmp_path, monkeypatch) -> None: + src = _child_project(tmp_path, edge=".override_config(extra='edge')") + + monkeypatch.setenv("CFG_ENV", "prod") + _compile(src, tmp_path, name="prod") + + monkeypatch.setenv("CFG_ENV", "dev") + with pytest.raises(CompileError, match=r"'extra' is not a key in that child's config.yaml"): + _compile(src, tmp_path, name="dev") + + +def test_override_config_wins_over_selected_child_branch(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "dev") + src = _child_project(tmp_path, edge=".override_config(value='edge')") + + _, result = _compile(src, tmp_path) + + assert _args(_child_sidecar(result))["value"] == "edge" + + +# --- propagate_config broadcast --------------------------------------------------------- + + +@pytest.mark.parametrize("selection", ["prod", "dev"]) +def test_broadcast_carries_resolved_root_values(tmp_path, monkeypatch, selection) -> None: + monkeypatch.setenv("CFG_ENV", selection) + monkeypatch.setenv("CFG_TOKEN", f"{selection}-token") + root_config = ( + "_select:\n" + " env: CFG_ENV\n" + " cases:\n" + " prod: {value: {_env: CFG_TOKEN}, limit: 9}\n" + " dev: {value: {_env: CFG_TOKEN}, limit: 8}\n" + ) + src = _child_project(tmp_path, propagate=True, root_config=root_config) + + _, result = _compile(src, tmp_path) + + # The child declares `value`/`limit`, so the root's RESOLVED values win; + # a broadcast still holding `_select`/`_env` nodes would not match. + assert _args(_child_sidecar(result)) == {"value": f"{selection}-token", "kind": "int"} + + +# --- compile identity --------------------------------------------------------------------- + + +def _legacy_child_sidecar_name(src: Path) -> str: + key = PipelineCompileKey( + source_path=canonical_repo_path(src.resolve()), + function_qualname="child", + pipeline_name="Child", + config_path=canonical_repo_path((src.parent / "child_config.yaml").resolve()), + overrides_fingerprint=overrides_fingerprint({}), + ) + return f"child-{key.hash8()}.yaml" + + +def test_compile_identity_differs_across_selections(tmp_path, monkeypatch) -> None: + src = _child_project(tmp_path) + names = {} + for selection in ("prod", "dev", "also-prod"): + monkeypatch.setenv("CFG_ENV", selection) + _, result = _compile(src, tmp_path, name=selection) + names[selection] = _child_sidecar(result).name + + assert names["prod"] != names["dev"] + # Identical resolved values keep an identical identity. + assert names["prod"] == names["also-prod"] + assert _legacy_child_sidecar_name(src) not in names.values() + + +def test_compile_identity_differs_across_env_values(tmp_path, monkeypatch) -> None: + src = _child_project(tmp_path, child="value: {_env: CFG_TOKEN}\nlimit: 1\n") + names = [] + for token in ("a", "b", "a"): + monkeypatch.setenv("CFG_TOKEN", token) + _, result = _compile(src, tmp_path, name=f"run-{len(names)}") + names.append(_child_sidecar(result).name) + + assert names[0] != names[1] + assert names[0] == names[2] + + +def test_compile_identity_unchanged_without_directives(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + src = _child_project(tmp_path, child="value: plain\nlimit: 4\n") + + out, result = _compile(src, tmp_path) + + assert _child_sidecar(result).name == _legacy_child_sidecar_name(src) + assert _args(_child_sidecar(result)) == {"value": "plain", "kind": "int"} + assert "_select" not in out.read_text() + + +def test_directive_free_cfg_is_unchanged(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("CFG_ENV", "prod") + path = tmp_path / "config.yaml" + # Only a ROOT `_select` selects; a nested one stays ordinary data. + path.write_text( + "a: 1\nnested: {_select: {env: CFG_ENV, cases: {prod: 1}}, x: [1, 2]}\n_helper: true\n" + ) + + cfg = load_cfg(path, {"a": "2"}) + + assert cfg.a == 2 + assert cfg.nested.x == [1, 2] + assert cfg.nested._select.env == "CFG_ENV" + assert cfg.nested._select.cases.prod == 1 + assert cfg._helper is True + + +def test_resolved_cfg_document_holds_no_directive_nodes(tmp_path, monkeypatch) -> None: + from tangle_cli.python_pipeline.cfg import read_cfg_document + + monkeypatch.setenv("CFG_ENV", "prod") + monkeypatch.setenv("CFG_TOKEN", "t") + path = tmp_path / "config.yaml" + path.write_text(SELECT_ROOT.replace("value: prod-value", "value: {_env: CFG_TOKEN}")) + + document = read_cfg_document(path) + + assert document.data == {"limit": 5, "value": "t"} + assert document.env_dependent + assert len(document.branches) == 3 # prod, dev, default -- dormant included + + +def test_compile_key_config_identity_envelope() -> None: + from tangle_cli.pipeline_compiler import _compile_key_for + from tangle_cli.python_pipeline import pipeline + + @pipeline("Keyed") + def keyed() -> None: # pragma: no cover - never traced + return None + + cfg_path = Path("/nonexistent/config.yaml") + legacy = _compile_key_for(keyed, cfg_path, {"a": 1}) + with_a = _compile_key_for(keyed, cfg_path, {"a": 1}, config_identity="a" * 64) + with_b = _compile_key_for(keyed, cfg_path, {"a": 1}, config_identity="b" * 64) + + assert legacy.overrides_fingerprint == overrides_fingerprint({"a": 1}) + assert len({legacy, with_a, with_b}) == 3 + assert with_a == _compile_key_for(keyed, cfg_path, {"a": 1}, config_identity="a" * 64) diff --git a/uv.lock b/uv.lock index 6a0724f..2e7a4cd 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.19" +version = "0.1.20" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },