diff --git a/CHANGELOG.md b/CHANGELOG.md index 6deb6e95f..b0c463527 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`fluid validate` and `fluid plan` warn about an id the build phase refuses.** The build + phase puts `contract.id` and each `builds[].id` in directory names and generated code, so they + must start with a letter or underscore, use only letters, digits, `_`, `.` and `-`, and be at + most 128 characters. The contract schema allows more (`1orders`, or a longer id), and + `fluid apply` in a build mode refuses such an id before it applies anything, but validate and + plan said nothing. For a contract with builds, both now warn and name the id: plain + `fluid validate` passes and `--strict` fails, and `fluid plan` still writes the plan, whatever + its `--mode`. A contract without builds is not checked, as in `fluid apply`. + - **IaC plugins load when a contract needs their platform, and cannot take a built-in's name unless the operator says so.** A package that ships an OpenTofu emitter through the `fluid_build.iac_providers` entry-point group used to be imported whenever `fluid_build.iac` @@ -275,6 +284,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 as before. The Command Center run now reports `apply_build_id_not_found`, where it reported `build_not_found:`. +- **`fluid odcs validate` stopped with `Unexpected error: closing tag '[/]' ... has nothing to + close`** when a value that fails a schema check contained `[/]` or another Rich markup tag. The + error line quotes the value, and it was printed as markup. It is printed as plain text now, so + the command reports the error and exits 1. + - **A missing optional dependency names the extra to install, and silent fallbacks warn.** litellm, the MCP SDK, openlineage, boto3 and duckdb are imported through one gate: when one is missing, the command prints ` needs , which is not installed.` and @@ -328,6 +342,44 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 does, and reports its columns at `...schema.fields.`. No bundled schema accepts the wrapper, so `fluid diff --baseline` refuses such a contract as invalid (see Security). +- **`fluid diff --baseline` compares column types more strictly, so `--fail-on-breaking` stops + passing type changes that break consumers.** Contracts that used to diff clean can now report + breaking type changes (a pipeline that runs `fluid diff --baseline --fail-on-breaking` can start + exiting 1), and some findings that were `info` or `non_breaking` are now `breaking`. Rule by rule: + - A type name that is not plain ASCII is compared as written, not case-folded, so apart from + surrounding whitespace it matches nothing but itself, and the finding says so. Type text the + comparison cannot order (non-ASCII text, or parameters that are not plain integers, such as + `enum('a','b')` or `varchar(max)`) is a different type when the text differs after ASCII case + folding; it used to read as "no parameters" or as a whitespace difference. Case inside a quoted + parameter is still folded, so `enum('a')` and `enum('A')` still compare equal. + - A parameter change that is neither a pure widening nor a pure narrowing is now `breaking`, not + `info`: `varchar` to `varchar(10)` or back, `decimal(10,2)` to `decimal(12,1)` or to a bare + `decimal`. The kind stays `column_type_param_changed`; its text now reads "(not a pure + widening)". + - A bigger parameter is a widening only for types whose parameter measures capacity: lengths + (`varchar`, `char`, `string`, `text`, `varbinary` and their variants), precisions (`float`, + `time`, `timestamp`, `datetime` and their variants) and the decimal family. A parameter that + names something is not one: `geometry(4326)` to `geometry(32633)`, `binary(16)` to + `binary(32)` and `vector(3)` to `vector(1536)` are now `breaking`. + - A `decimal`, `dec`, `numeric`, `number` or `bignumeric` with a precision and a scale widens only + when the scale and the integer digits (precision minus scale) both hold. `decimal(10,2)` to + `decimal(12,4)` is still a widening; `decimal(10,2)` to `decimal(11,4)`, which holds one + integer digit fewer, is now `breaking`. + - A change to a different base type widens only when the new type declares no parameters of its + own: `int` to `decimal(5,0)` is now `breaking`, `int` to a bare `decimal` still widens. + - `time` to `timestamp` or `datetime`, `float` to `decimal` or `numeric`, and `int`, `integer` + or `int32` to `float` are no longer widenings. `float` to `double` and `int` to `double` are. + +- **`fluid diff --baseline` reports a name the new contract declares twice.** The comparison matched + exposes, columns and nested fields by name and kept the last of a repeated name, so an earlier + declaration was never compared. An expose id (`exposeId`, or `id`), a column name or a nested + field name that the new contract declares more than once is now `breaking` + (`expose_duplicate_id`, `column_duplicate_name`, `nested_field_duplicate_name`), in an expose both + contracts have and in one that is new, and whatever the baseline holds: a contract that already + declares a name twice is reported against itself until one declaration is removed, so a pipeline + that gates on it can start exiting 1, and the `fluid forge --refine` preview lists it as a + breaking change. + - **A failed `fluid apply` on OpenTofu printed raw JSON and no way forward.** The runner uses `-json`, so the error was the last 800 characters of diagnostic lines. It now reads the diagnostic (the summary, the resource address and the provider's detail) and says that OpenTofu @@ -546,8 +598,34 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 into `main.tf.json`. The module's allowlists now end in `\Z`, so such a name is refused with `Invalid SQL identifier: 'spectrum_events\n'`. That includes a name that reaches it with a trailing newline from an environment variable or a file; strip the value where it is set. + +- **The acquisition orchestrator check judged a second DNS answer, not the address the config + went to.** For kafka_connect, debezium and airbyte, the per-request guard that keeps a + contract-named `deployment.server_url` from receiving the connector config, resolved source + password included, unless its host is private, loopback or listed in + `FLUID_ACQUISITION_DEPLOYMENT_HOST_ALLOWLIST`, resolved the hostname again instead of looking at + the IP `safe_httpx_client` had pinned the connection to. A name that answered the two lookups + differently (DNS rebinding), or a non-ASCII name that Python's resolver spells as IDNA2003 and + httpx as IDNA2008 (`faß.de` is `fass.de` to one and `xn--fa-hia.de` to the other), passed as + private while the config went to an unlisted public address, and a listed name could be sent + to instance metadata the same way. The guard now judges the pinned address. The name still + decides the allowlist, the address decides public or private, a link-local address is refused + even for a listed name, and the refusal names both: `host 'connect.example.com', connecting to + '93.184.216.34', is public`. A URL that cannot be parsed, such as `http://[::1`, is refused with + "it has no host" where it raised `ValueError`. + +- **litellm 1.108.0 reads seven provider endpoints the dotenv guard did not know.** + `BESPOKE_API_BASE`, `CORALBRICKS_API_BASE`, `DATABRICKS_AGENT_API_BASE`, `INFERENCEAPIS_API_BASE`, + `LAYA_API_BASE`, `MICROSOFT_365_COPILOT_API_BASE` and `SCALEDOWN_API_BASE` choose the host + litellm sends the provider key to. They are now in `credentials/litellm_endpoints.py`, so a + project dotenv file cannot point them elsewhere while `FLUID_LLM_API_KEY` or a stored key is in + play, as for every other provider endpoint. + ### Upgrade notes +- New validate warning, so `--strict` fails: a contract with builds whose `contract.id` or a + `builds[].id` the build phase refuses (`1orders`, or more than 128 characters). `fluid apply` + in a build mode already refuses it; rename the id to fit. - A contract with a column typed `array`, `struct`, `map`, `object`, `record` or `row` (or `array(...)`) on an `aws` Glue table or a `gcp` BigQuery table now fails `fluid validate`, `fluid generate iac` and `fluid apply` with `column-type`, where it validated and wrote a @@ -604,6 +682,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 and |- or >- drops)`. A `sink.catalogAuth` field gets no hint, since its value is never shown. The schemas themselves are unchanged. +- **The ODCS and plugin-extension checks let a value with a trailing newline through a `pattern`, + too.** The fix above covered the FLUID schemas only. `fluid odcs validate` and + `fluid validate-artifacts` accepted an ODCS `schema[].id: "t1\n"` while refusing `"t 1"`, and + the forge copilot kept an LLM-proposed extension block whose value ended in a line break, such + as `use: "ci:basic\n"` for `customScaffold`. Every JSON Schema check now reads `$` as the end of + the value, including the ODPS, OPDS and Ossie checks, whose vendored schemas have no such + pattern yet. Their errors carry the same hint; an extension block that fails is dropped, as + before. An extension schema is read in the dialect its `$schema` declares, and as draft-07 when + it declares none, so a 2020-12 `unevaluatedProperties: false` is no longer ignored. + `patternProperties` keys are still matched as before, so OPDS's language-code key `^[a-z]{2}$` + accepts `"en\n"`. + - **`fluid diff --baseline` compared contracts that no schema accepts.** Its rules read only the shapes a schema allows, so a value of the wrong type hid a change: a column retyped from `decimal(18,2)` to the number `18` passed `--fail-on-breaking` with exit 0. Both contracts are diff --git a/fluid_build/api/changelog.py b/fluid_build/api/changelog.py index 3c0efdc8c..17b008268 100644 --- a/fluid_build/api/changelog.py +++ b/fluid_build/api/changelog.py @@ -39,6 +39,7 @@ diff_agent_policy, diff_columns, diff_consumes, + diff_duplicates, diff_metadata, diff_quality_severity, diff_sovereignty, @@ -72,7 +73,9 @@ def compare_contracts( """Return a structured changelog comparing two parsed contracts. ``baseline`` is treated as "old"; ``new`` as the candidate. Idempotent — - passing the same dict twice produces an empty report. + passing the same dict twice produces an empty report — unless the contract + declares an expose id, a column name or a nested field name twice: that is + reported whatever the baseline holds (see :func:`diff_duplicates`). Both are first validated (:func:`_require_valid`), and an invalid one raises :class:`ContractValidationError`. ``validate=False`` diffs whatever @@ -102,6 +105,10 @@ def compare_contracts( # What streaming builds read (builds[].properties.stream). report.extend(diff_streams(baseline, new)) + # A name the new contract declares twice: the comparison below keeps only + # the last declaration of it. + report.extend(diff_duplicates(new)) + # Per-expose schema and quality. for expose_id, idx, base_expose, new_expose in iter_expose_pairs(baseline, new): if base_expose is None and new_expose is not None: diff --git a/fluid_build/api/changelog_rules.py b/fluid_build/api/changelog_rules.py index f1aff1504..ca9f355a9 100644 --- a/fluid_build/api/changelog_rules.py +++ b/fluid_build/api/changelog_rules.py @@ -37,28 +37,82 @@ # types that strictly widen them. Anything not in this table is treated as # a breaking type change. Names normalized to upper-case AND stripped of # precision/scale annotations (those are compared separately below). +# +# Only conversions that keep every existing value are listed, and only into a +# target that declares no parameters of its own (see ``_classify_type_diff``). +# Apache Iceberg ("Schema Evolution" in the table spec) and Delta Lake ("Type +# widening") draw the same line: int -> long, float -> double and +# date -> timestamp (Iceberg v3, Delta timestampNTZ) widen; int -> float, +# float -> decimal and anything out of TIME are on neither list. _WIDENING: Mapping[str, frozenset[str]] = { - "INT": frozenset({"BIGINT", "INTEGER", "LONG", "DECIMAL", "NUMERIC", "FLOAT", "DOUBLE"}), - "INT32": frozenset({"INT64", "BIGINT", "LONG", "DECIMAL", "FLOAT", "DOUBLE"}), - "INTEGER": frozenset({"BIGINT", "LONG", "DECIMAL", "FLOAT", "DOUBLE"}), + "INT": frozenset({"BIGINT", "INTEGER", "LONG", "DECIMAL", "NUMERIC", "DOUBLE"}), + "INT32": frozenset({"INT64", "BIGINT", "LONG", "DECIMAL", "DOUBLE"}), + "INTEGER": frozenset({"BIGINT", "LONG", "DECIMAL", "DOUBLE"}), "SMALLINT": frozenset({"INT", "INT32", "INT64", "BIGINT", "INTEGER", "LONG"}), - "FLOAT": frozenset({"DOUBLE", "DECIMAL", "NUMERIC"}), + "FLOAT": frozenset({"DOUBLE"}), "VARCHAR": frozenset({"TEXT", "STRING"}), "STRING": frozenset({"TEXT"}), "DATE": frozenset({"DATETIME", "TIMESTAMP"}), - "TIME": frozenset({"DATETIME", "TIMESTAMP"}), } -# Type-name + parameter regex. Captures e.g. ``DECIMAL(10,2)`` as ("DECIMAL", -# "10,2") and ``VARCHAR(255)`` as ("VARCHAR", "255"). Suffix-attribute-less -# types ("BIGINT") fall through with empty params. -_TYPE_PARAM_RE = re.compile(r"^([A-Z0-9_]+)\s*(?:\(([^)]+)\))?\s*$") +# Fixed-point types: two parameters, (precision, scale). +_FIXED_POINT = frozenset({"DECIMAL", "DEC", "NUMERIC", "NUMBER", "BIGNUMERIC"}) + +# Types whose parameter measures capacity (a length or a precision), so a bigger +# one holds every value the smaller one did. The parameters of any other type +# name something instead (``GEOMETRY(4326)`` is a spatial reference id, +# ``VECTOR(1536)`` a dimension, ``BINARY(16)`` an exact width): changing them is +# not a widening, whichever way the number moves. +_MEASURED = _FIXED_POINT | frozenset( + { + "VARCHAR", + "VARCHAR2", + "NVARCHAR", + "NVARCHAR2", + "CHAR", + "NCHAR", + "CHARACTER", + "STRING", + "TEXT", + "VARBINARY", + "FLOAT", + "TIME", + "TIMESTAMP", + "TIMESTAMP_NTZ", + "TIMESTAMP_LTZ", + "TIMESTAMP_TZ", + "DATETIME", + "DATETIME2", + } +) + +# Type-name + parameter grammar. ASCII only, matched whole with ``fullmatch`` +# (``$`` would let a trailing newline through), and linear in the input: the one +# ``\s*`` sits inside the optional group, so a name followed by a long run of +# spaces can't backtrack quadratically (the input is stripped before it is +# matched). Captures e.g. ``DECIMAL(10,2)`` as ("DECIMAL", "10,2") and +# ``VARCHAR(255)`` as ("VARCHAR", "255"). Suffix-attribute-less types ("BIGINT") +# fall through with empty params. +_TYPE_PARAM_RE = re.compile(r"([A-Z0-9_]+)(?:\s*\(([^)]+)\))?", re.ASCII) + +# The parameters the classifier can order: plain ASCII integers. +_TYPE_INT_PARAMS_RE = re.compile(r"[0-9]+(?:\s*,\s*[0-9]+)*", re.ASCII) + +# The whitespace that may surround a parameter: ``str.strip()`` with no argument +# also removes non-ASCII spaces, which would let them pass for ASCII ones. +_ASCII_WS = " \t\n\r\f\v" def _norm_type(t: Any) -> str: if not isinstance(t, str): return "" - return t.strip().upper() + t = t.strip() + # Fold case for ASCII names only. ``str.upper()`` maps some non-ASCII + # letters onto ASCII ones (U+0131 dotless i -> "I", U+017F long s -> "S"), + # which would make an unknown type compare equal to a real one although the + # IaC emitters do not fold them. A non-ASCII name stays as written and + # matches nothing but itself. + return t.upper() if t.isascii() else t def _parse_type(t: str) -> Tuple[str, Optional[Tuple[int, ...]]]: @@ -70,21 +124,29 @@ def _parse_type(t: str) -> Tuple[str, Optional[Tuple[int, ...]]]: ``"BIGINT"`` → ``("BIGINT", None)`` ``"DECIMAL"`` → ``("DECIMAL", None)`` (no precision specified) - Returns ``("", None)`` when the input doesn't parse cleanly — caller - decides whether to treat that as a generic mismatch. + Returns ``("", None)`` for an empty input. A type that doesn't parse + cleanly (non-ASCII text, or parameters that aren't plain integers, such as + ``ENUM('A','B')`` or ``VARCHAR(MAX)``) comes back as ``(whole_string, + None)``: it equals itself and no other type, so a change in it is a + mismatch and never a normalization. """ if not t: return ("", None) - m = _TYPE_PARAM_RE.match(t.strip()) + t = t.strip() + m = _TYPE_PARAM_RE.fullmatch(t) if not m: - return (t.strip(), None) + return (t, None) base, params = m.group(1), m.group(2) if not params: return (base, None) + params = params.strip(_ASCII_WS) + if not _TYPE_INT_PARAMS_RE.fullmatch(params): + return (t, None) try: - nums = tuple(int(p.strip()) for p in params.split(",")) + nums = tuple(int(p) for p in params.split(",")) except ValueError: - return (base, None) + # Past the interpreter's integer-string limit. + return (t, None) return (base, nums) @@ -95,10 +157,15 @@ def _classify_param_change( ) -> Optional[str]: """Classify a same-base type parameter change as 'widening' / 'narrowing'. - Returns ``"widening"`` if every parameter grew (or stayed equal), - ``"narrowing"`` if any parameter shrank, ``"changed"`` for ambiguous / - not-comparable cases, or ``None`` when both sides have no params (so - there's no diff to classify). + Returns ``"widening"`` if no parameter shrank and the type's parameters + measure capacity (see ``_MEASURED``), ``"narrowing"`` if none grew, + ``"changed"`` for everything else (one side has no parameters, a different + number of them, parameters that moved in different directions, or ones that + name something rather than measure it), or ``None`` when both sides carry + the same parameters, or none (so there's no diff to classify). + + A fixed-point type's ``(precision, scale)`` is compared as (integer digits, + scale), not as the two numbers it is written with. """ if old_params is None and new_params is None: return None @@ -112,9 +179,19 @@ def _classify_param_change( if len(old_params) != len(new_params): # DECIMAL(10) vs DECIMAL(10,2) — semantically different shapes. return "changed" - pairs = list(zip(old_params, new_params, strict=True)) + if base in _FIXED_POINT and len(old_params) == 2: + # (precision, scale) holds ``precision - scale`` integer digits and + # ``scale`` fractional ones, so a bigger precision with a bigger scale + # can still shrink the integer part: (10,2) holds 99999999.99 and + # (11,4) holds 9999999.9999. Compare the two capacities. Delta Lake + # widens decimal(p,s) to (p+k1,s+k2) only when k1 >= k2 >= 0, the same + # condition; Iceberg widens precision at a fixed scale. + (o_prec, o_scale), (n_prec, n_scale) = old_params, new_params + pairs = [(o_prec - o_scale, n_prec - n_scale), (o_scale, n_scale)] + else: + pairs = list(zip(old_params, new_params, strict=True)) if all(n >= o for o, n in pairs): - return "widening" + return "widening" if base in _MEASURED else "changed" if all(n <= o for o, n in pairs): return "narrowing" # Mixed: e.g. precision grew but scale shrank — neither pure widening @@ -134,16 +211,26 @@ def _is_widening(old_type: str, new_type: str) -> bool: return new_type in widens_to +def _why_not_widening(old_type: str, new_type: str) -> str: + """The reason, for a finding's text, a type change is not a widening.""" + if old_type.isascii() and new_type.isascii(): + return "not a widening cast" + return "a type name with non-ASCII characters matches no other type" + + def _classify_type_diff(old_full: str, new_full: str) -> Tuple[str, str]: """Classify a (possibly parameterized) type change. Returns ``(severity, kind)``: - * ``("non_breaking", "column_type_widened")`` — base widened or - parameters strictly grew + * ``("non_breaking", "column_type_widened")`` — base widened to a type + that declares no parameters, or parameters strictly grew * ``("breaking", "column_type_changed")`` — base narrowed, parameters shrank, or unrelated type swap - * ``("info", "column_type_param_changed")`` — same base, parameters - changed in a way that's neither pure widening nor pure narrowing + * ``("breaking", "column_type_param_changed")`` — same base, parameters + changed in a way that's neither pure widening nor pure narrowing, or + only one side declares parameters + * ``("info", "column_type_normalized")`` — same base and parameters, + written with different formatting (whitespace, leading zeros) """ old_base, old_params = _parse_type(old_full) new_base, new_params = _parse_type(new_full) @@ -153,30 +240,85 @@ def _classify_type_diff(old_full: str, new_full: str) -> Tuple[str, str]: change = _classify_param_change(old_params, new_params, old_base) if change is None: # Same base, no param difference but full strings differ — - # whitespace or capitalization. Treat as info-level. + # whitespace or leading zeros (case is folded before this point). return ("info", "column_type_normalized") if change == "widening": return ("non_breaking", "column_type_widened") if change == "narrowing": return ("breaking", "column_type_changed") - return ("info", "column_type_param_changed") + return ("breaking", "column_type_param_changed") - # Different base type — fall back to the base-name widening table. - if _is_widening(old_base, new_base): + # Different base type — fall back to the base-name widening table. The + # table knows base names only: a target that declares parameters + # (``INT`` -> ``DECIMAL(5,0)``) may hold less than the source did. + if new_params is None and _is_widening(old_base, new_base): return ("non_breaking", "column_type_widened") return ("breaking", "column_type_changed") +def _column_name(col: Any) -> Optional[str]: + """The name a column is matched by: ``name``, else ``id``.""" + if isinstance(col, Mapping): + name = col.get("name") or col.get("id") + if isinstance(name, str): + return name + return None + + def _columns_by_name(columns: Sequence[Mapping[str, Any]]) -> dict[str, Mapping[str, Any]]: + """Index columns by name; the last of a repeated name wins. + + :func:`diff_duplicates` reports the names this would collapse. + """ out: dict[str, Mapping[str, Any]] = {} for col in columns or []: - if isinstance(col, Mapping): - name = col.get("name") or col.get("id") - if isinstance(name, str): - out[name] = col + name = _column_name(col) + if name is not None: + out[name] = col return out +def _duplicate_columns(columns: Sequence[Mapping[str, Any]]) -> dict[str, list[Mapping[str, Any]]]: + """Every name declared more than once, with all of its declarations.""" + seen: dict[str, list[Mapping[str, Any]]] = {} + if not isinstance(columns, Sequence): + return {} + for col in columns: + name = _column_name(col) + if name is not None: + seen.setdefault(name, []).append(col) + return {name: cols for name, cols in seen.items() if len(cols) > 1} + + +def _duplicate_changes( + declarations: Sequence[Mapping[str, Any]], + *, + kind: str, + noun: str, + expose_id: str, + path_prefix: str, +) -> List[Change]: + """A breaking ``Change`` for every name the new contract declares more than once. + + :func:`_columns_by_name` keeps the last declaration of a name, so without this + the earlier one would never be compared against the baseline. + """ + return [ + Change( + path=f"{path_prefix}.{name}", + kind=kind, + severity="breaking", + description=( + f"{noun} '{name}' is declared {len(declared)} times in expose " + f"'{expose_id}'; keep one (a later declaration hides an earlier one)" + ), + before=None, + after=[dict(d) for d in declared], + ) + for name, declared in _duplicate_columns(declarations).items() + ] + + def _expose_schema(expose: Mapping[str, Any]) -> Tuple[Sequence[Any], Optional[str]]: """Return an expose's columns and where, below the expose, they were found. @@ -417,17 +559,17 @@ def diff_columns( elif kind == "column_type_param_changed": desc = ( f"column '{col_name}' type parameters changed " - f"{old_type} -> {new_type} (review: not pure widening or narrowing)" + f"{old_type} -> {new_type} (not a pure widening)" ) elif kind == "column_type_normalized": desc = ( f"column '{col_name}' type normalized " - f"{old_type} -> {new_type} (whitespace/casing only)" + f"{old_type} -> {new_type} (formatting only)" ) else: desc = ( f"column '{col_name}' type changed " - f"{old_type} -> {new_type} (not a widening cast)" + f"{old_type} -> {new_type} ({_why_not_widening(old_type, new_type)})" ) changes.append( Change( @@ -991,14 +1133,10 @@ def iter_expose_pairs( base_exposes = list(baseline.get("exposes") or []) new_exposes = list(new.get("exposes") or []) base_by_id = { - (e.get("id") or e.get("exposeId") or f"_pos_{i}"): (i, e) - for i, e in enumerate(base_exposes) - if isinstance(e, Mapping) + _expose_key(i, e): (i, e) for i, e in enumerate(base_exposes) if isinstance(e, Mapping) } new_by_id = { - (e.get("id") or e.get("exposeId") or f"_pos_{i}"): (i, e) - for i, e in enumerate(new_exposes) - if isinstance(e, Mapping) + _expose_key(i, e): (i, e) for i, e in enumerate(new_exposes) if isinstance(e, Mapping) } all_ids = set(base_by_id) | set(new_by_id) for eid in all_ids: @@ -1006,3 +1144,89 @@ def iter_expose_pairs( n = new_by_id.get(eid) idx_in_new = n[0] if n else (b[0] if b else 0) yield eid, idx_in_new, (b[1] if b else None), (n[1] if n else None) + + +def _expose_key(index: int, expose: Mapping[str, Any]) -> Any: + """The id exposes are matched by; one without an id is matched by position.""" + return expose.get("id") or expose.get("exposeId") or f"_pos_{index}" + + +def _nested_duplicates(columns: Sequence[Any], *, expose_id: str, path_prefix: str) -> List[Change]: + """Every ``fields`` name declared twice below *columns*, at any depth.""" + changes: List[Change] = [] + if not isinstance(columns, Sequence): + return changes + for col in columns: + name = _column_name(col) + fields = col.get("fields") if name is not None else None + if isinstance(fields, Sequence): + prefix = f"{path_prefix}.{name}.fields" + changes.extend( + _duplicate_changes( + fields, + kind="nested_field_duplicate_name", + noun="nested field", + expose_id=expose_id, + path_prefix=prefix, + ) + ) + changes.extend(_nested_duplicates(fields, expose_id=expose_id, path_prefix=prefix)) + return changes + + +def _duplicate_exposes(exposes: Sequence[Any]) -> List[Change]: + """Every expose id declared twice. An expose without an id is matched by position.""" + declared_at: dict[Any, list[int]] = {} + for i, expose in enumerate(exposes): + if isinstance(expose, Mapping): + expose_id = expose.get("id") or expose.get("exposeId") + if expose_id: + declared_at.setdefault(expose_id, []).append(i) + return [ + Change( + path=f"exposes[{at[-1]}]", + kind="expose_duplicate_id", + severity="breaking", + description=( + f"expose '{expose_id}' is declared {len(at)} times " + f"(at {', '.join(f'exposes[{i}]' for i in at)}); " + f"keep one (a later declaration hides an earlier one)" + ), + before=None, + after={"id": expose_id, "indexes": at}, + ) + for expose_id, at in declared_at.items() + if len(at) > 1 + ] + + +def diff_duplicates(new: Mapping[str, Any]) -> List[Change]: + """Flag every expose id, column name and nested field name *new* declares twice. + + Breaking, and reported whatever the baseline holds. The comparison indexes + exposes, columns and fields by name and keeps the last of a repeated name, + so an earlier declaration would never be compared against the baseline, + and an emitter is free to read either one. It follows that a contract that + already declares a name twice is reported against itself, until one + declaration is removed. + """ + exposes = list(new.get("exposes") or []) + changes = _duplicate_exposes(exposes) + for i, expose in enumerate(exposes): + if not isinstance(expose, Mapping): + continue + columns, where_found = _expose_schema(expose) + if where_found is None: + continue + key, prefix = _expose_key(i, expose), f"exposes[{i}].{where_found}" + changes.extend( + _duplicate_changes( + columns, + kind="column_duplicate_name", + noun="column", + expose_id=key, + path_prefix=prefix, + ) + ) + changes.extend(_nested_duplicates(columns, expose_id=key, path_prefix=prefix)) + return changes diff --git a/fluid_build/build_runners/_deployment_host_policy.py b/fluid_build/build_runners/_deployment_host_policy.py index a808a701c..d8c526202 100644 --- a/fluid_build/build_runners/_deployment_host_policy.py +++ b/fluid_build/build_runners/_deployment_host_policy.py @@ -302,8 +302,13 @@ def refuse_unless_host_allowed( refusal = host_refusal(url, engine=engine, environ=environ, operator_supplied=operator_supplied) if refusal is None: return str(url).strip() + raise _host_error(refusal, engine=engine, operator_supplied=operator_supplied) + + +def _host_error(refusal: str, *, engine: str, operator_supplied: bool) -> DeploymentHostError: + """The :class:`DeploymentHostError` for ``refusal``, worded for whoever chose the URL.""" if operator_supplied: - raise DeploymentHostError( + return DeploymentHostError( what=f"the {engine} endpoint given on the command line cannot be used", why=f"Refused because {refusal}.", fix=( @@ -313,7 +318,7 @@ def refuse_unless_host_allowed( ), doc=doc_url(), ) - raise DeploymentHostError( + return DeploymentHostError( what=f"the {engine} orchestrator may not receive this product's credential", why=( f"forge-cli hands the whole connector config -- the resolved source password " @@ -332,24 +337,58 @@ def refuse_unless_host_allowed( ) +def _pinned_refusal(verdict: HostVerdict) -> str: + """:func:`host_refusal`'s words for a :func:`pinned_verdict`, naming both. + + The host is what an allowlist entry would list. The address is where the + connection was going, and the only part that tells a rebinding answer from + a host that really is public. + """ + if verdict.code == "no-host": + return "the request has no address to check" + named = f"host {verdict.host!r}" + if verdict.pinned and verdict.pinned != verdict.host: + named += f", connecting to {verdict.pinned!r}," + if verdict.code == "link-local": + return ( + f"{named} is link-local or unspecified, where cloud instance metadata lives; " + "no orchestrator is there, and this is refused even with an allowlist entry" + ) + return ( + f"{named} is public, and the config posted to it carries the source credential " + "in clear text" + ) + + def request_guard( url: Optional[str], *, engine: str, operator_supplied: bool = False ) -> Callable[[Any], None]: - """An httpx request hook that re-applies the policy to ``url``'s host. + """An httpx request hook that judges the address the request is pinned to. The authoritative check. :func:`host_refusal` runs once at construction to fail fast, but its verdict on a host that did not resolve yet is "allow", so the policy has to be asked again when the request actually goes out -- by then the name resolves or the request fails on its own. - It closes over the URL given at construction rather than reading - ``request.url``, because ``safe_httpx_client`` inserts its DNS-pinning hook - ahead of this one and that hook rewrites the URL to the pinned IP. The - hostname is what an allowlist entry names, so the original is what must be - judged. + ``safe_httpx_client`` inserts its DNS-pinning hook ahead of this one, and + that hook rewrites ``request.url`` to the IP the connection will use. This + judges that IP rather than resolving the name again: a second lookup can + disagree with the first (DNS rebinding) or encode a non-ASCII name + differently (Python's IDNA2003 against httpx's IDNA2008). The name in + ``url``, closed over at construction, is still checked as it was there and + still decides the allowlist, because the hostname is what an entry names. """ - def _guard(request: Any) -> None: # pragma: no cover - exercised via the clients + def _guard(request: Any) -> None: refuse_unless_host_allowed(url, engine=engine, operator_supplied=operator_supplied) + verdict = pinned_verdict( + url, + getattr(getattr(request, "url", None), "host", None), + operator_supplied=operator_supplied, + ) + if verdict is not None: + raise _host_error( + _pinned_refusal(verdict), engine=engine, operator_supplied=operator_supplied + ) return _guard diff --git a/fluid_build/build_runners/_ids.py b/fluid_build/build_runners/_ids.py index 5fdb0f241..c68e89c89 100644 --- a/fluid_build/build_runners/_ids.py +++ b/fluid_build/build_runners/_ids.py @@ -47,6 +47,12 @@ # can never traverse out of the directory it is joined into. _IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.\-]{0,127}\Z") +# :data:`_IDENT_RE` in words, for a message that tells an author what to change. +IDENT_RULE = ( + "start with a letter or underscore, use only letters, digits, '_', '.' and '-', " + "and be at most 128 characters" +) + class IdentifierViolation(ValueError): """Raised when a contract identifier fails validation. @@ -103,3 +109,27 @@ def validate_build_ids(contract: Mapping[str, Any]) -> None: for build in builds: if isinstance(build, Mapping) and build.get("id"): validate_identifier(build["id"], kind="build.id") + + +def build_id_warning(contract: Mapping[str, Any]) -> Optional[str]: + """The warning ``fluid validate`` and ``fluid plan`` give for an id the build phase refuses. + + The schema's ``$defs.identifier`` admits more than :data:`_IDENT_RE` (a + leading digit, any length), and the two stay different by decision. So + validate and plan accept such an id and warn, and ``fluid apply`` in a + build mode refuses it. The check is :func:`validate_build_ids`, so the + warning names the id apply would name. Returns ``None`` when it passes. + + The text does not say the schema accepted the id: validate also runs this + on a contract the schema refused, such as one whose ``id`` is ``123``. + """ + try: + validate_build_ids(contract) + except IdentifierViolation as exc: + return ( + f"{exc.kind} {exc.value!r} cannot be used by the build phase, which puts it in " + f"directory names and generated code, so it must {IDENT_RULE}. fluid apply in a " + "build mode (amend-and-build, replace-and-build) refuses this id before it " + "applies anything. Rename it to fit." + ) + return None diff --git a/fluid_build/cli/apply.py b/fluid_build/cli/apply.py index 080c08a12..1ccf09fae 100644 --- a/fluid_build/cli/apply.py +++ b/fluid_build/cli/apply.py @@ -313,7 +313,8 @@ def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None: The build phase turns ``contract.id`` and ``builds[].id`` into directory names, so it takes a narrower grammar than the schema: ``1orders``, or an - id of 129 characters, passes ``fluid validate`` and ``fluid plan``. The + id of 129 characters, passes ``fluid validate`` and ``fluid plan``, which + only warn (``build_runners._ids.build_id_warning``). The build runner checks the ids itself, but on the OpenTofu engine it runs after ``tofu apply``, so such an id was refused once the product was provisioned, as an "Unexpected error". This is the runner's own check @@ -323,7 +324,11 @@ def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None: Raises: CLIError: ``apply_build_identifier_invalid``. """ - from fluid_build.build_runners._ids import IdentifierViolation, validate_build_ids + from fluid_build.build_runners._ids import ( + IDENT_RULE, + IdentifierViolation, + validate_build_ids, + ) try: validate_build_ids(contract) @@ -337,10 +342,8 @@ def _gate_build_ids_for_apply(contract: Dict[str, Any]) -> None: "hint": ( "The build phase keeps each build's run records, cursors and " "watermarks under .fluid/runs///, so both ids " - "must start with a letter or underscore, use only letters, digits, " - "'_', '.' and '-', and be at most 128 characters. The contract schema " - "allows more, so fluid validate and fluid plan accept it. Nothing was " - "applied." + f"must {IDENT_RULE}. The contract schema allows more, so fluid " + "validate and fluid plan only warn about it. Nothing was applied." ), }, ) from exc diff --git a/fluid_build/cli/odcs.py b/fluid_build/cli/odcs.py index b277d913e..bebc294e3 100644 --- a/fluid_build/cli/odcs.py +++ b/fluid_build/cli/odcs.py @@ -504,6 +504,15 @@ def _validate_imported_contract(contract: dict) -> list: return [] if result.is_valid else list(result.errors) +def _brief(message: str, limit: int = 120) -> str: + """``message`` cut to ``limit`` characters, keeping a line-break hint whole: cut, + ``... block keeps and |`` reads as advice to use ``|``.""" + from fluid_build.schema_manager import LINE_BREAK_HINT + + head = message.removesuffix(LINE_BREAK_HINT) + return head[:limit] + message[len(head) :] + + def _run_odcs_validate(args): """Run ODCS validate command — multi-pass with structured report. @@ -559,7 +568,8 @@ def _run_odcs_validate(args): report["ok"] = False cprint(f"✗ jsonschema: {len(errors)} error(s)") for e in errors[:10]: - cprint(f" {e['path'] or ''}: {e['message'][:120]}") + # markup=False: the message quotes the value, and "[/]" in it is a Rich tag. + cprint(f" {e['path'] or ''}: {_brief(e['message'])}", markup=False) if len(errors) > 10: cprint(f" ... +{len(errors) - 10} more") else: diff --git a/fluid_build/cli/plan.py b/fluid_build/cli/plan.py index 7e84c35ac..09390e680 100644 --- a/fluid_build/cli/plan.py +++ b/fluid_build/cli/plan.py @@ -30,7 +30,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional -from fluid_build.cli.console import cprint, warning +from fluid_build.cli.console import cprint, cprint_literal, warning # 11-stage pipeline: stage-6 → stage-7 cryptographic plan-binding. ``plan.json`` # carries ``bundleDigest`` (pins input bundle) + ``planDigest`` (catches @@ -305,6 +305,7 @@ def run(args, logger: logging.Logger) -> int: # plan generation or write. A failure raises ``CLIError`` → # non-zero exit, no plan.json written. _gate_contract_for_plan_or_apply(contract, logger, command="plan") + _warn_unusable_build_ids(contract) # --- --provider gate ------------------------------------------------- # ``--provider`` reaches ``build_provider`` only on the legacy planner @@ -780,6 +781,27 @@ def _gate_contract_for_plan_or_apply( ) +def _warn_unusable_build_ids(contract: Dict[str, Any]) -> None: + """Warn when the build phase will refuse ``contract.id`` or a ``builds[].id``. + + The schema admits ids the build phase refuses (``1orders``, or one of 129 + characters), and ``fluid apply`` in a build mode refuses such an id before + it applies anything. Plan only warns, whatever ``--mode`` it plans for, and + writes the plan. Not part of :func:`_gate_contract_for_plan_or_apply`: + apply shares that gate and refuses the id with a typed error of its own + (``cli/apply.py::_gate_build_ids_for_apply``). Printed, not logged: a + ``warn()`` event would print the same text again as a JSON line. + """ + from fluid_build.build_runners._ids import build_id_warning + + message = build_id_warning(contract) + if message is not None: + # As validate prints its warnings (validate._cprint_literal): literal, + # because the id is contract data, not markup; the terminal wraps the + # line, so Rich never breaks one inside the id; and no highlighter. + cprint_literal(f"\n⚠️ {message}", highlight=False, soft_wrap=True) + + def _validate_plan_actions(plan: Dict[str, Any], logger: logging.Logger) -> None: """Run SDK ``validate_actions()`` over the plan's action list. diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py index 0183b04e2..c47696400 100644 --- a/fluid_build/cli/validate.py +++ b/fluid_build/cli/validate.py @@ -1186,6 +1186,9 @@ def _run_contract_rules( "the contract was NOT checked for a continuous trigger on a batch build" ) + # --- An id the build phase refuses (WARNING) ----------------------- + _build_id_check(validation_result, contract) + # --- Packaging checks (RFC-packaging-modes.md file 9) --------------- # ``fluid validate`` had NO packaging awareness: a `shared` block with # no `pool` validated clean and then failed at `fluid plan` with an @@ -1306,6 +1309,32 @@ def _databricks_binding_check(validation_result, contract: Mapping[str, Any]) -> validation_result.add_warning(msg) +def _build_id_check(validation_result, contract: Mapping[str, Any]) -> None: + """A ``contract.id`` or ``builds[].id`` the build phase refuses, as a warning. + + The schema's ``$defs.identifier`` admits ids the build phase refuses + (``1orders``, or more than 128 characters), and the two grammars stay + different by decision. ``fluid apply`` in a build mode refuses such an id + before it applies anything, with the same rule + (``build_runners._ids.validate_build_ids``). A warning, not an error: plain + validate passes and ``--strict`` fails. A crash in the check is reported, + never swallowed. + """ + try: + from fluid_build.build_runners._ids import build_id_warning + + message = build_id_warning(contract) + except Exception as exc: # noqa: BLE001 — reported, never swallowed + validation_result.add_warning( + f"Build-id check could not run ({type(exc).__name__}: {exc}); " + "contract.id and builds[].id were NOT checked against the ids the build " + "phase accepts" + ) + return + if message is not None: + validation_result.add_warning(message) + + def _duplicate_contract_id_errors(contract: Mapping[str, Any]) -> List[str]: """Refuse two exposes whose output ports would be published under one contract id. diff --git a/fluid_build/credentials/litellm_endpoints.py b/fluid_build/credentials/litellm_endpoints.py index 67af1e4c8..53d71d680 100644 --- a/fluid_build/credentials/litellm_endpoints.py +++ b/fluid_build/credentials/litellm_endpoints.py @@ -24,9 +24,9 @@ project's own setting and follows the name-shape rule. Vendored because importing litellm takes seconds and dotenv hydration runs on -every command. Generated from the litellm 1.83.14, 1.97.0, 1.105.0 and -1.106.0 wheels (FLUID's floor, the last release Python 3.10 can import, and the -two newest), as the union of: +every command. Generated from the litellm 1.83.14, 1.97.0, 1.105.0, 1.106.0 +and 1.108.0 wheels (FLUID's floor, the last release Python 3.10 can import, and +the newest), as the union of: * ``_API_BASE`` for every ``litellm.provider_list`` entry (``-`` read as ``_``); @@ -83,6 +83,7 @@ "BASETEN_API_BASE", "BEDROCK_API_BASE", "BEDROCK_MANTLE_API_BASE", + "BESPOKE_API_BASE", "BFL_API_BASE", "BLACK_FOREST_LABS_API_BASE", "BRAVE_API_BASE", @@ -100,6 +101,7 @@ "COMETAPI_API_BASE", "COMETAPI_BASE_URL", "COMPACTIFAI_API_BASE", + "CORALBRICKS_API_BASE", "CORTECS_API_BASE", "CRUSOE_API_BASE", "CRW_API_BASE", @@ -108,6 +110,7 @@ "CUSTOM_OPENAI_API_BASE", "DARKBLOOM_API_BASE", "DASHSCOPE_API_BASE", + "DATABRICKS_AGENT_API_BASE", "DATABRICKS_API_BASE", "DATAFORSEO_API_BASE", "DATAROBOT_API_BASE", @@ -156,6 +159,7 @@ "HUMANLOOP_API_BASE", "HYPERBOLIC_API_BASE", "INCEPTION_API_BASE", + "INFERENCEAPIS_API_BASE", "INFINITY_API_BASE", "JINA_AI_API_BASE", "LAMBDA_AI_API_BASE", @@ -163,6 +167,7 @@ "LANGFLOW_API_BASE", "LANGFUSE_API_BASE", "LANGGRAPH_API_BASE", + "LAYA_API_BASE", "LEMONADE_API_BASE", "LIBERTAI_API_BASE", "LINKUP_API_BASE", @@ -175,6 +180,7 @@ "MARITALK_API_BASE", "META_API_BASE", "META_LLAMA_API_BASE", + "MICROSOFT_365_COPILOT_API_BASE", "MILVUS_API_BASE", "MINIMAX_API_BASE", "MISTRAL_API_BASE", @@ -228,6 +234,7 @@ "SAIL_API_BASE", "SAMBANOVA_API_BASE", "SAP_API_BASE", + "SCALEDOWN_API_BASE", "SCALEWAY_API_BASE", "SCX_AI_API_BASE", "SCX_API_BASE", diff --git a/fluid_build/extension_schemas.py b/fluid_build/extension_schemas.py index db8304993..98a3555ce 100644 --- a/fluid_build/extension_schemas.py +++ b/fluid_build/extension_schemas.py @@ -22,7 +22,8 @@ customScaffold = "data_product_forge_custom_scaffold.validation:get_extension_schema" The provider has the signature ``get_extension_schema(fluid_version=None) -> dict`` -and returns a draft-07 JSON Schema describing the data *under* the extension key. +and returns a JSON Schema describing the data *under* the extension key: draft-07, +unless the schema declares another dialect in its ``$schema``. These helpers let the ``fluid forge`` copilot (a) ground contract generation on the installed extension schemas and (b) validate the generated blocks before @@ -200,6 +201,7 @@ def assemble_proposed_extensions( from jsonschema import Draft7Validator except ImportError: # pragma: no cover — jsonschema is a hard dependency Draft7Validator = None # type: ignore[assignment] + from fluid_build.schema_manager import schema_validator out: Dict[str, Any] = {} for key, block in proposed.items(): @@ -207,7 +209,19 @@ def assemble_proposed_extensions( if schema is None: continue if Draft7Validator is not None: - if next(iter(Draft7Validator(schema).iter_errors(block)), None) is not None: + # The dialect the plugin's schema declares, draft-07 when it declares + # none; Draft 7 ignores a newer keyword rather than enforcing it. A + # pattern's "$" is the end of the value, as ECMA-262 reads it. + try: + validator = schema_validator(schema, default=Draft7Validator) + invalid = next(iter(validator.iter_errors(block)), None) is not None + except Exception as e: # a schema that cannot be applied, e.g. a non-string $schema + from fluid_build.observability.secret_redactor import redact_secret_text + + log = logger or logging.getLogger(__name__) + log.warning("%s", redact_secret_text(f"extension schema {key!r} unusable: {e}")) + continue # unchecked — don't emit it + if invalid: continue # invalid — let the repair loop surface it; don't emit junk out[key] = block return out diff --git a/fluid_build/forge/core/artifact_validators.py b/fluid_build/forge/core/artifact_validators.py index 5577ffd9c..1d0c99f97 100644 --- a/fluid_build/forge/core/artifact_validators.py +++ b/fluid_build/forge/core/artifact_validators.py @@ -36,7 +36,6 @@ from __future__ import annotations import hashlib -import importlib.util import json import logging import shutil @@ -113,10 +112,6 @@ def _load_schema(path: Path) -> Optional[Dict[str, Any]]: # --------------------------------------------------------------------------- -def _jsonschema_available() -> bool: - return importlib.util.find_spec("jsonschema") is not None - - def _conftest_available() -> bool: """``conftest`` binary on PATH — independent of python package install.""" return shutil.which("conftest") is not None @@ -306,7 +301,10 @@ def _validate_against_schema( ) ] - if not _jsonschema_available(): + from fluid_build.schema_manager import JSONSCHEMA_AVAILABLE, schema_validator + + # The flag schema_validator needs: a package that is found can still fail to import. + if not JSONSCHEMA_AVAILABLE: return [ ValidationIssue( file=artifact_path, @@ -320,8 +318,6 @@ def _validate_against_schema( ) ] - import jsonschema # type: ignore - issues: List[ValidationIssue] = [] # Validate with the dialect the schema DECLARES, not a hardcoded one. # @@ -335,7 +331,10 @@ def _validate_against_schema( # and guards nine objects with `unevaluatedProperties: false`. A server entry # carrying an unexpected key validated with ZERO errors, reachable from # `fluid validate-artifacts`. See tests/test_jsonschema_dialect.py. - validator = jsonschema.validators.validator_for(schema)(schema) + # + # schema_validator also reads a pattern's `$` as the end of the value, as + # ECMA-262 does: ODCS refused `schema[].id: "t 1"` but took `"t1\n"`. + validator = schema_validator(schema) errors = sorted(validator.iter_errors(doc), key=lambda e: list(e.absolute_path)) for err in errors: @@ -397,7 +396,9 @@ def validate_opds(path: str, content: bytes) -> List[ValidationIssue]: ) ] - if not _jsonschema_available(): + from fluid_build.schema_manager import JSONSCHEMA_AVAILABLE + + if not JSONSCHEMA_AVAILABLE: return [ ValidationIssue( file=path, diff --git a/fluid_build/forge_datamodel/emit/osi_sidecar.py b/fluid_build/forge_datamodel/emit/osi_sidecar.py index d9d5f2f10..206749753 100644 --- a/fluid_build/forge_datamodel/emit/osi_sidecar.py +++ b/fluid_build/forge_datamodel/emit/osi_sidecar.py @@ -93,14 +93,14 @@ def validate_osi_document(document: Dict[str, Any]) -> List[str]: Returns a sorted list of human-readable issue strings (empty when conformant). ``jsonschema`` is imported lazily — this module sits on the forge pipeline path, not the CLI cold path, but there is no - reason to tax importers that never validate. + reason to tax importers that never validate. A ``pattern``'s ``$`` is + the end of the value, as ECMA-262 reads it. """ - import jsonschema + from fluid_build.schema_manager import schema_validator schema = _load_ossie_schema() - validator_cls = jsonschema.validators.validator_for(schema) errors = sorted( - validator_cls(schema).iter_errors(document), + schema_validator(schema).iter_errors(document), key=lambda e: ([str(p) for p in e.absolute_path], e.message), ) return [f"{'/'.join(str(p) for p in e.absolute_path) or ''}: {e.message}" for e in errors] diff --git a/fluid_build/providers/odcs/validation.py b/fluid_build/providers/odcs/validation.py index 7cb2260db..9711f5822 100644 --- a/fluid_build/providers/odcs/validation.py +++ b/fluid_build/providers/odcs/validation.py @@ -74,15 +74,17 @@ def collect_errors(odcs: Mapping[str, Any], schema: Mapping[str, Any]) -> List[D Empty list = valid. No-op (empty list) when :mod:`jsonschema` is not installed. Used by ``validate()`` (which raises on non-empty) and by ``fluid odcs validate --report`` (which serializes to JSON for CI). + + A ``pattern``'s ``$`` is the end of the value, as ECMA-262 reads it, so + ``schema[].id: "t1\\n"`` fails ``^[A-Za-z0-9_-]+$`` like ``"t 1"`` does. """ - try: - import jsonschema - except ImportError: + from fluid_build.schema_manager import JSONSCHEMA_AVAILABLE, schema_validator + + if not JSONSCHEMA_AVAILABLE: LOG.warning("jsonschema not installed, skipping ODCS validation") return [] - validator_cls = jsonschema.validators.validator_for(schema) - validator = validator_cls(schema) + validator = schema_validator(schema) out: List[Dict[str, Any]] = [] for err in validator.iter_errors(odcs): out.append( diff --git a/fluid_build/providers/odps_standard/validation.py b/fluid_build/providers/odps_standard/validation.py index b6428c2ea..7f33d6bca 100644 --- a/fluid_build/providers/odps_standard/validation.py +++ b/fluid_build/providers/odps_standard/validation.py @@ -75,13 +75,18 @@ def schema_for_document(odps: Mapping[str, Any]) -> Optional[Dict[str, Any]]: def validate(odps: Mapping[str, Any], schema: Mapping[str, Any]) -> None: - """Validate an ODPS product against the vendored v1.0.0 JSON Schema.""" + """Validate an ODPS product against ``schema``, the vendored one for its + ``apiVersion`` (:func:`schema_for_document`).""" try: import jsonschema except ImportError: LOG.warning("jsonschema not installed, skipping ODPS validation") return + from fluid_build.schema_manager import schema_validator + try: - jsonschema.validate(instance=odps, schema=schema) + # cls: the declared dialect, with a pattern's "$" read as the end of the + # value, as ECMA-262 does. jsonschema.validate still checks the schema first. + jsonschema.validate(instance=odps, schema=schema, cls=type(schema_validator(schema))) except jsonschema.ValidationError as exc: raise ProviderError(f"ODPS validation failed: {exc.message}") from exc diff --git a/fluid_build/providers/opds/validator.py b/fluid_build/providers/opds/validator.py index 1f131cf07..c75b00803 100644 --- a/fluid_build/providers/opds/validator.py +++ b/fluid_build/providers/opds/validator.py @@ -119,8 +119,14 @@ def validate_against_opds_schema( valid, errors = _basic_validation(opds_data, version) return valid, errors, "basic" + from fluid_build.schema_manager import schema_validator + try: - validator = jsonschema.Draft202012Validator(schema) + # The dialect the schema declares (v4.1: 2020-12, also the default), + # with a pattern's "$" read as the end of the value, as ECMA-262 does. + # The patternProperties language-code keys (^[a-z]{2}$) keep + # jsonschema's own re.search; see schema_manager.schema_validator. + validator = schema_validator(schema, default=jsonschema.Draft202012Validator) errors = list(validator.iter_errors(opds_data)) # Filter a known upstream OPDS v4.1 schema false-positive: the diff --git a/fluid_build/schema_manager.py b/fluid_build/schema_manager.py index a157506fb..97f7d5c0a 100644 --- a/fluid_build/schema_manager.py +++ b/fluid_build/schema_manager.py @@ -28,7 +28,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path -from typing import Any, Callable, Dict, FrozenSet, Iterator, List, Optional, Union +from typing import Any, Callable, Dict, FrozenSet, Iterator, List, Mapping, Optional, Union import yaml @@ -43,6 +43,7 @@ # anywhere in this module — and RefResolver is deprecated, so the day a # jsonschema release drops it this ImportError would have silently disabled # schema validation across the CLI. Never name more in a probe than you use. +_JSONSCHEMA_IMPORT_ERROR: Optional[ImportError] = None try: import jsonschema @@ -66,6 +67,9 @@ #: Validators whose jsonschema message names only keys, never a value. _KEY_ONLY_VALIDATORS = frozenset({"required", "additionalProperties"}) +#: Ends a ``pattern`` error when the value's final line break is all that fails it. +LINE_BREAK_HINT = " (it ends in a line break, which a YAML | or > block keeps and |- or >- drops)" + def format_schema_error(error: Any) -> str: """``error.message``, without the failing value under :data:`_UNQUOTED_VALUE_PATHS`. @@ -175,7 +179,7 @@ def _pattern_keyword(validator: Any, pattern: str, instance: Any, schema: Any) - return message = f"{instance!r} does not match {pattern!r}" if re.search(pattern, instance): # only the final line break fails - message += " (it ends in a line break, which a YAML | or > block keeps and |- or >- drops)" + message += LINE_BREAK_HINT yield jsonschema.ValidationError(message) @@ -184,20 +188,32 @@ def _end_anchored_validator(validator_class: Any) -> Any: return jsonschema.validators.extend(validator_class, {"pattern": _pattern_keyword}) -def schema_validator(schema: Dict[str, Any]) -> Any: +def schema_validator(schema: Mapping[str, Any], default: Any = None) -> Any: """A jsonschema validator for ``schema``, in the dialect it declares, whose ``pattern`` ends at the end of the value (:func:`_end_anchored`). Validate - against a FLUID schema through this, never ``validator_for`` directly. + against a FLUID schema, or a vendored or plugin one, through this, never + ``validator_for`` directly. ``default`` is the dialect of a schema that + declares none, jsonschema's newest when not given. Raises ``ImportError`` + when jsonschema could not be imported; check :data:`JSONSCHEMA_AVAILABLE` + first to skip instead. The keyword is replaced the way check-jsonschema replaces it, with ``jsonschema.validators.extend``. Two places keep jsonschema's own ``re.search``: ``patternProperties`` keys, as its ``additionalProperties`` and :func:`_locate_error` read them, and a subschema that declares its own - ``$schema`` (or a ``$ref`` to the root, which does), for which jsonschema - picks that dialect's own class. No FLUID schema has either; a test fails if - one does. + ``$schema`` (or a ``$ref``, ``$recursiveRef`` or ``$dynamicRef`` to the + root, which does), for which jsonschema picks that dialect's own class. No + FLUID schema has either. Of the vendored ones, only OPDS keys + ``patternProperties``, with ``^[a-z]{2}$`` language codes. A test fails if + either changes (``_KEY_PATTERNS`` in tests/test_schema_pattern_end_anchor.py). """ - return _end_anchored_validator(jsonschema.validators.validator_for(schema))(schema) + if not JSONSCHEMA_AVAILABLE: + raise ImportError("jsonschema is not available") from _JSONSCHEMA_IMPORT_ERROR + if default is None: + validator_class = jsonschema.validators.validator_for(schema) + else: + validator_class = jsonschema.validators.validator_for(schema, default=default) + return _end_anchored_validator(validator_class)(schema) def _schema_version_sort_key(version: str) -> tuple[int, int, int]: diff --git a/fluid_build/streaming/compiler.py b/fluid_build/streaming/compiler.py index 5afa62194..1a42ecd56 100644 --- a/fluid_build/streaming/compiler.py +++ b/fluid_build/streaming/compiler.py @@ -35,6 +35,7 @@ ) from fluid_build.providers._iceberg_catalog_auth import FORBIDDEN_TEXT_PATTERN, parse_catalog_auth from fluid_build.providers._sql_safety import quote_ansi_string_literal, validate_ident +from fluid_build.schema_manager import schema_validator from fluid_build.streaming.mapping import SOURCE_TABLE, bind_by_name, compile_select, parse_stream from fluid_build.streaming.types import LogicalType, columns_sql, parse_fields from fluid_build.util.contract import is_externally_run_build @@ -236,9 +237,10 @@ def compile_build(contract: Mapping[str, Any], build: Mapping[str, Any]) -> Flin f"parameters.flink.source.{moved[0]}: the topic, format and schema of the records " "are declared in properties.stream; parameters.flink.source keeps the connection" ) - errors = sorted( - jsonschema.Draft202012Validator(_CONFIG).iter_errors(cfg), key=lambda e: str(e.path) - ) + # schema_validator, so that a pattern ends at the end of the value, not + # before a trailing newline as jsonschema's own re.search lets it. + validator = schema_validator(_CONFIG, default=jsonschema.Draft202012Validator) + errors = sorted(validator.iter_errors(cfg), key=lambda e: str(e.path)) if errors: # Do not echo instance values: an invalid value might be a pasted credential. problem = errors[0] diff --git a/tests/_infrastructure/respx_fixtures.py b/tests/_infrastructure/respx_fixtures.py index 58b4e5cf4..74e04d19f 100644 --- a/tests/_infrastructure/respx_fixtures.py +++ b/tests/_infrastructure/respx_fixtures.py @@ -46,11 +46,26 @@ def requires_respx(reason: str = "respx not installed") -> Any: return pytest.mark.skipif(respx is None, reason=reason) +def _list_orchestrator(monkeypatch: pytest.MonkeyPatch, host: str) -> None: + """Name a mock orchestrator in the operator allowlist, as its owner would. + + The unit suite replaces safe_http's pin hook with a no-op (tests/conftest.py), + so the deployment-host request guard sees this name where production sees + the IP the request was pinned to. A name that does not resolve is no address + the guard can allow, so it is refused unless listed. Listing it keeps the + link-local refusal in force. + """ + from fluid_build.build_runners._deployment_host_policy import ALLOWLIST_ENV + + monkeypatch.setenv(ALLOWLIST_ENV, host) + + @pytest.fixture -def airbyte_mock() -> Iterator["AirbyteMockServer"]: +def airbyte_mock(monkeypatch: pytest.MonkeyPatch) -> Iterator["AirbyteMockServer"]: """In-memory Airbyte API. Tests mutate state via ``airbyte_mock.create_source(...)``; runner code under test sees realistic CRUD responses. """ + _list_orchestrator(monkeypatch, "airbyte.test") server = AirbyteMockServer() with respx.mock(base_url="https://airbyte.test", assert_all_called=False) as router: server.attach(router) @@ -58,7 +73,8 @@ def airbyte_mock() -> Iterator["AirbyteMockServer"]: @pytest.fixture -def kafka_connect_mock() -> Iterator["KafkaConnectMockServer"]: +def kafka_connect_mock(monkeypatch: pytest.MonkeyPatch) -> Iterator["KafkaConnectMockServer"]: + _list_orchestrator(monkeypatch, "kafka-connect.test") server = KafkaConnectMockServer() with respx.mock(base_url="http://kafka-connect.test:8083", assert_all_called=False) as router: server.attach(router) diff --git a/tests/build_runners/test_deployment_host_policy.py b/tests/build_runners/test_deployment_host_policy.py index bc9bfe20a..a69868503 100644 --- a/tests/build_runners/test_deployment_host_policy.py +++ b/tests/build_runners/test_deployment_host_policy.py @@ -47,6 +47,10 @@ patched where the runner looks the name up. 3. **The per-request guard really fires.** Asserting ``len(event_hooks) >= 2`` passed for a no-op guard; this fires a request and asserts nothing is sent. +4. **The guard judges the address the request is pinned to.** It resolved the + name a second time, so DNS rebinding, or ``faß.de`` looked up as IDNA2003 + ``fass.de`` while httpx connects to IDNA2008 ``xn--fa-hia.de``, could pass + a name whose connection went to an unlisted public address or to metadata. """ from __future__ import annotations @@ -56,6 +60,7 @@ import socket import threading from http.server import BaseHTTPRequestHandler, HTTPServer +from types import SimpleNamespace from typing import Any, Dict, Iterator, List, Tuple import pytest @@ -81,6 +86,11 @@ "orchestrator.mixed.test": ["fd00::1", "93.184.216.34"], "orchestrator.v6public.test": ["2606:2800:220:1:248:1893:25c8:1946"], "orchestrator.mappedprivate.test": ["::ffff:10.1.2.3"], + # Private when asked a second time, whatever the pinning lookup got. + "orchestrator.rebind.test": ["10.1.2.3"], + # What socket.getaddrinfo answers for faß.de: it asks about IDNA2003 + # fass.de, while httpx connects to IDNA2008 xn--fa-hia.de. + "faß.de": ["10.1.2.3"], } @@ -227,6 +237,7 @@ def test_the_address_predicate_is_shared_not_copied() -> None: ("file:///etc/passwd", "", False, "not an http(s) orchestrator"), ("", "", False, "nothing configured"), ("https://", "", False, "no host"), + ("http://[::1", "", False, "an unbalanced bracket has no host"), ], ) def test_the_policy_matrix(url: str, listed: str, allowed: bool, because: str) -> None: @@ -397,21 +408,77 @@ def log_message(self, *_a: Any) -> None: thread.join(timeout=5) -def test_the_guard_judges_the_original_host_not_the_pinned_ip() -> None: - """The SSRF hook rewrites the URL to an IP before this one runs.""" +def _pinned(address: str) -> Any: + """A request as the guard sees it: safe_httpx_client's pin hook has put the IP in.""" + return SimpleNamespace(url=SimpleNamespace(host=address)) - class _Explodes: - @property - def url(self) -> str: - raise AssertionError("the guard must not read the rewritten request URL") - import os +@pytest.mark.parametrize( + "url, pinned, listed, refused", + [ + # The guard's own lookup of the name says private (rebinding, or the + # IDNA2003 spelling of faß.de), yet the connection is pinned to a + # public address: refused. + ("https://orchestrator.rebind.test", "93.184.216.34", "", True), + ("https://faß.de", "93.184.216.34", "", True), + ("http://orchestrator.rebind.test:8083", "10.1.2.3", "", False), + # A listed name may connect to its public address, never to metadata. + ("https://orchestrator.public.test", "93.184.216.34", "orchestrator.public.test", False), + ("https://orchestrator.public.test", "169.254.169.254", "orchestrator.public.test", True), + ("https://orchestrator.public.test", "0.0.0.0", "orchestrator.public.test", True), + # No address to judge is not a pass. + ("https://orchestrator.public.test", "", "orchestrator.public.test", True), + ], +) +def test_the_guard_judges_the_pinned_address_not_a_second_lookup( + monkeypatch: pytest.MonkeyPatch, url: str, pinned: str, listed: str, refused: bool +) -> None: + """The pin hook runs first and puts the IP the connection will use in the URL. - os.environ[ALLOWLIST_ENV] = "orchestrator.public.test" - try: - request_guard("https://orchestrator.public.test", engine="airbyte")(_Explodes()) - finally: - del os.environ[ALLOWLIST_ENV] + Resolving the name again can answer differently: DNS rebinding between the + two lookups, or Python's IDNA2003 against httpx's IDNA2008. The name still + decides the allowlist; the pinned address decides the scope. + """ + monkeypatch.setenv(ALLOWLIST_ENV, listed) + guard = request_guard(url, engine="kafka-connect") + if refused: + with pytest.raises(DeploymentHostError): + guard(_pinned(pinned)) + else: + guard(_pinned(pinned)) + + +def test_a_pinned_refusal_names_the_host_to_list_and_the_address_it_reached() -> None: + with pytest.raises(DeploymentHostError) as excinfo: + request_guard("https://orchestrator.rebind.test", engine="debezium")( + _pinned("93.184.216.34") + ) + error = excinfo.value + assert "'orchestrator.rebind.test', connecting to '93.184.216.34'," in (error.why or "") + assert "debezium" in (error.why or "") + assert f'{ALLOWLIST_ENV}="connect.example.com"' in (error.fix or "") + + +def test_an_operator_chosen_url_still_never_connects_to_metadata() -> None: + guard = request_guard( + "https://orchestrator.public.test", engine="airbyte", operator_supplied=True + ) + guard(_pinned("93.184.216.34")) + with pytest.raises(DeploymentHostError) as excinfo: + guard(_pinned("169.254.169.254")) + error = excinfo.value + rendered = " ".join([str(error), error.why or "", error.fix or ""]) + assert "connecting to '169.254.169.254'" in rendered + assert "contract" not in rendered, rendered + + +def test_a_malformed_url_is_refused_not_raised() -> None: + """``http://[::1`` made ``urlsplit`` raise ValueError through the policy.""" + with pytest.raises(DeploymentHostError) as excinfo: + refuse_unless_host_allowed("http://[::1", engine="kafka-connect") + assert "no host" in (excinfo.value.why or "") + with pytest.raises(DeploymentHostError): + request_guard("http://[::1", engine="kafka-connect")(_pinned("127.0.0.1")) @pytest.mark.parametrize( diff --git a/tests/build_runners/test_deployment_host_policy_real_pin_hook.py b/tests/build_runners/test_deployment_host_policy_real_pin_hook.py new file mode 100644 index 000000000..08b48a81b --- /dev/null +++ b/tests/build_runners/test_deployment_host_policy_real_pin_hook.py @@ -0,0 +1,123 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""The deployment-host guard judges what the REAL pin hook pinned. + +``tests/conftest.py`` replaces safe_http's pin hook with a no-op for the unit +suite, so ``test_deployment_host_policy.py`` hands the guard a stand-in +request. That cannot catch the one ordering the fix relies on: the pin hook +runs first and writes the IP the connection will use into ``request.url``, +and the guard reads it there. This file is on that fixture's real-guard list, +so the clients below run the real ``safe_httpx_client`` stack. A guard moved +ahead of the pin hook, or one that resolves the name again, lets the request +reach the transport and fails here. + +DNS is faked and the transport records instead of sending, so nothing leaves +the machine. +""" + +from __future__ import annotations + +import socket +import sys +from typing import Any, Iterator, List, Tuple + +import httpx +import pytest + +from fluid_build.build_runners._deployment_host_policy import ALLOWLIST_ENV, DeploymentHostError + +PUBLIC = "93.184.216.34" +PRIVATE = "10.1.2.3" + +_REAL_GETADDRINFO = socket.getaddrinfo + + +def _answer(address: str) -> List[Tuple[Any, ...]]: + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, 0))] + + +def _getaddrinfo(host: str, *a: Any, **k: Any) -> List[Tuple[Any, ...]]: + if host == "rebind.test": + # The pin hook's lookup gets the attacker's public answer; every + # lookup the policy makes of its own gets a private one. + pinning = sys._getframe(1).f_code.co_name == "resolve_and_validate" + return _answer(PUBLIC if pinning else PRIVATE) + if host == "faß.de": # what Python's resolver asks about: IDNA2003 fass.de + return _answer(PRIVATE) + if host == "xn--fa-hia.de": # what httpx connects to: IDNA2008 + return _answer(PUBLIC) + if host == "connect.internal": + return _answer(PRIVATE) + return _REAL_GETADDRINFO(host, *a, **k) + + +@pytest.fixture(autouse=True) +def sent(monkeypatch: pytest.MonkeyPatch) -> Iterator[List[str]]: + """The URLs that reached the transport, which answers instead of sending.""" + reached: List[str] = [] + + def _record(self: Any, request: httpx.Request) -> httpx.Response: + reached.append(str(request.url)) + return httpx.Response(200, json={}, request=request) + + monkeypatch.setattr(socket, "getaddrinfo", _getaddrinfo) + monkeypatch.setattr(httpx.HTTPTransport, "handle_request", _record) + monkeypatch.delenv(ALLOWLIST_ENV, raising=False) + yield reached + + +@pytest.mark.parametrize( + "url", + [ + "http://rebind.test:8083", # DNS rebinding between the two lookups + "http://faß.de:8083", # IDNA2003 against httpx's IDNA2008 + ], +) +def test_a_name_that_looks_private_twice_is_refused_at_its_pinned_address( + url: str, sent: List[str] +) -> None: + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + client = KafkaConnectRestClient(url) # every check of the name says private + try: + with pytest.raises(DeploymentHostError) as excinfo: + client.list_connectors() + finally: + client.close() + assert sent == [], f"the config must not leave for {PUBLIC}, got {sent}" + assert f"connecting to '{PUBLIC}'" in (excinfo.value.why or "") + + +def test_the_airbyte_client_is_guarded_the_same_way(sent: List[str]) -> None: + from fluid_build.build_runners.airbyte.runner import AirbyteRestClient + + client = AirbyteRestClient("https://rebind.test", api_token="t") + try: + with pytest.raises(DeploymentHostError): + client.list_sources("workspace") + finally: + client.close() + assert sent == [] + + +def test_a_private_orchestrator_is_reached_at_its_pinned_address(sent: List[str]) -> None: + from fluid_build.build_runners.kafka_connect.runner import KafkaConnectRestClient + + client = KafkaConnectRestClient("http://connect.internal:8083") + try: + client.list_connectors() + finally: + client.close() + assert sent == [f"http://{PRIVATE}:8083/connectors"] diff --git a/tests/cli/test_validate_plan_build_ids_warning.py b/tests/cli/test_validate_plan_build_ids_warning.py new file mode 100644 index 000000000..5fb393bf3 --- /dev/null +++ b/tests/cli/test_validate_plan_build_ids_warning.py @@ -0,0 +1,229 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""``fluid validate`` and ``fluid plan`` warn about an id the build phase refuses. + +The schema's ``$defs.identifier`` admits ids the build phase refuses (``1orders``, +or an id longer than 128 characters), and the two grammars stay different by +decision. ``fluid apply`` in a build mode refuses such an id before it applies +anything (``test_apply_build_ids_before_tofu.py``). Validate and plan used to say +nothing about it. Now validate warns, so ``--strict`` fails, and plan warns and +still writes the plan, whatever its ``--mode``. The warning comes from the rule +apply uses, ``build_runners._ids.validate_build_ids``, so the three agree on +which ids are refused. +""" + +from __future__ import annotations + +import copy +import json +import logging +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Dict + +import pytest +import yaml + +import fluid_build.cli as cli_mod +from fluid_build.build_runners import _ids +from fluid_build.cli.plan import _gate_contract_for_plan_or_apply, _warn_unusable_build_ids +from fluid_build.cli.plan import run as plan_run +from fluid_build.cli.validate import _run_contract_rules +from fluid_build.cli.validate import run as validate_run +from fluid_build.schema_manager import ValidationResult +from tests.cli.test_apply_build_ids_before_tofu import _CONTRACT, _UNUSABLE, _with + +LOG = logging.getLogger("test.validate_plan_build_ids_warning") + +# What the warning says besides the id, in the words an author acts on. +_RULE = "start with a letter or underscore" +_REFUSED_BY = "fluid apply in a build mode" + + +def _squash(text: str) -> str: + """``text`` without whitespace: Rich folds a long id across lines.""" + return "".join(text.split()) + + +def _names(out: str, field: str, value: str) -> bool: + return _squash(f"{field} {value!r}") in _squash(out) + + +def _without_builds(contract: Dict[str, Any]) -> Dict[str, Any]: + contract = copy.deepcopy(contract) + del contract["builds"] + return contract + + +@pytest.fixture +def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("FLUID_ENV", raising=False) + return tmp_path + + +def _write(contract: Dict[str, Any]) -> str: + path = Path("contract.fluid.yaml") + path.write_text(yaml.safe_dump(contract, sort_keys=False), encoding="utf-8") + return str(path) + + +def _validate(contract: Dict[str, Any], *flags: str) -> int: + """``fluid validate contract.fluid.yaml --offline `` through the real parser.""" + argv = ["validate", _write(contract), "--offline", *flags] + return validate_run(cli_mod.build_parser().parse_args(argv), LOG) + + +def _plan(contract: Dict[str, Any], *flags: str) -> int: + """``fluid plan contract.fluid.yaml --out plan.json `` through the real parser.""" + argv = ["plan", _write(contract), "--out", "plan.json", *flags] + return plan_run(cli_mod.build_parser().parse_args(argv), LOG) + + +class TestTheWarning: + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_it_names_the_id_the_rule_and_the_refusal(self, field: str, value: str) -> None: + message = _ids.build_id_warning(_with(field, value)) + assert message is not None + assert message.startswith(f"{field} {value!r} ") + assert _RULE in message + assert "'_', '.' and '-'" in message + assert "at most 128 characters" in message + assert _REFUSED_BY in message + + def test_usable_ids_get_none(self) -> None: + assert _ids.build_id_warning(copy.deepcopy(_CONTRACT)) is None + + @pytest.mark.parametrize( + "contract", + [ + pytest.param(_without_builds(_with("contract.id", "1orders")), id="no-builds"), + pytest.param({"id": "1orders", "builds": []}, id="empty-builds"), + pytest.param({"id": "", "builds": [{"id": ""}]}, id="empty-ids"), + pytest.param({"builds": [{"engine": "sql"}]}, id="absent-ids"), + pytest.param({"id": "orders", "builds": ["1make_rows", None]}, id="non-mapping"), + ], + ) + def test_what_apply_does_not_check_gets_none(self, contract: Dict[str, Any]) -> None: + """The ids ``validate_build_ids`` skips are not warned about either.""" + assert _ids.build_id_warning(contract) is None + + +@pytest.mark.usefixtures("workspace") +class TestValidate: + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_plain_validate_warns_and_passes( + self, capsys: pytest.CaptureFixture[str], field: str, value: str + ) -> None: + assert _validate(_with(field, value)) == 0 + out = capsys.readouterr().out + assert _names(out, field, value) + assert _squash(_REFUSED_BY) in _squash(out) + + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_strict_fails(self, capsys: pytest.CaptureFixture[str], field: str, value: str) -> None: + assert _validate(_with(field, value), "--strict") == 1 + out = " ".join(capsys.readouterr().out.split()) + assert "warnings treated as errors (--strict)" in out + + def test_json_lists_the_warning(self, capsys: pytest.CaptureFixture[str]) -> None: + assert _validate(_with("contract.id", "1orders"), "--format", "json") == 0 + report = json.loads(capsys.readouterr().out) + assert report["valid"] is True + assert [w for w in report["warnings"] if w.startswith("contract.id '1orders' ")] + + def test_usable_ids_pass_strict(self) -> None: + """The base contract is clean, so the warning is what fails ``--strict`` above.""" + assert _validate(copy.deepcopy(_CONTRACT), "--strict") == 0 + + def test_a_contract_without_builds_is_not_warned(self) -> None: + assert _validate(_without_builds(_with("contract.id", "1orders")), "--strict") == 0 + + def test_a_schema_refused_id_is_not_called_schema_valid( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + """Contract rules also run on a contract the schema refused, so the warning + must not claim the schema accepted the id.""" + assert _validate(_with("contract.id", 123)) == 1 + out = capsys.readouterr().out + assert _squash("contract.id 123 cannot be used by the build phase") in _squash(out) + assert "schema allows" not in " ".join(out.split()) + + def test_a_crash_in_the_check_is_reported(self, monkeypatch: pytest.MonkeyPatch) -> None: + def boom(contract: Any) -> None: + raise RuntimeError("boom") + + monkeypatch.setattr(_ids, "build_id_warning", boom) + result = ValidationResult(is_valid=True) + args = SimpleNamespace(verbose=False, quiet=True, strict=False, offline=True) + _run_contract_rules(copy.deepcopy(_CONTRACT), None, result, args, LOG) + assert [w for w in result.warnings if w.startswith("Build-id check could not run")] + + +@pytest.mark.usefixtures("workspace") +class TestPlan: + @pytest.mark.parametrize("field,value", _UNUSABLE) + def test_plan_warns_and_writes_the_plan( + self, capsys: pytest.CaptureFixture[str], field: str, value: str + ) -> None: + assert _plan(_with(field, value)) == 0 + assert Path("plan.json").is_file() + out = capsys.readouterr().out + assert _names(out, field, value) + assert _squash(_REFUSED_BY) in _squash(out) + + @pytest.mark.parametrize("mode", ["amend", "amend-and-build", "replace-and-build"]) + def test_every_mode_only_warns(self, capsys: pytest.CaptureFixture[str], mode: str) -> None: + """Even the build modes whose apply will refuse it (founder's call).""" + assert _plan(_with("contract.id", "1orders"), "--mode", mode) == 0 + assert json.loads(Path("plan.json").read_text(encoding="utf-8"))["mode"] == mode + assert _names(capsys.readouterr().out, "contract.id", "1orders") + + def test_it_is_printed_once( + self, capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture + ) -> None: + """Not also as a log event, which the console prints again as a JSON line.""" + with caplog.at_level(logging.DEBUG): + assert _plan(_with("build.id", "1make_rows")) == 0 + captured = capsys.readouterr() + assert _squash(captured.out + captured.err).count(_squash(_REFUSED_BY)) == 1 + assert not [r for r in caplog.records if _REFUSED_BY in r.getMessage()] + + def test_a_marked_up_id_is_printed_as_written(self, capsys: pytest.CaptureFixture[str]) -> None: + """The id is contract data: Rich must not read ``[/]`` as a tag. The schema + refuses ``[`` today, so this calls the warning past the gate.""" + _warn_unusable_build_ids(_with("build.id", "[/]make_rows")) + assert _names(capsys.readouterr().out, "build.id", "[/]make_rows") + + def test_usable_ids_are_not_warned(self, capsys: pytest.CaptureFixture[str]) -> None: + assert _plan(copy.deepcopy(_CONTRACT)) == 0 + assert _squash(_REFUSED_BY) not in _squash(capsys.readouterr().out) + + def test_a_contract_without_builds_is_not_warned( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + assert _plan(_without_builds(_with("contract.id", "1orders"))) == 0 + assert _squash(_REFUSED_BY) not in _squash(capsys.readouterr().out) + + +def test_the_gate_apply_shares_with_plan_does_not_warn( + capsys: pytest.CaptureFixture[str], caplog: pytest.LogCaptureFixture +) -> None: + """``fluid apply`` refuses the id in a build mode (``CLIError``) and ignores it in + the others, so the warning stays out of the gate plan shares with apply.""" + with caplog.at_level(logging.DEBUG, logger=LOG.name): + _gate_contract_for_plan_or_apply(_with("contract.id", "1orders"), LOG, command="apply") + assert _squash(_REFUSED_BY) not in _squash(capsys.readouterr().out) + assert not [r for r in caplog.records if "build_identifier" in r.getMessage()] diff --git a/tests/conftest.py b/tests/conftest.py index 0eaf60cbb..0657541d9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -359,6 +359,7 @@ def _disable_ssrf_guard_for_unit_tests(request): _real_guard_files = ( "tests/util/test_safe_http.py", "tests/test_forge_web_tools.py", + "tests/build_runners/test_deployment_host_policy_real_pin_hook.py", ) if any(marker in fspath for marker in _real_guard_files): yield diff --git a/tests/streaming/test_compiler.py b/tests/streaming/test_compiler.py index 7a376354d..5a933f233 100644 --- a/tests/streaming/test_compiler.py +++ b/tests/streaming/test_compiler.py @@ -64,6 +64,15 @@ def test_sql_and_runtime_changes_change_digest(contract): "mutate,match", [ (lambda c, f: f["deployment"].update(image="flink:latest"), "invalid configuration"), + # A pattern ends at the end of the value: a trailing newline is refused. + ( + lambda c, f: f["deployment"].update(name=f["deployment"]["name"] + "\n"), + "invalid configuration", + ), + ( + lambda c, f: f["deployment"].update(image=f["deployment"]["image"] + "\n"), + "invalid configuration", + ), (lambda c, f: f["deployment"].pop("context"), "invalid configuration"), (lambda c, f: f.update(typo=True), "invalid configuration"), (lambda c, f: _stream(c).update(format="protobuf"), "format: expected one of"), diff --git a/tests/test_diff_changelog.py b/tests/test_diff_changelog.py index f4e03f6cc..53a754c3f 100644 --- a/tests/test_diff_changelog.py +++ b/tests/test_diff_changelog.py @@ -21,8 +21,11 @@ from __future__ import annotations import argparse +import copy import json import logging +import sys +import time from pathlib import Path import pytest @@ -30,10 +33,12 @@ from fluid_build.api.changelog import compare_contracts, render_markdown, render_text from fluid_build.api.changelog_rules import ( _classify_type_diff, + _norm_type, _parse_type, diff_agent_policy, diff_columns, diff_consumes, + diff_duplicates, diff_quality_severity, diff_sovereignty, ) @@ -42,6 +47,14 @@ FIXTURES = Path(__file__).parent / "fixtures" / "changelog" +# Characters the type tests need that look like ASCII, built with chr() so the source +# stays plain ASCII and greppable. +DOTLESS_I = chr(0x131) # str.upper() maps it to "I" +LONG_S = chr(0x17F) # str.upper() maps it to "S" +DOTTED_CAPITAL_I = chr(0x130) +FULLWIDTH_ONE = chr(0xFF11) +NBSP = chr(0xA0) + @pytest.fixture def v1_v2(): @@ -475,6 +488,14 @@ def test_cli_diff_baseline_rejects_state(tmp_path): ("BIGINT", "BIGINT", None), ("DECIMAL(38, 18)", "DECIMAL", (38, 18)), ("", "", None), + # Off-grammar text is its own type: it equals itself and nothing else, so a + # change in it can never be read as "no parameters" or as whitespace. + ("ENUM('A','B')", "ENUM('A','B')", None), + ("VARCHAR(MAX)", "VARCHAR(MAX)", None), + ("DECIMAL(1_0,2)", "DECIMAL(1_0,2)", None), + ("DECIMAL(+10,2)", "DECIMAL(+10,2)", None), + (f"DECIMAL({FULLWIDTH_ONE}0,2)", f"DECIMAL({FULLWIDTH_ONE}0,2)", None), + (f"DECIMAL({NBSP}10,2)", f"DECIMAL({NBSP}10,2)", None), ], ) def test_parse_type_extracts_base_and_params(raw, expected_base, expected_params): @@ -495,10 +516,14 @@ def test_decimal_precision_narrowing_is_breaking(): assert kind == "column_type_changed" -def test_decimal_mixed_param_change_is_info(): - """Precision grew but scale shrank — neither pure widening nor narrowing.""" +def test_decimal_mixed_param_change_is_breaking(): + """Precision grew but scale shrank — neither pure widening nor narrowing. + + It used to be ``info``, which ``--fail-on-breaking`` lets through; a column that + holds fewer fractional digits than before loses data for its consumers. + """ sev, kind = _classify_type_diff("DECIMAL(20,4)", "DECIMAL(30,2)") - assert sev == "info" + assert sev == "breaking" assert kind == "column_type_param_changed" @@ -894,24 +919,18 @@ def test_a_column_dropped_while_moving_to_contract_schema_is_still_found(): ] -@pytest.mark.parametrize("fail_on_breaking, expected_rc", [(True, 1), (False, 0)]) -def test_cli_diff_baseline_sees_a_required_column_added_to_a_076_contract( - tmp_path, fail_on_breaking, expected_rc -): - """``--fail-on-breaking`` is the exit code CI gates on; it must see 0.7.x columns.""" +def _cli_diff_baseline(tmp_path, old, new, fail_on_breaking): + """Run ``fluid diff --baseline `` on two contract dicts; return (rc, JSON report).""" import yaml from fluid_build.cli import diff as diff_mod - old = tmp_path / "old.fluid.yaml" - new = tmp_path / "new.fluid.yaml" - old.write_text(yaml.safe_dump(_contract_076([_ID_COLUMN]))) - added = {"name": "region", "type": "string", "required": True} - new.write_text(yaml.safe_dump(_contract_076([_ID_COLUMN, added]))) + (tmp_path / "old.fluid.yaml").write_text(yaml.safe_dump(old)) + (tmp_path / "new.fluid.yaml").write_text(yaml.safe_dump(new)) out = tmp_path / "diff.json" args = argparse.Namespace( - contract=str(new), - baseline=str(old), + contract=str(tmp_path / "new.fluid.yaml"), + baseline=str(tmp_path / "old.fluid.yaml"), env=None, state=None, out=str(out), @@ -920,11 +939,386 @@ def test_cli_diff_baseline_sees_a_required_column_added_to_a_076_contract( format="json", cmd="diff", ) - rc = diff_mod.run(args, logging.getLogger("test.diff")) + return rc, json.loads(out.read_text()) + + +@pytest.mark.parametrize("fail_on_breaking, expected_rc", [(True, 1), (False, 0)]) +def test_cli_diff_baseline_sees_a_required_column_added_to_a_076_contract( + tmp_path, fail_on_breaking, expected_rc +): + """``--fail-on-breaking`` is the exit code CI gates on; it must see 0.7.x columns.""" + added = {"name": "region", "type": "string", "required": True} + + rc, report = _cli_diff_baseline( + tmp_path, _contract_076([_ID_COLUMN]), _contract_076([_ID_COLUMN, added]), fail_on_breaking + ) + + assert rc == expected_rc + assert [c["kind"] for c in report["breaking"]] == ["column_added"] + + +# --------------------------------------------------------------------------- +# Type comparison: type changes that used to pass ``--fail-on-breaking`` +# --------------------------------------------------------------------------- + + +def _assert_schema_valid(*contracts): + """Every contract below is one the bundled 0.7.6 schema accepts, so the engine must handle it.""" + from fluid_build.schema_manager import FluidSchemaManager + + for contract in contracts: + result = FluidSchemaManager().validate_contract( + contract, schema_version="0.7.6", offline_only=True + ) + assert result.is_valid, result.errors + + +def _type_change_report(old_type, new_type): + old = _contract_076([{"name": "c", "type": old_type}]) + new = _contract_076([{"name": "c", "type": new_type}]) + _assert_schema_valid(old, new) + return compare_contracts(old, new) + + +@pytest.mark.parametrize( + "old_type, new_type, severity, kind", + [ + # A name that only looks like a real type once str.upper() has folded it. + pytest.param("int", f"{DOTLESS_I}nt", "breaking", "column_type_changed", id="dotless-i"), + pytest.param( + "timestamp", f"time{LONG_S}tamp", "breaking", "column_type_changed", id="long-s" + ), + # Parameters used to be ignored when the base type changed. + ("int", "decimal(5,0)", "breaking", "column_type_changed"), + # Parameters present on one side only, or moving in different directions. + ("varchar", "varchar(10)", "breaking", "column_type_param_changed"), + ("varchar(255)", "varchar", "breaking", "column_type_param_changed"), + ("decimal(10,2)", "decimal(12,1)", "breaking", "column_type_param_changed"), + ("decimal(10,2)", "decimal", "breaking", "column_type_param_changed"), + # Precision and scale both grow, the integer digits shrink (8 -> 7). + ("decimal(10,2)", "decimal(11,4)", "breaking", "column_type_param_changed"), + # Parameters that name something (a spatial reference id, an exact width) don't widen. + ("geometry(4326)", "geometry(32633)", "breaking", "column_type_param_changed"), + ("binary(16)", "binary(32)", "breaking", "column_type_param_changed"), + # Entries that left the widening table. + ("time", "timestamp", "breaking", "column_type_changed"), + ("time", "datetime", "breaking", "column_type_changed"), + ("int", "float", "breaking", "column_type_changed"), + ("integer", "float", "breaking", "column_type_changed"), + ("int32", "float", "breaking", "column_type_changed"), + ("float", "decimal", "breaking", "column_type_changed"), + ("float", "numeric", "breaking", "column_type_changed"), + # Parameters the classifier can't order used to read as "no parameters". + ("enum('a','b')", "enum('a')", "breaking", "column_type_changed"), + ("varchar", "varchar(max)", "breaking", "column_type_changed"), + ("decimal(10,2)", "decimal(1_0,2)", "breaking", "column_type_changed"), + ("decimal(10,2)", f"decimal({NBSP}10,2)", "breaking", "column_type_changed"), + # Controls, which pass on the code before this change too: what still widens, and + # what is only whitespace. + ("int", "bigint", "non_breaking", "column_type_widened"), + ("int", "double", "non_breaking", "column_type_widened"), + ("int", "decimal", "non_breaking", "column_type_widened"), + ("varchar", "text", "non_breaking", "column_type_widened"), + ("varchar(50)", "varchar(255)", "non_breaking", "column_type_widened"), + ("date", "timestamp", "non_breaking", "column_type_widened"), + ("float", "double", "non_breaking", "column_type_widened"), + ("decimal(10,2)", "decimal(12,4)", "non_breaking", "column_type_widened"), + ("char(10)", "char(20)", "non_breaking", "column_type_widened"), + ("float(24)", "float(53)", "non_breaking", "column_type_widened"), + ("decimal(10,2)", "decimal(10, 2)", "info", "column_type_normalized"), + ("varchar(255)", "varchar(0255)", "info", "column_type_normalized"), + ], +) +def test_076_type_comparison(old_type, new_type, severity, kind): + report = _type_change_report(old_type, new_type) + + assert report.total == 1 + (change,) = getattr(report, severity) + assert change.kind == kind + assert change.path == "exposes[0].contract.schema.c.type" + + +@pytest.mark.parametrize( + "raw, expected", + [ + (" int ", "INT"), + ("DeCiMaL(10,2)", "DECIMAL(10,2)"), + (f"{DOTLESS_I}nt", f"{DOTLESS_I}nt"), + (f"time{LONG_S}tamp", f"time{LONG_S}tamp"), + (f"{DOTTED_CAPITAL_I}nt", f"{DOTTED_CAPITAL_I}nt"), + (None, ""), + (5, ""), + ], +) +def test_norm_type_folds_case_for_ascii_names_only(raw, expected): + assert _norm_type(raw) == expected + + +@pytest.mark.parametrize( + "old, new, expected", + [ + # Same integer digits, bigger scale: still holds every old value. + ("DECIMAL(10,2)", "DECIMAL(12,4)", ("non_breaking", "column_type_widened")), + ("DECIMAL(10,2)", "DECIMAL(11,4)", ("breaking", "column_type_param_changed")), + ("NUMBER(10,2)", "NUMBER(11,4)", ("breaking", "column_type_param_changed")), + ("NUMERIC(10,2)", "NUMERIC(12,1)", ("breaking", "column_type_param_changed")), + ("DECIMAL(10)", "DECIMAL(10,2)", ("breaking", "column_type_param_changed")), + ("DECIMAL(10)", "DECIMAL(12)", ("non_breaking", "column_type_widened")), + # Types that are not fixed-point compare their parameters one by one. + ("TIMESTAMP(3)", "TIMESTAMP(6)", ("non_breaking", "column_type_widened")), + ], +) +def test_fixed_point_widens_only_when_scale_and_integer_digits_both_hold(old, new, expected): + assert _classify_type_diff(old, new) == expected + + +@pytest.mark.parametrize( + "old, new", + [ + ("GEOMETRY(4326)", "GEOMETRY(32633)"), + ("GEOMETRY(32633)", "GEOMETRY(4326)"), + ("VECTOR(3)", "VECTOR(1536)"), + ("BINARY(16)", "BINARY(32)"), + ("BIT(8)", "BIT(16)"), + ], +) +def test_parameters_that_name_something_are_never_a_widening(old, new): + """A spatial reference id, a dimension or an exact width does not measure capacity.""" + assert _classify_type_diff(old, new)[0] == "breaking" + + +@pytest.mark.parametrize( + "old, new", + [ + ("VARCHAR(10)", "VARCHAR(20)"), + ("CHAR(1)", "CHAR(8)"), + ("FLOAT(24)", "FLOAT(53)"), + ("TIME(0)", "TIME(6)"), + ("DATETIME2(3)", "DATETIME2(7)"), + ], +) +def test_a_bigger_length_or_precision_is_a_widening(old, new): + assert _classify_type_diff(old, new) == ("non_breaking", "column_type_widened") + + +def test_parsing_a_long_run_of_spaces_or_an_unclosed_list_stays_linear(): + """A type is untrusted text: the parser must not backtrack quadratically on it.""" + started = time.perf_counter() + _parse_type("A" + " " * 50_000 + "x") + _parse_type("A(" + "1" * 50_000) + assert time.perf_counter() - started < 2.0 + + +def test_an_oversized_integer_parameter_is_unparseable_text(): + if not hasattr(sys, "set_int_max_str_digits"): + pytest.skip("this interpreter has no integer-string length limit") + previous = sys.get_int_max_str_digits() + sys.set_int_max_str_digits(640) + try: + raw = "VARCHAR(" + "9" * 700 + ")" + assert _parse_type(raw) == (raw, None) + finally: + sys.set_int_max_str_digits(previous) + + +def test_a_non_ascii_type_name_says_why_it_is_not_the_same_type(): + report = _type_change_report("int", f"{DOTLESS_I}nt") + + (change,) = report.breaking + assert "non-ASCII" in change.description + + +@pytest.mark.parametrize("fail_on_breaking, expected_rc", [(True, 1), (False, 0)]) +def test_cli_diff_baseline_fails_on_a_type_change_that_used_to_pass( + tmp_path, fail_on_breaking, expected_rc +): + """``int`` -> ``decimal(5,0)`` was reported as a widening, so the CI gate stayed green.""" + old = _contract_076([{"name": "id", "type": "int"}]) + new = _contract_076([{"name": "id", "type": "decimal(5,0)"}]) + _assert_schema_valid(old, new) + + rc, report = _cli_diff_baseline(tmp_path, old, new, fail_on_breaking) assert rc == expected_rc - assert [c["kind"] for c in json.loads(out.read_text())["breaking"]] == ["column_added"] + assert [c["kind"] for c in report["breaking"]] == ["column_type_changed"] + + +def _contract_with_exposes(*exposes): + """A 0.7.6 contract with one expose per ``(expose_id, columns)`` pair.""" + contract = _contract_076([]) + template = contract["exposes"][0] + contract["exposes"] = [] + for expose_id, columns in exposes: + expose = copy.deepcopy(template) + expose["exposeId"] = expose_id + expose["contract"]["schema"] = columns + contract["exposes"].append(expose) + return contract + + +@pytest.mark.parametrize( + "declared, kinds", + [ + pytest.param(["int", "string"], {"column_duplicate_name"}, id="last-matches-baseline"), + pytest.param( + ["string", "int"], {"column_duplicate_name", "column_type_changed"}, id="last-differs" + ), + pytest.param(["string", "string"], {"column_duplicate_name"}, id="identical"), + pytest.param(["string"] * 3, {"column_duplicate_name"}, id="three-times"), + ], +) +def test_076_a_column_declared_twice_in_the_new_contract_is_breaking(declared, kinds): + """The index keeps the last declaration of a name, so the diff never saw the other one.""" + old = _contract_076([_ID_COLUMN, {"name": "note", "type": "string"}]) + new = _contract_076([_ID_COLUMN] + [{"name": "note", "type": t} for t in declared]) + _assert_schema_valid(old, new) + + report = compare_contracts(old, new) + + assert {c.kind for c in report.breaking} == kinds + assert report.total == len(kinds) + (duplicate,) = [c for c in report.breaking if c.kind == "column_duplicate_name"] + assert duplicate.path == "exposes[0].contract.schema.note" + assert [c["type"] for c in duplicate.after] == declared + + +_ZIP = {"name": "zip", "type": "STRING"} + + +def _address(*fields): + return {"name": "addr", "type": "STRUCT", "fields": list(fields)} + + +@pytest.mark.parametrize( + "old_columns, new_columns, non_breaking", + [ + pytest.param( + [_address(_ZIP)], + [_address({"name": "zip", "type": "INT"}, _ZIP)], + [], + id="column-in-both-contracts", + ), + pytest.param( + [{"name": "id", "type": "STRING"}], + [{"name": "id", "type": "STRING"}, _address(_ZIP, _ZIP)], + ["column_added"], + id="column-is-new", + ), + ], +) +def test_a_nested_field_declared_twice_in_the_new_contract_is_breaking( + old_columns, new_columns, non_breaking +): + report = compare_contracts( + _contract_flat(old_columns), _contract_flat(new_columns), validate=False + ) + + assert [c.kind for c in report.non_breaking] == non_breaking + assert [(c.kind, c.path) for c in report.breaking] == [ + ("nested_field_duplicate_name", "exposes[0].schema.addr.fields.zip") + ] + + +def test_076_an_expose_id_declared_twice_in_the_new_contract_is_breaking(): + as_string = [{"name": "a", "type": "string"}] + as_int = [{"name": "a", "type": "int"}] + old = _contract_with_exposes(("orders", as_string)) + new = _contract_with_exposes(("orders", as_int), ("orders", as_string)) + _assert_schema_valid(old, new) + + report = compare_contracts(old, new) + + assert report.total == 1 + (change,) = report.breaking + assert (change.kind, change.path) == ("expose_duplicate_id", "exposes[1]") + assert change.after == {"id": "orders", "indexes": [0, 1]} + + +def test_076_distinct_names_and_ids_are_not_duplicates(): + columns = [_ID_COLUMN, {"name": "note", "type": "string"}] + contract = _contract_with_exposes(("orders", columns), ("customers", columns)) + _assert_schema_valid(contract) + + assert compare_contracts(contract, copy.deepcopy(contract)).total == 0 + + +def test_a_duplicate_in_an_expose_that_is_new_to_the_contract_is_breaking(): + """A new expose has no baseline to hide a change from, but it is just as ambiguous.""" + customers = ("customers", [_ID_COLUMN]) + note = {"name": "note", "type": "string"} + old = _contract_with_exposes(customers) + new = _contract_with_exposes(customers, ("orders", [_ID_COLUMN, note, note])) + _assert_schema_valid(old, new) + + report = compare_contracts(old, new) + + assert [c.kind for c in report.non_breaking] == ["expose_added"] + assert [(c.kind, c.path) for c in report.breaking] == [ + ("column_duplicate_name", "exposes[1].contract.schema.note") + ] + + +def test_a_nested_field_declared_twice_below_the_first_level_is_breaking(): + lat = {"name": "lat", "type": "DOUBLE"} + geo = {"name": "geo", "type": "STRUCT", "fields": [lat, lat]} + contract = _contract_flat([{"name": "addr", "type": "STRUCT", "fields": [geo]}]) + + assert [(c.kind, c.path) for c in diff_duplicates(contract)] == [ + ("nested_field_duplicate_name", "exposes[0].schema.addr.fields.geo.fields.lat") + ] + + +def test_076_a_duplicate_both_contracts_declare_is_still_reported(): + """The documented exception to idempotence: the earlier declaration is hidden in both.""" + note = {"name": "note", "type": "string"} + contract = _contract_076([_ID_COLUMN, note, note]) + _assert_schema_valid(contract) + + report = compare_contracts(contract, copy.deepcopy(contract)) + + assert [c.kind for c in report.breaking] == ["column_duplicate_name"] + assert report.total == 1 + + +def test_076_a_duplicate_only_the_baseline_declares_is_not_reported(): + note = {"name": "note", "type": "string"} + old = _contract_076([_ID_COLUMN, note, note]) + new = _contract_076([_ID_COLUMN, note]) + + assert compare_contracts(old, new).total == 0 + + +def test_an_expose_is_named_by_id_or_exposeid_and_an_unnamed_one_by_position(): + exposes = [{"id": "a"}, {"exposeId": "a"}, {"id": "b"}, {"schema": []}, {"id": "_pos_3"}] + + found = diff_duplicates({"exposes": exposes}) + + assert [(c.kind, c.path) for c in found] == [("expose_duplicate_id", "exposes[1]")] + + +@pytest.mark.parametrize("schema", [5, True, 3.5, "text", {"x": 1}]) +def test_a_malformed_schema_in_an_expose_that_is_new_does_not_crash(schema): + """The duplicate walk reads exposes the comparison used to skip as merely added. + + A schema-invalid contract only gets here with ``validate=False``. + """ + old = {"exposes": [{"id": "a", "schema": [{"name": "x", "type": "INT"}]}]} + new = {"exposes": [old["exposes"][0], {"id": "b", "schema": schema}]} + + report = compare_contracts(old, new, validate=False) + + assert [c.kind for c in report.non_breaking] == ["expose_added"] + assert report.total == 1 + + +def test_the_duplicate_findings_render(): + note = {"name": "note", "type": "string"} + report = compare_contracts(_contract_076([_ID_COLUMN]), _contract_076([_ID_COLUMN, note, note])) + + assert "column_duplicate_name" in render_text(report) + assert "column_duplicate_name" in render_markdown(report) + assert json.loads(json.dumps(report.to_dict()))["breaking"][0]["after"] == [note, note] # --------------------------------------------------------------------------- diff --git a/tests/test_schema_pattern_end_anchor.py b/tests/test_schema_pattern_end_anchor.py index 6b12793da..b51d057bc 100644 --- a/tests/test_schema_pattern_end_anchor.py +++ b/tests/test_schema_pattern_end_anchor.py @@ -239,12 +239,33 @@ def _nodes(node: Any) -> Iterator[Dict[str, Any]]: yield from _nodes(value) +#: Every schema fluid validates against: the FLUID and Ossie ones, and the vendored +#: standards (tests/test_standard_schema_pattern_end_anchor.py tests their checks). +_SHIPPED = sorted(_SCHEMA_DIR.glob("*.json")) + [ + _SCHEMA_DIR.parent / "providers" / "odcs" / "odcs-schema-v3.1.0.json", + _SCHEMA_DIR.parent / "providers" / "odps_standard" / "schemas" / "odps-product-v1.0.0.json", + _SCHEMA_DIR.parent / "providers" / "odps_standard" / "schemas" / "odps-product-v1.1.0.json", + _SCHEMA_DIR.parent / "providers" / "opds" / "opds-schema-v4.1.0.json", +] + +#: ``patternProperties`` keys, by schema. They keep jsonschema's own ``re.search``: +#: jsonschema matches them in ``patternProperties``, ``additionalProperties`` and +#: ``unevaluatedProperties``, and overriding ``patternProperties`` alone, as +#: check-jsonschema does, lets a key that one matches and another does not through +#: ``additionalProperties: false`` unchecked. So OPDS's ISO 639-1 language codes, +#: ``^[a-z]{2}$``, also match ``"en\n"``. Under ``product.details`` and ``declarative``, +#: which take any key, the value is then checked as a language entry. +#: ``$defs/Languages`` takes no other key, so a hand-written OPDS file can carry +#: ``{"en\n": ...}`` there; fluid's emitter writes ``en``. +_KEY_PATTERNS = {"opds-schema-v4.1.0.json": [("^[a-z]{2}$",)] * 3} + + def test_every_bundled_schema_pattern_is_rewritten_anchor_for_anchor() -> None: """The scan reads escapes and sets as Python does, checked against Python's own - parser for every ``pattern`` the bundled schemas hold.""" + parser for every ``pattern`` the shipped schemas hold.""" patterns = { node["pattern"] - for path in _SCHEMA_DIR.glob("fluid-*schema-*.json") + for path in _SHIPPED for node in _nodes(json.loads(path.read_text(encoding="utf-8"))) if isinstance(node.get("pattern"), str) } @@ -252,16 +273,17 @@ def test_every_bundled_schema_pattern_is_rewritten_anchor_for_anchor() -> None: assert [p for p in patterns if not _only_end_anchors_changed(p)] == [] -@pytest.mark.parametrize( - "path", sorted(_SCHEMA_DIR.glob("fluid-schema-*.json")), ids=lambda path: path.name -) +@pytest.mark.parametrize("path", _SHIPPED, ids=lambda path: path.name) def test_every_pattern_of_the_schema_is_checked_with_the_rewrite(path: Path) -> None: """jsonschema picks the validator class again for each subschema it enters, by - that subschema's ``$schema``. One that declares it, or a ``$ref`` to the root, - which does, would be checked with jsonschema's own ``pattern`` again. - ``patternProperties`` keys are not rewritten (``schema_validator`` says why), so - a schema that starts to use the keyword has to come back here.""" - below_root = list(_nodes(json.loads(path.read_text(encoding="utf-8"))))[1:] - assert [n for n in below_root if isinstance(n.get("$schema"), str)] == [] - assert [n for n in below_root if n.get("$ref") == "#"] == [] - assert [n for n in below_root if "patternProperties" in n] == [] + that subschema's ``$schema``. One that declares it, or a ``$ref``, + ``$recursiveRef`` or ``$dynamicRef`` to the root, which does, would be checked + with jsonschema's own ``pattern`` again. ``patternProperties`` keys are not + rewritten, so a schema that starts to use the keyword has to come back to + :data:`_KEY_PATTERNS`.""" + nodes = list(_nodes(json.loads(path.read_text(encoding="utf-8")))) + assert [n for n in nodes[1:] if isinstance(n.get("$schema"), str)] == [] + assert [n for n in nodes if n.get("$ref") == "#"] == [] + assert [n for n in nodes if "$recursiveRef" in n or "$dynamicRef" in n] == [] + keys = [tuple(n["patternProperties"]) for n in nodes if "patternProperties" in n] + assert keys == _KEY_PATTERNS.get(path.name, []) diff --git a/tests/test_standard_schema_pattern_end_anchor.py b/tests/test_standard_schema_pattern_end_anchor.py new file mode 100644 index 000000000..23eb38ec3 --- /dev/null +++ b/tests/test_standard_schema_pattern_end_anchor.py @@ -0,0 +1,315 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""The other schema checks read a ``pattern``'s ``$`` as ECMA-262 does, too. + +``tests/test_schema_pattern_end_anchor.py`` covers the FLUID schemas. fluid also +validates against the vendored ODCS, Bitol ODPS and OPDS schemas, the Ossie +schema, and the schema an installed plugin gives its ``extensions.`` block. +Each of those checks ran jsonschema's own ``pattern``, Python's ``re.search``, +whose ``$`` also matches before a final ``\\n``. So ODCS refused +``schema[].id: "t 1"`` but took ``"t1\\n"``, which an ``id: |`` block produces. + +Of the vendored schemas only ODCS has an end-anchored ``pattern`` today, so the +other checks are tested with a schema of the test's own. The tripwires in +``tests/test_schema_pattern_end_anchor.py`` cover the vendored schemas too: what +they would have to start using for a check to leave the rewrite. +""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path +from typing import Any, Dict, Iterator + +import jsonschema +import pytest +import yaml + +import fluid_build +import fluid_build.extension_schemas as extension_schemas +from fluid_build.forge.core.artifact_validators import validate_odcs, validate_opds +from fluid_build.forge_datamodel.emit import osi_sidecar +from fluid_build.providers.base import ProviderError +from fluid_build.providers.odcs.validation import collect_errors, load_schema +from fluid_build.providers.odps_standard import validation as odps_validation +from fluid_build.providers.opds import validator as opds_validator +from tests.test_jsonschema_dialect import _ODCS_BASE + +pytestmark = [pytest.mark.unit] + +_LINE_BREAK_HINT = " (it ends in a line break, which a YAML | or > block keeps and |- or >- drops)" +#: A schema of the test's own, for the checks whose vendored schema has no pattern. +_NAME_SCHEMA: Dict[str, Any] = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": {"name": {"type": "string", "pattern": "^[a-z]+$"}}, +} + + +def _odcs(table_id: str) -> Dict[str, Any]: + return {**_ODCS_BASE, "schema": [{"id": table_id, "name": "orders", "logicalType": "object"}]} + + +def _odcs_yaml(block: str, table_id: str = "t1") -> str: + """An ODCS document whose ``schema[0].id`` is a YAML block scalar: ``|`` or ``|-``.""" + body = yaml.safe_dump(_ODCS_BASE) + return f"{body}schema:\n- id: {block}\n {table_id}\n name: orders\n logicalType: object\n" + + +# ── ODCS: fluid odcs validate, fluid validate-artifacts ───────────────────── + + +def test_odcs_refuses_a_table_id_that_ends_in_a_newline() -> None: + """``collect_errors`` is ``fluid odcs validate``'s jsonschema pass.""" + schema = load_schema() + assert schema is not None + assert collect_errors(_odcs("t1"), schema) == [] + + errors = [e for e in collect_errors(_odcs("t1\n"), schema) if e["validator"] == "pattern"] + + assert [e["path"] for e in errors] == ["schema.0.id"], errors + assert errors[0]["message"].endswith(_LINE_BREAK_HINT), errors + + +def test_fluid_odcs_validate_refuses_a_block_scalar_table_id(tmp_path: Path, capsys: Any) -> None: + from fluid_build.cli import main + + path = tmp_path / "orders.odcs.yaml" + path.write_text(_odcs_yaml("|"), encoding="utf-8") + assert yaml.safe_load(path.read_text(encoding="utf-8"))["schema"][0]["id"] == "t1\n" + + rc = main(["odcs", "validate", str(path), "--no-vowl"]) + + text = capsys.readouterr().out + assert rc == 1, text + assert "schema.0.id: 't1\\n' does not match" in text, text + + +def test_fluid_odcs_validate_keeps_the_hint_whole(tmp_path: Path, capsys: Any) -> None: + """The command cuts each message to 120 characters. Cut inside the hint, a long id's + message ended ``... block keeps and |``, which reads as advice to use ``|``.""" + from fluid_build.cli import main + + path = tmp_path / "orders.odcs.yaml" + path.write_text(_odcs_yaml("|", "customer_orders_v1"), encoding="utf-8") + + rc = main(["odcs", "validate", str(path), "--no-vowl"]) + + text = " ".join(capsys.readouterr().out.split()) # undo the console's line wrapping + assert rc == 1, text + assert f"'customer_orders_v1\\n' does not match '^[A-Za-z0-9_-]+$'{_LINE_BREAK_HINT}" in text + + +def test_fluid_odcs_validate_prints_a_value_with_brackets_literally( + tmp_path: Path, capsys: Any +) -> None: + """The message quotes the value, and Rich read ``[/]`` in it as a closing tag: the + command ended in ``Unexpected error: closing tag '[/]' ... has nothing to close``.""" + from fluid_build.cli import main + + path = tmp_path / "orders.odcs.yaml" + path.write_text(yaml.safe_dump(_odcs("t[/]1")), encoding="utf-8") + + rc = main(["odcs", "validate", str(path), "--no-vowl"]) + + text = " ".join(capsys.readouterr().out.split()) + assert rc == 1, text + assert "schema.0.id: 't[/]1' does not match" in text, text + + +def test_validate_artifacts_refuses_a_block_scalar_table_id() -> None: + """``validate_odcs`` is what ``fluid validate-artifacts`` runs on ``odcs/*.odcs.yaml``.""" + assert validate_odcs("odcs/orders.odcs.yaml", _odcs_yaml("|-").encode()) == [] + + issues = validate_odcs("odcs/orders.odcs.yaml", _odcs_yaml("|").encode()) + + messages = [i.message for i in issues] + assert any(m.startswith("at schema/0/id: 't1\\n' does not match ") for m in messages), messages + + +# ── Bitol ODPS, OPDS, Ossie: no pattern in the vendored schema yet ────────── + + +def test_odps_validate_reads_a_pattern_as_ecma_does() -> None: + odps_validation.validate({"name": "orders"}, _NAME_SCHEMA) + + with pytest.raises(ProviderError, match=re.escape("'orders\\n' does not match '^[a-z]+$'")): + odps_validation.validate({"name": "orders\n"}, _NAME_SCHEMA) + + +def test_odps_validate_still_checks_the_schema_first() -> None: + """It still runs ``jsonschema.validate``, now with the rewritten class.""" + with pytest.raises(jsonschema.SchemaError): + odps_validation.validate({}, {"type": 5}) + + +def test_opds_validation_reads_a_pattern_as_ecma_does(monkeypatch: Any) -> None: + monkeypatch.setitem(opds_validator._SCHEMA_CACHE, "0.0-test", _NAME_SCHEMA) + validate = opds_validator.validate_against_opds_schema + assert validate({"name": "orders"}, "0.0-test") == (True, None, "full_schema") + + valid, errors, kind = validate({"name": "orders\n"}, "0.0-test") + + assert (valid, kind) == (False, "full_schema") + assert errors == [f"name: 'orders\\n' does not match '^[a-z]+$'{_LINE_BREAK_HINT}"] + + +def test_an_opds_schema_that_declares_no_dialect_is_2020_12(monkeypatch: Any) -> None: + """What the check pinned before. Draft 7 has no ``dependentRequired`` and would + pass this document.""" + schema = {"type": "object", "dependentRequired": {"a": ["b"]}} + monkeypatch.setitem(opds_validator._SCHEMA_CACHE, "0.0-test", schema) + + valid, errors, _ = opds_validator.validate_against_opds_schema({"a": 1}, "0.0-test") + + assert (valid, errors) == (False, ["root: 'b' is a dependency of 'a'"]) + + +def test_ossie_document_validation_reads_a_pattern_as_ecma_does(monkeypatch: Any) -> None: + monkeypatch.setattr(osi_sidecar, "_load_ossie_schema", lambda: _NAME_SCHEMA) + assert osi_sidecar.validate_osi_document({"name": "orders"}) == [] + + issues = osi_sidecar.validate_osi_document({"name": "orders\n"}) + + assert issues == [f"name: 'orders\\n' does not match '^[a-z]+$'{_LINE_BREAK_HINT}"] + + +def test_without_jsonschema_each_check_says_it_is_missing(monkeypatch: Any) -> None: + """The checks read the flag ``schema_validator`` needs. A jsonschema that is found + but fails to import is reported as missing, not as ``NameError: jsonschema``.""" + monkeypatch.setattr("fluid_build.schema_manager.JSONSCHEMA_AVAILABLE", False) + + assert [i.code for i in validate_odcs("odcs/o.odcs.yaml", b"id: x")] == ["ODCS-LIB-MISSING"] + assert [i.code for i in validate_opds("opds/o.opds.json", b"{}")] == ["OPDS-LIB-MISSING"] + assert collect_errors(_odcs("t1\n"), {}) == [] + with pytest.raises(ImportError, match="jsonschema is not available"): + osi_sidecar.validate_osi_document({}) + + +# ── plugin extension blocks ───────────────────────────────────────────────── + +#: The shape of data-product-forge-custom-scaffold's schema: draft-07, "$"-anchored. +_SCAFFOLD_SCHEMA: Dict[str, Any] = { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "patterns": { + "type": "array", + "items": { + "type": "object", + "properties": { + "use": { + "type": "string", + "pattern": "^[a-zA-Z][a-zA-Z0-9_-]*:[a-zA-Z][a-zA-Z0-9_-]*$", + } + }, + }, + } + }, +} + + +def _assemble(monkeypatch: Any, schema: Dict[str, Any], block: Any) -> Dict[str, Any]: + monkeypatch.setattr(extension_schemas, "iter_extension_schemas", lambda *a, **k: {"x": schema}) + return extension_schemas.assemble_proposed_extensions({"x": block}) + + +def test_an_extension_block_whose_value_ends_in_a_newline_is_dropped(monkeypatch: Any) -> None: + kept = {"patterns": [{"use": "ci:basic"}]} + assert _assemble(monkeypatch, _SCAFFOLD_SCHEMA, kept) == {"x": kept} + assert _assemble(monkeypatch, _SCAFFOLD_SCHEMA, {"patterns": [{"use": "ci:basic\n"}]}) == {} + + +def test_an_extension_schema_is_read_in_the_dialect_it_declares(monkeypatch: Any) -> None: + """Draft 7 ignores ``unevaluatedProperties``, so the extra key used to be kept.""" + schema = { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": {"libraries": {"type": "array"}}, + "unevaluatedProperties": False, + } + assert _assemble(monkeypatch, schema, {"libraries": []}) == {"x": {"libraries": []}} + assert _assemble(monkeypatch, schema, {"libraries": [], "librarys": []}) == {} + + +def test_an_extension_schema_that_cannot_be_applied_drops_only_its_block( + monkeypatch: Any, caplog: Any +) -> None: + """A malformed plugin schema neither stops the run nor lets its block through.""" + schemas = {"bad": {"$schema": 7, "type": "string"}, "x": _SCAFFOLD_SCHEMA} + monkeypatch.setattr(extension_schemas, "iter_extension_schemas", lambda *a, **k: schemas) + kept = {"patterns": [{"use": "ci:basic"}]} + + assert extension_schemas.assemble_proposed_extensions({"bad": "a", "x": kept}) == {"x": kept} + assert "extension schema 'bad' unusable" in caplog.text + + +def test_an_extension_schema_that_declares_no_dialect_is_draft_07(monkeypatch: Any) -> None: + """The plugin contract says draft-07. 2020-12, jsonschema's default, has no + ``dependencies`` keyword and would keep this block.""" + schema = {"type": "object", "dependencies": {"output": ["patterns"]}} + kept = {"output": "x", "patterns": []} + assert _assemble(monkeypatch, schema, kept) == {"x": kept} + assert _assemble(monkeypatch, schema, {"output": "x"}) == {} + + +# ── one way to build a validator ──────────────────────────────────────────── + + +def _validator_builds(path: Path) -> Iterator[str]: + """Spellings in ``path`` that build a jsonschema validator without ``schema_validator``. + It goes by names, so a class first bound to another name (``V = Draft7Validator``) + gets past it.""" + text = path.read_text(encoding="utf-8") + if not any(word in text for word in ("jsonschema", "Validator", "validator_for")): + return + tree = ast.parse(text) + modules = {"jsonschema"} # names that hold jsonschema or jsonschema.validators + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules |= {a.asname for a in node.names if a.name == "jsonschema" and a.asname} + elif isinstance(node, ast.ImportFrom) and (node.module or "").startswith("jsonschema"): + modules |= {a.asname or a.name for a in node.names if a.name == "validators"} + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and (node.module or "").startswith("jsonschema"): + for a in node.names: + builder = a.name == "validator_for" or re.fullmatch(r"Draft\d+Validator", a.name) + if a.name == "validate" or (a.asname and builder): + yield f"{path.name}:{node.lineno}: from {node.module} import {a.name}" + if not isinstance(node, ast.Call): + continue + func = node.func + name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") + if name == "validator_for" or re.fullmatch(r"Draft\d+Validator", name): + yield f"{path.name}:{node.lineno}: {name}()" + elif name in ("validate", "extend", "create") and isinstance(func, ast.Attribute): + root = func.value + while isinstance(root, ast.Attribute): + root = root.value + cls = next((k.value for k in node.keywords if k.arg == "cls"), None) + routed = cls is not None and "schema_validator" in ast.unparse(cls) + if isinstance(root, ast.Name) and root.id in modules and not routed: + yield f"{path.name}:{node.lineno}: {ast.unparse(func)}()" + + +def test_every_validator_is_built_by_schema_validator() -> None: + """A check that builds its own validator reads ``$`` as Python does again; #799 and + this change each found such checks one at a time. Give ``schema_validator`` the + dialect class as ``default=`` instead of building it.""" + package = Path(fluid_build.__file__).parent + paths = [p for p in sorted(package.rglob("*.py")) if p != package / "schema_manager.py"] + assert len(paths) > 500, len(paths) # the glob found the package + assert [build for path in paths for build in _validator_builds(path)] == []