diff --git a/.vale/styles/spelling-exceptions.txt b/.vale/styles/spelling-exceptions.txt index 54d22545..e81bb335 100644 --- a/.vale/styles/spelling-exceptions.txt +++ b/.vale/styles/spelling-exceptions.txt @@ -120,6 +120,7 @@ REST RIRs rowcount Rowcount +rumdl schema_mapping sdk Slurp'it diff --git a/dev/knowledge/orchestration-prefect.md b/dev/knowledge/orchestration-prefect.md index 29cb0976..f0ca4516 100644 --- a/dev/knowledge/orchestration-prefect.md +++ b/dev/knowledge/orchestration-prefect.md @@ -6,10 +6,16 @@ `infrahub_sync/orchestration/` is the direct Prefect integration: a flow that runs one plan or one confirmed sync, and a serve entrypoint that exposes it as a locally served -deployment. It is the only package in the repository that imports `prefect`, it is installed -by the optional `prefect` extra, and nothing in the base package imports it — see +deployment. It is installed by the optional `prefect` extra, and nothing in the base +package imports it — see [ADR 9](../adr/0009-optional-integrations-live-in-their-own-package.md). +Two packages under `infrahub_sync/` import `prefect`, and both are optional: this one, and +`infrahub_sync/service/` below, which the `service` extra installs. No other package under +`infrahub_sync/` imports it, so a base install loads neither. The vendored +`opsmill_prefect_extras/` imports Prefect too; it sits outside `infrahub_sync/` and the +`service` extra is what brings it into a working install. + The flow calls [the shared execution surface](execution-surface.md) in-process. It never spawns the CLI. @@ -36,11 +42,29 @@ Exactly those four parameters. None of them accepts a path, a CLI fragment, a cr an environment override. Everything else the run needs comes from the serving process's own environment. -The separate `infrahub-sync-service/run` deployment accepts exactly seven parameters: -`run_id`, `sync_name`, `stage`, `configuration_reference`, `branch`, `expected_checksum`, -and `confirm_writes`. It does not replace or extend the four-parameter direct Prefect -flow. Credentials, endpoints, adapter instances, product-cache locations, and saved-plan -cache locations stay in the service worker environment. +The separate `infrahub-sync-service/run` deployment serves a different flow, with eight +parameters: + +```python +@flow(name="infrahub-sync-service") +def service_sync_run( + run_id: str, + stage: Literal["plan", "verify", "apply", "sync"], + config_id: str | None = None, + registry_version: int | None = None, + package_checksum: str | None = None, + branch: str | None = None, + expected_checksum: str | None = None, + confirm_writes: bool = False, +) -> dict[str, Any]: ... +``` + +The three registry parameters travel together: when all three are set, they name the +registered version the run is bound to; when all three are unset, the run uses the legacy +path; and a partial carrier is refused. +This flow does not replace or extend the four-parameter direct Prefect flow. Credentials, +endpoints, adapter instances, product-cache locations, and saved-plan cache locations stay +in the service worker environment. When the service flow runs outside Prefect context in offline executor tests, `get_run_logger()` raises `MissingContextError`. The flow catches only that exception and @@ -62,7 +86,7 @@ SUMMARY_LINE_FORMAT = "run %s finished: status=%s changed=%s summary=create:%d,u ``` This line is how a remote caller reads a run's outcome, and its format is contractual — -never a Python dict repr. It carries five `RunResult` fields: `run_id` (the leading +never a Python dictionary `repr`. It carries five `RunResult` fields: `run_id` (the leading substitution), `status`, `changed`, the three summary counts, and `artifact_path`. `sync_name` and `operation` deliberately do not appear. Changing the format is a breaking change for consumers. @@ -127,7 +151,7 @@ The directory path is fixed at serve start; its *contents* are re-resolved on ev configurations added, edited or removed take effect on the next run without re-serving. The serve process must be started from the repository root for the shipped example to work: -its `config.yml` uses repo-root-relative paths resolved against the serving process's +its `config.yml` uses repository-root-relative paths resolved against the serving process's working directory, and the cache root defaults to `Path.cwd()/.infrahub-sync-cache`. Started elsewhere, the example degrades to a silently empty plan or an adapter import failure. diff --git a/docs/docs/contributing.mdx b/docs/docs/contributing.mdx index 2ddf4500..255a1730 100644 --- a/docs/docs/contributing.mdx +++ b/docs/docs/contributing.mdx @@ -6,7 +6,8 @@ This guide covers how to set up a development environment for `infrahub-sync` an ## Prerequisites -- Python 3.10–3.13 (3.12 recommended) +- Python 3.11–3.13 for the full development profile (3.12 recommended). Python 3.10 runs + everything except the Sync service. - [uv](https://docs.astral.sh/uv/) for dependency management - Git @@ -32,10 +33,19 @@ Or see the [uv installation guide](https://docs.astral.sh/uv/getting-started/ins ### Install dependencies ```bash -uv sync --group dev +uv sync --extra dev --extra prefect --extra service ``` -This installs all runtime and development dependencies defined in `pyproject.toml`. +The `prefect` and `service` extras are not optional for development. Without them the type +checker cannot resolve the imports in `infrahub_sync/orchestration/` and +`infrahub_sync/service/`, and the tests that cover them skip themselves. + +On Python 3.10 the Sync service is unavailable, so install the direct Prefect profile +instead and exclude the service from type checking: + +```bash +uv sync --python 3.10 --extra dev --extra prefect +``` ### Verify your setup @@ -49,11 +59,16 @@ uv run infrahub-sync configs --help Before committing any changes, run the following commands in order: ```bash -uv run invoke format # Format code with ruff -uv run invoke lint # Lint code with ruff and pylint -uv run mypy infrahub_sync/ --ignore-missing-imports +# Run `rumdl fmt .`, then Ruff formatting and safe fixes. +uv run invoke format +# Run `rumdl check .`, then Ruff, Pylint, yamllint, and ty; stop at the first failure. +uv run invoke lint ``` +`invoke lint` stops after the first gate that fails. Pylint does not pass on a clean +checkout: its leg compares the run against a recorded baseline and fails only on a new +diagnostic code or a count above the recorded maximum. + ### Validate the CLI After making changes, verify the CLI still works: @@ -66,14 +81,26 @@ uv run infrahub-sync runs plan --help ### Running tests +The offline gate is what a change has to keep green. It deselects the tests that need a +running stack or an external service: + ```bash -uv run pytest -q +uv run pytest -m "not preview and not integration" -q ``` ### Running the full stack locally To run the Sync HTTP API, its Prefect worker, and a disposable Infrahub against your checkout, see the [local development stack](./development-stack.mdx). +The suite that exercises that stack is opt-in under the `preview` marker, and it skips +rather than fails when the stack is not running: + +```bash +uv run invoke preview.up # start the stack +uv run invoke preview.smoke # seed, then run `pytest -m preview tests/preview` +uv run invoke preview.down --volumes +``` + ## Code standards ### Python style @@ -81,7 +108,7 @@ To run the Sync HTTP API, its Prefect worker, and a disposable Infrahub against - Python 3.10–3.13 compatible - Type hints on new or changed code - Ruff-formatted and lint-clean -- Mypy-checked (do not increase existing error count) +- Clean under `ty`; do not add `[[tool.ty.overrides]]` blocks to mask an error - Public functions and classes require documentation strings - Raise specific exceptions; avoid broad `except Exception:` @@ -104,7 +131,7 @@ uv run invoke docs.generate First-time setup (requires Node.js): ```bash -cd docs && npm install +cd docs && pnpm install --frozen-lockfile ``` Build the site: @@ -115,9 +142,19 @@ uv run invoke docs.docusaurus ### Lint markdown files +Markdown structure is checked with [rumdl](https://github.com/rvben/rumdl), configured in +`pyproject.toml`: + +```bash +uv run invoke docs.format-rumdl # run `rumdl fmt .` +uv run invoke docs.rumdl # run `rumdl check .` +``` + +Prose style is checked with [Vale](https://vale.sh/), configured in `.vale.ini`. Run it on +the files you changed: + ```bash -npx markdownlint-cli "docs/docs/**/*.{md,mdx}" -npx markdownlint-cli --fix "docs/docs/**/*.{md,mdx}" +vale docs/docs/contributing.mdx ``` ## Adding a new adapter @@ -145,7 +182,10 @@ Common tasks: | `linter.lint-ruff` | Lint Python code with ruff | | `linter.lint-pylint` | Lint Python code with pylint | | `linter.lint-yaml` | Lint YAML files with yamllint | +| `linter.lint-ty` | Type-check with ty | +| `docs.format-rumdl` | Format Markdown and MDX with rumdl | +| `docs.rumdl` | Lint Markdown and MDX with rumdl | | `docs.generate` | Generate CLI documentation | | `docs.docusaurus` | Build documentation website | -| `format` | Alias for ruff format | -| `lint` | Run all linters | +| `format` | Run rumdl formatting, then Ruff formatting and safe fixes | +| `lint` | Run rumdl, Ruff, Pylint, yamllint, and ty in order | diff --git a/docs/docs/reference/durable-product-records.mdx b/docs/docs/reference/durable-product-records.mdx index 39aab882..42c3bce1 100644 --- a/docs/docs/reference/durable-product-records.mdx +++ b/docs/docs/reference/durable-product-records.mdx @@ -60,7 +60,7 @@ that reports. | `credential-path-not-declared` | A `$credential` reference sits somewhere that does not accept one. Usually a misspelled setting name, or a reference placed in `schema_mapping` or `order`. | the referencing node | | `endpoint-not-absolute` | A `url` or `base_url` setting is not an absolute `http` or `https` URL. | the setting | | `endpoint-not-relative` | An `api_endpoint` or `endpoint` setting carries a scheme or a host. It names a path beneath the absolute URL, not a second address. | the setting | -| `finding-limit-reached` | The package carries more than 256 defects and the rest were not reported. It is reported first, not last, and it is counted as a finding: when it fires the reported set holds 257 items, not 256. | the whole package | +| `finding-limit-reached` | The package carries more than 256 defects and the rest were not reported. It is reported first, not last, and it is counted as a finding: when it fires the reported set holds 257 items, not 256. Its severity is the highest one among the findings it stands for: `error` when any suppressed finding is an error, `warning` when they are all warnings — so a cut can neither hide an error nor invent one. | the whole package | | `inline-credential-value` | A credential-bearing setting holds a literal value instead of a `{"$credential": ""}` reference. A package never contains a credential value. | the setting | | `malformed-credential-reference` | A credential declaration's environment identifier is not a valid variable name, or a `$credential` node carries keys beyond the reference itself. | the declaration or the node | | `missing-adapter` | No adapter is installed under the declared name — check the spelling and the case. That role's settings are not judged at all, because there is no declared surface to judge them against, so expect exactly one finding for the role. | `/configuration/` | diff --git a/docs/docs/tutorials/netbox-demo-to-infrahub.mdx b/docs/docs/tutorials/netbox-demo-to-infrahub.mdx index 5c853cf2..5458ccad 100644 --- a/docs/docs/tutorials/netbox-demo-to-infrahub.mdx +++ b/docs/docs/tutorials/netbox-demo-to-infrahub.mdx @@ -303,7 +303,10 @@ export INFRAHUB_API_TOKEN="06438eb2-8019-4776-878c-0941b1f1d1ec" ``` Use the complete NetBox token created earlier. The worker, not the CLI, reads the NetBox -and Infrahub adapter credentials. +and Infrahub adapter credentials. The package declares them as references — `$credential` +entries under `credentials` naming `NETBOX_TOKEN` and `INFRAHUB_API_TOKEN` — so the worker +resolves the values from these variables. No NetBox or Infrahub adapter credential value is +written into the package or sent to the Sync API. 4. In terminal 2, create the process pool, deploy the service flow, and start its worker: @@ -457,9 +460,11 @@ docker compose -f sync-services.yml down ERROR | infrahub_sync.cli | Failed to initialize the Sync Instance: Error initializing InfrahubAdapter: Both url and token must be specified! ``` -The service worker cannot find a NetBox or Infrahub adapter credential. Adapter credentials -belong in the worker environment, not the CLI environment or `config.yml`. Export these -variables before starting the Prefect worker: +The service worker cannot resolve a credential reference declared by the package. The +package and `config.yml` contain credential references and declarations, but secret +credential values belong only in the worker environment. The package names `NETBOX_TOKEN` +and `INFRAHUB_API_TOKEN`, and the worker reads their values. Export these variables before +starting the Prefect worker: ```bash export NETBOX_URL="https://demo.netbox.dev" diff --git a/examples/netbox_to_infrahub/config.yml b/examples/netbox_to_infrahub/config.yml index d3a35cda..358cc073 100644 --- a/examples/netbox_to_infrahub/config.yml +++ b/examples/netbox_to_infrahub/config.yml @@ -34,14 +34,20 @@ source: name: netbox settings: url: "https://demo.netbox.dev" + token: + $credential: netbox-token destination: name: infrahub settings: url: "http://localhost:8000" + token: + $credential: infrahub-token -# Adapter credentials are read by the service worker, not by the CLI and not -# from this file: +# This file is the `configuration` body of package.yml, byte for byte, which is +# where the `credentials` block naming these two references lives. Each token is +# a reference, resolved by the service worker from its own environment. No +# credential value is ever written here. Export before starting the worker: # export NETBOX_URL="https://demo.netbox.dev" # export NETBOX_TOKEN="nbt_..." # export INFRAHUB_ADDRESS="http://localhost:8000" diff --git a/examples/netbox_to_infrahub/package.yml b/examples/netbox_to_infrahub/package.yml index 68fc8b34..e6728f85 100644 --- a/examples/netbox_to_infrahub/package.yml +++ b/examples/netbox_to_infrahub/package.yml @@ -35,14 +35,19 @@ configuration: name: netbox settings: url: "https://demo.netbox.dev" + token: + $credential: netbox-token destination: name: infrahub settings: url: "http://localhost:8000" + token: + $credential: infrahub-token - # Adapter credentials are read by the service worker, not by the CLI and not - # from this file: + # Each token above is a reference, resolved by the service worker from its own + # environment. No credential value is ever written here, posted to the Sync + # API, or recorded in a registered version. Export before starting the worker: # export NETBOX_URL="https://demo.netbox.dev" # export NETBOX_TOKEN="nbt_..." # export INFRAHUB_ADDRESS="http://localhost:8000" @@ -673,3 +678,13 @@ configuration: # - IpamSVLAN / IpamCVLAN (schemas/extensions/qinq): Infrahub-only # concept (802.1ad Q-in-Q split), no equivalent Netbox source field. # ----------------------------------------------------------------------- + +# What each `$credential` above points at. `env` is the installed provider; the +# identifier is the exact environment variable name the worker reads. +credentials: + netbox-token: + provider: env + identifier: NETBOX_TOKEN + infrahub-token: + provider: env + identifier: INFRAHUB_API_TOKEN diff --git a/tests/preview/conftest.py b/tests/preview/conftest.py index d3bd0c80..50a158a8 100644 --- a/tests/preview/conftest.py +++ b/tests/preview/conftest.py @@ -51,6 +51,18 @@ def pytest_collection_modifyitems(items: list[Item]) -> None: items[resume_at:resume_at] = observers +@pytest.fixture(scope="session") +def evidence_dir() -> Path: + """Where the qualification rows write their captured HTTP transcripts. + + Under the preview's own gitignored runtime state, next to the service logs, so one + directory holds everything a run of the matrix leaves behind for a reader. + """ + path = REPO_ROOT / ".preview" / "evidence" + path.mkdir(parents=True, exist_ok=True) + return path + + @pytest.fixture(scope="session") def preview_settings() -> dict[str, Any]: """Shipped-plus-local preview settings, with derived URLs and tokens. diff --git a/tests/preview/evidence.py b/tests/preview/evidence.py new file mode 100644 index 00000000..f1d270e8 --- /dev/null +++ b/tests/preview/evidence.py @@ -0,0 +1,77 @@ +"""What the live qualification rows capture: an HTTP transcript and a secret canary scan. + +Two helpers, shared by every module that drives the running stack. The transcript hook is +what makes an in-process `httpx` client an auditable direct-HTTP surface: without a +recorded request and response pair, "the raw route was exercised" rests on the test's own +narration. The canary scan is the single place the preview's Infrahub token is looked for, +so a newly captured artifact is covered by naming it in one mapping rather than by writing +another check. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + from pathlib import Path + + import httpx + +REDACTED = "" + + +def _headers(headers: httpx.Headers) -> dict[str, str]: + """Header names with the authorization value replaced, never the header itself. + + Which routes were called authenticated is the evidence; the bearer value is not. + """ + return {name: (REDACTED if name.lower() == "authorization" else value) for name, value in headers.items()} + + +def transcript_hooks(path: Path) -> dict[str, list[Callable[[httpx.Response], None]]]: + """Return `httpx` event hooks appending one JSON record per exchange to `path`. + + The hook reads the response before serializing it. A response handed to an event hook + has not been read yet, and touching `.text` first raises `ResponseNotRead` — which + would lose the exchange the transcript exists to record. + """ + path.write_text("", encoding="utf-8") + + def record(response: httpx.Response) -> None: + response.read() + request = response.request + entry = { + "method": request.method, + "path": request.url.path, + "status": response.status_code, + "request_headers": _headers(request.headers), + "response_headers": _headers(response.headers), + "request_body": request.content.decode("utf-8", "replace"), + "response_body": response.text, + } + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(entry, sort_keys=True) + "\n") + + return {"response": [record]} + + +def canary_leaks(canary: str, artifacts: Mapping[str, object]) -> list[str]: + """Name every captured artifact whose rendered text carries the canary value. + + Bytes are decoded and everything else is rendered through `repr`, because a run + produces exactly those three shapes: CLI output and transcript text, artifact and + response bytes, and the typed resources the Python client returns. + """ + leaked = [] + for name, artifact in sorted(artifacts.items()): + if isinstance(artifact, str): + text = artifact + elif isinstance(artifact, bytes): + text = artifact.decode("utf-8", "replace") + else: + text = repr(artifact) + if canary in text: + leaked.append(name) + return leaked diff --git a/tests/preview/test_cli_client.py b/tests/preview/test_cli_client.py index bf21db03..b6aa4b7b 100644 --- a/tests/preview/test_cli_client.py +++ b/tests/preview/test_cli_client.py @@ -25,11 +25,12 @@ import uuid from typing import TYPE_CHECKING, Any -import httpx import pytest from tasks.preview import REPO_ROOT, SHARED_DEVICE_NAME, SMOKE_BRANCH +from tests.preview.evidence import canary_leaks from tests.preview.test_service_api import ( + authenticated_client, device_types, infrahub_client, seed_source_branch, @@ -38,6 +39,7 @@ ) if TYPE_CHECKING: + from collections.abc import MutableMapping from pathlib import Path pytestmark = pytest.mark.preview @@ -48,10 +50,10 @@ POLL_INTERVAL_SECONDS = 3 PROCESS_TIMEOUT_SECONDS = WAIT_TIMEOUT_SECONDS + 60 -_ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") +ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") -def _cli_environment(preview_env: dict[str, Any]) -> dict[str, str]: +def cli_environment(preview_env: dict[str, Any]) -> dict[str, str]: """The settings one CLI invocation needs, with the renderer pinned. `NO_COLOR` and a fixed `COLUMNS` remove the only two things that make the shipped @@ -67,69 +69,91 @@ def _cli_environment(preview_env: dict[str, Any]) -> dict[str, str]: } -def _fields(output: str) -> dict[str, str]: +def fields(output: str) -> dict[str, str]: """Parse the CLI's `name: value` field lines, ignoring anything else it prints.""" fields: dict[str, str] = {} - for line in _ANSI.sub("", output).splitlines(): + for line in ANSI.sub("", output).splitlines(): name, separator, value = line.partition(": ") if separator and name.isidentifier(): fields[name] = value.strip() return fields -def _run_cli(preview_env: dict[str, Any], *arguments: str) -> dict[str, str]: - """Run the installed console script to success and return its parsed fields.""" +def run_cli_command( + preview_env: dict[str, Any], + *arguments: str, + artifacts: MutableMapping[str, object], + artifact_name: str, +) -> subprocess.CompletedProcess[str]: + """Run the installed console script and return the completed process, whatever it did. + + The exit code is part of the shipped contract for several rows — a refused `--kind` + filter, an expired bounded wait — so the process is returned rather than asserted on. + """ completed = subprocess.run( # noqa: S603 ["uv", "run", "infrahub-sync", *arguments], # noqa: S607 — resolved from PATH by design cwd=REPO_ROOT, - env=_cli_environment(preview_env), + env=cli_environment(preview_env), capture_output=True, text=True, timeout=PROCESS_TIMEOUT_SECONDS, check=False, ) + artifacts[f"{artifact_name} stdout"] = completed.stdout + artifacts[f"{artifact_name} stderr"] = completed.stderr + return completed + + +def run_cli( + preview_env: dict[str, Any], + *arguments: str, + artifacts: MutableMapping[str, object], + artifact_name: str, +) -> dict[str, str]: + """Run the installed console script to success and return its parsed fields.""" + completed = run_cli_command( + preview_env, + *arguments, + artifacts=artifacts, + artifact_name=artifact_name, + ) assert completed.returncode == 0, ( f"`infrahub-sync {' '.join(arguments)}` exited {completed.returncode}\n" f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}" ) - return _fields(completed.stdout) + return fields(completed.stdout) -def _api(preview_env: dict[str, Any]) -> httpx.Client: - """The HTTP oracle: what the service recorded, read independently of the CLI.""" - return httpx.Client( - base_url=preview_env["urls"]["sync_api"], - headers={"Authorization": f"Bearer {preview_env['bearer_token']}"}, - timeout=30, - ) - - -def _package_file(preview_env: dict[str, Any], directory: Path) -> Path: +def package_file(preview_env: dict[str, Any], directory: Path) -> Path: + """Write the smoke package where `configs register` can read it as an argument.""" path = directory / "preview-smoke-package.json" path.write_text(json.dumps(smoke_package(preview_env["urls"]["infrahub"])), encoding="utf-8") return path -def test_cli_registers_plans_reviews_and_applies_against_the_service( - preview_env: dict[str, Any], tmp_path: Path +def test_cli_registers_plans_reviews_and_applies_against_the_service( # noqa: PLR0914 + preview_env: dict[str, Any], tmp_path: Path, evidence_dir: Path ) -> None: """`configs register` → `diff` → `runs plan` → `apply`, proved through the API.""" mutated_type = seed_source_branch(preview_env) assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] != mutated_type + artifacts: dict[str, object] = {} - registered = _run_cli( + registered = run_cli( preview_env, "configs", "register", - str(_package_file(preview_env, tmp_path)), + str(package_file(preview_env, tmp_path)), "--reason", "preview CLI smoke: register the smoke configuration", "--idempotency-key", f"preview-cli-{uuid.uuid4()}", + artifacts=artifacts, + artifact_name="CLI lifecycle configs register", ) config_id, registry_version = registered["config_id"], registered["registry_version"] - planned = _run_cli( + planned = run_cli( preview_env, "diff", "--config-id", @@ -146,10 +170,13 @@ def test_cli_registers_plans_reviews_and_applies_against_the_service( str(WAIT_TIMEOUT_SECONDS), "--poll-interval", str(POLL_INTERVAL_SECONDS), + artifacts=artifacts, + artifact_name="CLI lifecycle diff", ) run_id, checksum = planned["run_id"], planned["plan_checksum"] - with _api(preview_env) as client: + oracle_transcript = evidence_dir / "cli-lifecycle-oracle-http.jsonl" + with authenticated_client(preview_env, transcript=oracle_transcript) as client: plan = client.get(f"/runs/{run_id}/plan") assert plan.status_code == 200, plan.text summary = plan.json()["summary"] @@ -158,12 +185,19 @@ def test_cli_registers_plans_reviews_and_applies_against_the_service( assert unwritten_plan_reasons(summary) == [], summary assert summary["by_action"].get("update", 0) == 1, summary - reviewed = _run_cli(preview_env, "runs", "plan", run_id) + reviewed = run_cli( + preview_env, + "runs", + "plan", + run_id, + artifacts=artifacts, + artifact_name="CLI lifecycle runs plan", + ) assert reviewed["run_id"] == run_id assert reviewed["plan_checksum"] == checksum assert reviewed["checksum_ok"] == "true" - _run_cli( + run_cli( preview_env, "apply", run_id, @@ -179,9 +213,12 @@ def test_cli_registers_plans_reviews_and_applies_against_the_service( str(WAIT_TIMEOUT_SECONDS), "--poll-interval", str(POLL_INTERVAL_SECONDS), + artifacts=artifacts, + artifact_name="CLI lifecycle apply", ) - with _api(preview_env) as client: + results_transcript = evidence_dir / "cli-lifecycle-results-oracle-http.jsonl" + with authenticated_client(preview_env, transcript=results_transcript) as client: recorded = client.get(f"/runs/{run_id}") assert recorded.status_code == 200, recorded.text assert recorded.json()["run"]["phase"] == "applied", recorded.text @@ -191,4 +228,13 @@ def test_cli_registers_plans_reviews_and_applies_against_the_service( assert applied_summary.get("update", 0) > 0, applied_summary assert applied_summary.get("delete", 0) == 0, applied_summary + artifacts.update( + { + str(oracle_transcript): oracle_transcript.read_text(encoding="utf-8"), + "CLI lifecycle results body": results.content, + } + ) + artifacts[str(results_transcript)] = results_transcript.read_text(encoding="utf-8") + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] == mutated_type diff --git a/tests/preview/test_config_lifecycle.py b/tests/preview/test_config_lifecycle.py new file mode 100644 index 00000000..99a6d0bc --- /dev/null +++ b/tests/preview/test_config_lifecycle.py @@ -0,0 +1,328 @@ +"""Configuration and discovery parity: every registry route through every interface. + +The lifecycle smokes register once and move straight to a run. The registry has seven +resources and the service has two unauthenticated ones, and a client can interpret any of +them wrongly on its own — so this module drives all nine through the CLI, the typed +`SyncClient`, and raw HTTP against the same running service, and compares what each one +returns with what the service recorded. + +Nothing here admits a run, so the module is not one of the Prefect surface's creators. It +registers its own configurations rather than reusing the lifecycle smokes': registration +is one of the rows, replay has to be observed on a key nothing else has used, and a second +version must be the second version of a configuration this module owns. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +import pytest + +from infrahub_sync.client import SyncClient +from infrahub_sync.client.models import ConfigMutationRequest +from tests.preview.evidence import canary_leaks +from tests.preview.test_cli_client import ANSI, package_file, run_cli, run_cli_command +from tests.preview.test_service_api import authenticated_client, idempotency_headers, register_request, smoke_package + +if TYPE_CHECKING: + from pathlib import Path + +pytestmark = pytest.mark.preview + +REASON = "preview qualification: exercise the configuration registry" + + +def revised_package(infrahub_url: str) -> dict[str, Any]: + """The smoke package with one declared setting added, so its checksum differs. + + `create_version` answers 200 with the stored version for a package the configuration + already holds and 201 for a new one, so proving both needs two packages that differ in + declared content and nothing else. `verify_ssl` is a declared `infrahub` setting and is + inert against the preview's plain-HTTP endpoint. + """ + package = smoke_package(infrahub_url) + package["configuration"]["destination"]["settings"]["verify_ssl"] = True + return package + + +def test_the_cli_drives_the_registry_and_checks_compatibility_first( + preview_env: dict[str, Any], tmp_path: Path +) -> None: + """register, replay, version 200/201, list, show, versions, get-version, validate.""" + first = package_file(preview_env, tmp_path) + second = tmp_path / "revised-package.json" + second.write_text(json.dumps(revised_package(preview_env["urls"]["infrahub"])), encoding="utf-8") + artifacts: dict[str, object] = {} + key = idempotency_headers("preview-config")["Idempotency-Key"] + + registered = run_cli( + preview_env, + "configs", + "register", + str(first), + "--reason", + REASON, + "--idempotency-key", + key, + artifacts=artifacts, + artifact_name="config lifecycle register", + ) + config_id = registered["config_id"] + replayed = run_cli( + preview_env, + "configs", + "register", + str(first), + "--reason", + REASON, + "--idempotency-key", + key, + artifacts=artifacts, + artifact_name="config lifecycle register replay", + ) + # A replayed key returns the stored response, so every field — the created timestamp + # above all — is the first one, not a second registration that happens to look alike. + assert replayed == registered + + identical = run_cli( + preview_env, + "configs", + "version", + config_id, + str(first), + "--reason", + REASON, + artifacts=artifacts, + artifact_name="config lifecycle identical version", + ) + assert identical["created"] == "false" + assert identical["registry_version"] == "1" + assert identical["package_checksum"] == registered["package_checksum"] + revised = run_cli( + preview_env, + "configs", + "version", + config_id, + str(second), + "--reason", + REASON, + artifacts=artifacts, + artifact_name="config lifecycle revised version", + ) + assert revised["created"] == "true" + assert revised["registry_version"] == "2" + + listed = run_cli_command( + preview_env, + "configs", + "list", + artifacts=artifacts, + artifact_name="config lifecycle list", + ) + assert listed.returncode == 0, listed.stderr + assert f"config_id: {config_id}" in ANSI.sub("", listed.stdout) + + shown = run_cli( + preview_env, + "configs", + "show", + config_id, + artifacts=artifacts, + artifact_name="config lifecycle show", + ) + assert shown["config_id"] == config_id + shown_version = run_cli( + preview_env, + "configs", + "show", + config_id, + "--version", + "1", + artifacts=artifacts, + artifact_name="config lifecycle show version", + ) + assert shown_version["package_checksum"] == registered["package_checksum"] + + versions = run_cli_command( + preview_env, + "configs", + "versions", + config_id, + artifacts=artifacts, + artifact_name="config lifecycle versions", + ) + assert versions.returncode == 0, versions.stderr + # The replay must not have added a row: exactly the two versions created above. + assert ANSI.sub("", versions.stdout).count("registry_version: ") == 2 + + validated = run_cli( + preview_env, + "configs", + "validate", + config_id, + "1", + artifacts=artifacts, + artifact_name="config lifecycle validate", + ) + assert validated["total_findings"] == "0" + assert validated["destination_schema_fingerprint"] == "" + + # The client checks `/version` before any operation, so a base URL that is not the + # Sync API is refused as an incompatible service rather than as a missing resource. + misdirected = run_cli_command( + preview_env, + "--api-url", + f"{preview_env['urls']['sync_api']}/not-the-sync-api", + "configs", + "list", + artifacts=artifacts, + artifact_name="config lifecycle compatibility refusal", + ) + assert misdirected.returncode == 1 + assert "error: compatibility" in ANSI.sub("", misdirected.stderr) + + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + + +def test_the_python_client_drives_the_registry_and_both_public_resources( # noqa: PLR0914 + preview_env: dict[str, Any], +) -> None: + """`/version`, `/status`, and all seven registry methods through the typed client.""" + package = smoke_package(preview_env["urls"]["infrahub"]) + revised = revised_package(preview_env["urls"]["infrahub"]) + key = idempotency_headers("preview-config")["Idempotency-Key"] + + with SyncClient(preview_env["urls"]["sync_api"], preview_env["bearer_token"], timeout=30.0) as client: + version = client.get_version() + assert "v3-unstable" in version.api_versions + + status = client.get_status() + assert status.service == "ready" + # The preview starts a worker, so an absent one is a broken environment, not a + # tolerable state: only the two live states are admitted here. + assert status.worker.state in {"ready", "busy"}, status.worker + + request = ConfigMutationRequest(package=package, reason=REASON) + registered = client.register_config(request, key) + replayed = client.register_config(request, key) + assert replayed == registered + config_id = registered.version.config_id + + identical = client.create_config_version( + config_id, + ConfigMutationRequest(package=package, reason=REASON), + idempotency_headers("preview-config")["Idempotency-Key"], + ) + assert identical.created is False + assert identical.version.registry_version == 1 + created = client.create_config_version( + config_id, + ConfigMutationRequest(package=revised, reason=REASON), + idempotency_headers("preview-config")["Idempotency-Key"], + ) + assert created.created is True + assert created.version.registry_version == 2 + + listed = client.list_configs() + assert config_id in {summary.config_id for summary in listed} + shown = client.get_config(config_id) + assert shown.config_id == config_id + versions = client.list_config_versions(config_id) + assert [entry.registry_version for entry in versions] == [1, 2] + fetched = client.get_config_version(config_id, 1) + assert fetched == registered.version + + report = client.validate_config(config_id, 1) + assert report.findings == () + assert report.total_findings == 0 + assert report.next_offset is None + # No shipped interface offers the destination-schema opt-in, so the fingerprint the + # report carries is always absent. Pinned here so its arrival is a deliberate change. + assert report.destination_schema_fingerprint is None + + assert ( + canary_leaks( + preview_env["infrahub_token"], + { + "get_version resource": version, + "get_status resource": status, + "register_config resource": registered, + "register_config replay resource": replayed, + "identical create_config_version resource": identical, + "create_config_version resource": created, + "list_configs resource": listed, + "get_config resource": shown, + "list_config_versions resource": versions, + "get_config_version resource": fetched, + "validate_config resource": report, + }, + ) + == [] + ) + + +def test_raw_http_drives_the_registry_and_records_the_transcript( + preview_env: dict[str, Any], evidence_dir: Path +) -> None: + """Every registry and public route over the wire, with the exchange captured.""" + transcript = evidence_dir / "config-lifecycle-http.jsonl" + package = smoke_package(preview_env["urls"]["infrahub"]) + body = register_request(preview_env["urls"]["infrahub"]) + key = idempotency_headers("preview-config") + + with authenticated_client(preview_env, transcript=transcript) as client: + assert client.get("/version").status_code == 200 + status = client.get("/status") + assert status.status_code == 200, status.text + assert status.json()["service"] == "ready" + + registered = client.post("/configs", headers=key, json=body) + assert registered.status_code == 201, registered.text + config_id = registered.json()["version"]["config_id"] + replayed = client.post("/configs", headers=key, json=body) + assert replayed.status_code == 201, replayed.text + assert replayed.json() == registered.json() + + identical = client.post( + f"/configs/{config_id}/versions", + headers=idempotency_headers("preview-config"), + json={"package": package, "reason": REASON}, + ) + assert identical.status_code == 200, identical.text + assert identical.json()["created"] is False + revised = client.post( + f"/configs/{config_id}/versions", + headers=idempotency_headers("preview-config"), + json={"package": revised_package(preview_env["urls"]["infrahub"]), "reason": REASON}, + ) + assert revised.status_code == 201, revised.text + assert revised.json()["version"]["registry_version"] == 2 + + assert config_id in {entry["config_id"] for entry in client.get("/configs").json()} + assert client.get(f"/configs/{config_id}").json()["config_id"] == config_id + versions = client.get(f"/configs/{config_id}/versions").json() + assert [entry["registry_version"] for entry in versions] == [1, 2] + assert client.get(f"/configs/{config_id}/versions/1").json() == registered.json()["version"] + + validated = client.post(f"/configs/{config_id}/versions/1/validate") + assert validated.status_code == 200, validated.text + assert validated.json()["findings"] == [] + assert validated.json()["destination_schema_fingerprint"] is None + + records = [json.loads(line) for line in transcript.read_text(encoding="utf-8").splitlines()] + assert [(record["method"], record["path"], record["status"]) for record in records] == [ + ("GET", "/version", 200), + ("GET", "/status", 200), + ("POST", "/configs", 201), + ("POST", "/configs", 201), + ("POST", f"/configs/{config_id}/versions", 200), + ("POST", f"/configs/{config_id}/versions", 201), + ("GET", "/configs", 200), + ("GET", f"/configs/{config_id}", 200), + ("GET", f"/configs/{config_id}/versions", 200), + ("GET", f"/configs/{config_id}/versions/1", 200), + ("POST", f"/configs/{config_id}/versions/1/validate", 200), + ] + assert {record["request_headers"]["authorization"] for record in records} == {""} + assert canary_leaks(preview_env["infrahub_token"], {str(transcript): transcript.read_text(encoding="utf-8")}) == [] diff --git a/tests/preview/test_evidence.py b/tests/preview/test_evidence.py new file mode 100644 index 00000000..d6d0a2a6 --- /dev/null +++ b/tests/preview/test_evidence.py @@ -0,0 +1,163 @@ +"""Offline cover for the two helpers the live qualification rows capture evidence with. + +These run in the offline gate, not against the preview: both helpers are pure enough to +drive with a mock transport and plain values, and the properties they carry — an +authorization value never reaching a transcript, a token never reaching any captured +artifact — are exactly the ones a live run cannot be relied on to exercise. The live rows +apply them; this module proves they work. +""" + +from __future__ import annotations + +import json +from inspect import Parameter, signature +from typing import TYPE_CHECKING, Any + +import httpx +import pytest + +from tasks.preview import REPO_ROOT +from tests.preview.evidence import canary_leaks, transcript_hooks +from tests.preview.test_cli_client import run_cli, run_cli_command +from tests.preview.test_service_api import authenticated_client + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + +CANARY = "06438eb2-8019-4776-878c-0941b1f1d1ec" + + +def _records(path: Path) -> list[dict[str, Any]]: + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + + +def _client(path: Path, handler: Callable[[httpx.Request], httpx.Response]) -> httpx.Client: + return httpx.Client( + base_url="http://sync.invalid", + transport=httpx.MockTransport(handler), + event_hooks=transcript_hooks(path), + ) + + +def test_the_transcript_records_the_exchange_with_the_authorization_value_removed(tmp_path: Path) -> None: + """Method, path, status and both header sets are recorded; only the value is dropped.""" + path = tmp_path / "transcript.jsonl" + with _client(path, lambda _request: httpx.Response(201, json={"config_id": "c-1"})) as client: + client.post("/configs", headers={"Authorization": "Bearer preview-tester-token-0001"}, json={"reason": "why"}) + + (record,) = _records(path) + assert record["method"] == "POST" + assert record["path"] == "/configs" + assert record["status"] == 201 + # The header name is evidence that the route was called authenticated; the value is not. + assert record["request_headers"]["authorization"] == "" + assert "preview-tester-token-0001" not in path.read_text(encoding="utf-8") + assert json.loads(record["request_body"]) == {"reason": "why"} + assert json.loads(record["response_body"]) == {"config_id": "c-1"} + + +def test_the_transcript_records_a_response_body_the_hook_receives_unread(tmp_path: Path) -> None: + """A streamed response is unread inside the hook; serializing it first loses the body. + + This is the fixture that separates a hook which reads before serializing from one that + does not: `httpx` raises `ResponseNotRead` for the second, so the exchange is never + recorded at all. + """ + path = tmp_path / "transcript.jsonl" + with _client(path, lambda _request: httpx.Response(200, content=iter([b'{"ok":', b"true}"]))) as client: + client.get("/status") + + (record,) = _records(path) + assert json.loads(record["response_body"]) == {"ok": True} + + +def test_canary_leaks_names_every_captured_artifact_carrying_the_token() -> None: + """Text, bytes and an object rendered through `repr` are all scanned, and all named.""" + leaks = canary_leaks( + CANARY, + { + "cli stdout": f"token: {CANARY}", + "artifact bytes": f'{{"token":"{CANARY}"}}'.encode(), + "client resource": httpx.URL(f"http://sync.invalid/?token={CANARY}"), + "clean transcript": '{"path":"/configs"}', + }, + ) + + assert leaks == ["artifact bytes", "cli stdout", "client resource"] + + +def test_canary_leaks_reports_nothing_when_no_artifact_carries_the_token() -> None: + """No leak is reported when the canary token is absent from every artifact.""" + leaks = canary_leaks(CANARY, {"cli stdout": "config_id: c-1", "artifact bytes": b'{"summary":{}}'}) + + assert leaks == [] + + +def test_authenticated_qualification_clients_require_a_transcript_path() -> None: + """A direct-HTTP row cannot silently opt out of its required evidence.""" + transcript = signature(authenticated_client).parameters["transcript"] + + assert transcript.default is Parameter.empty + + +def test_cli_helpers_capture_stdout_and_stderr_before_parsing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The success helper must not discard stderr while returning parsed fields.""" + completed = __import__("subprocess").CompletedProcess( + args=["infrahub-sync", "configs", "list"], + returncode=0, + stdout="config_id: config-1\n", + stderr="diagnostic\n", + ) + + def completed_run(*_args: object, **_kwargs: object) -> object: + return completed + + monkeypatch.setattr("tests.preview.test_cli_client.subprocess.run", completed_run) + artifacts: dict[str, object] = {} + preview_env = {"urls": {"sync_api": "http://sync.invalid"}, "bearer_token": "test-bearer"} + + parsed = run_cli( + preview_env, + "configs", + "list", + artifacts=artifacts, + artifact_name="configs list", + ) + + assert parsed == {"config_id": "config-1"} + assert artifacts == { + "configs list stdout": "config_id: config-1\n", + "configs list stderr": "diagnostic\n", + } + + direct_artifacts: dict[str, object] = {} + run_cli_command( + preview_env, + "configs", + "list", + artifacts=direct_artifacts, + artifact_name="direct configs list", + ) + assert direct_artifacts == { + "direct configs list stdout": "config_id: config-1\n", + "direct configs list stderr": "diagnostic\n", + } + + +@pytest.mark.parametrize(("original", "counterpart"), [("Text", "TextArea"), ("TextArea", "Text")]) +def test_schema_drift_selects_the_reversible_counterpart(original: str, counterpart: str) -> None: + """The live kind, not a seeded-schema assumption, selects the temporary mutation.""" + from tests.preview.test_schema_drift import _reversible_kind + + assert _reversible_kind(original) == counterpart + + +def test_contributing_uses_the_locked_pnpm_install() -> None: + """The contributor setup command matches the pinned package manager and CI.""" + contributing = (REPO_ROOT / "docs/docs/contributing.mdx").read_text(encoding="utf-8") + + assert "pnpm install --frozen-lockfile" in contributing + assert "cd docs && npm install" not in contributing diff --git a/tests/preview/test_netbox_package.py b/tests/preview/test_netbox_package.py new file mode 100644 index 00000000..e19870a2 --- /dev/null +++ b/tests/preview/test_netbox_package.py @@ -0,0 +1,302 @@ +"""The shipped `from-netbox` package against the running service: register and validate. + +The preview stack has no NetBox source, no schema library and no NetBox token, so this +module never plans or applies this package. What it does prove is the part that needs no +source: the package a reader actually copies out of `examples/` registers through all +three interfaces, resolves its credentials by reference rather than by value, and reports +the same findings whichever interface asked. + +The last row is the one the package's shape rests on. Default validation judges declared +content only — no schema read, no network — so the source URL is pointed at a listener +that answers anything and records what it was asked, and the recording has to stay empty. +""" + +from __future__ import annotations + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from threading import Thread +from typing import TYPE_CHECKING, Any + +import httpx +import pytest +import yaml + +from infrahub_sync.client import SyncClient +from infrahub_sync.client.models import ConfigMutationRequest +from tasks.preview import REPO_ROOT +from tests.preview.evidence import canary_leaks +from tests.preview.test_cli_client import ANSI, run_cli, run_cli_command +from tests.preview.test_service_api import authenticated_client, idempotency_headers + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +pytestmark = pytest.mark.preview + +PACKAGE_FILE = REPO_ROOT / "examples" / "netbox_to_infrahub" / "package.yml" +REASON = "preview qualification: register the shipped NetBox package" +# A destination kind the shipped package deliberately does not map, so declaring it omitted +# is accepted and reports exactly one warning. This is the copy that separates an interface +# which renders findings from one that renders none because there were none to render. +OMITTED_KIND = "DcimCable" +OMISSION_REASON = "NetBox cable terminations have no destination mapping yet" + +_sink_requests: list[str] = [] + + +class _SinkHandler(BaseHTTPRequestHandler): + """Answer anything and record it. A default validation must never reach this.""" + + def _record(self) -> None: + _sink_requests.append(f"{self.command} {self.path}") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b"{}") + + do_GET = _record # noqa: N815 -- the handler's fixed method-dispatch names + do_POST = _record # noqa: N815 + do_HEAD = _record # noqa: N815 + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002, ANN401 -- the base signature + """Stay silent: the recorded request list is this listener's only output.""" + + +@pytest.fixture +def source_sink() -> Iterator[str]: + """A live HTTP listener, and the URL a package can name as its source.""" + _sink_requests.clear() + server = ThreadingHTTPServer(("127.0.0.1", 0), _SinkHandler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def shipped_package() -> dict[str, Any]: + """The package exactly as `examples/netbox_to_infrahub/package.yml` declares it.""" + return dict(yaml.safe_load(PACKAGE_FILE.read_text(encoding="utf-8"))) + + +def _with_omission() -> dict[str, Any]: + """The shipped package plus one declared omission, which is one warning finding.""" + return {**shipped_package(), "omissions": [{"kind": OMITTED_KIND, "reason": OMISSION_REASON}]} + + +def _rendered(code: str, severity: str, location: str, message: str) -> str: + """One finding in the CLI's own line format, so the three renderings compare as bytes.""" + return f"finding: code={code} severity={severity} location={location} message={message}" + + +def _cli_findings(output: str) -> list[str]: + return [line for line in ANSI.sub("", output).splitlines() if line.startswith("finding: ")] + + +def _http_findings(payload: dict[str, Any]) -> list[str]: + return [ + _rendered(finding["code"], finding["severity"], finding["location"], finding["message"]) + for finding in payload["findings"] + ] + + +def _register_over_http(client: Any, package: dict[str, Any]) -> dict[str, Any]: # noqa: ANN401 — the raw client + response = client.post( + "/configs", + headers=idempotency_headers("preview-netbox"), + json={"package": package, "reason": REASON}, + ) + assert response.status_code == 201, response.text + return dict(response.json()["version"]) + + +def test_the_shipped_netbox_package_registers_through_every_interface( # noqa: PLR0914 + preview_env: dict[str, Any], evidence_dir: Path +) -> None: + """The file on disk registers as-is, replays on its key, and checksums the same everywhere.""" + package = shipped_package() + # The credential references are what makes registration possible at all: a package + # carrying a literal token is refused before anything is persisted. + assert package["configuration"]["source"]["settings"]["token"] == {"$credential": "netbox-token"} + assert package["configuration"]["destination"]["settings"]["token"] == {"$credential": "infrahub-token"} + assert package["credentials"] == { + "netbox-token": {"provider": "env", "identifier": "NETBOX_TOKEN"}, + "infrahub-token": {"provider": "env", "identifier": "INFRAHUB_API_TOKEN"}, + } + + artifacts: dict[str, object] = {} + key = idempotency_headers("preview-netbox")["Idempotency-Key"] + from_cli = run_cli( + preview_env, + "configs", + "register", + str(PACKAGE_FILE), + "--reason", + REASON, + "--idempotency-key", + key, + artifacts=artifacts, + artifact_name="NetBox package register", + ) + replayed_cli = run_cli( + preview_env, + "configs", + "register", + str(PACKAGE_FILE), + "--reason", + REASON, + "--idempotency-key", + key, + artifacts=artifacts, + artifact_name="NetBox package register replay", + ) + assert replayed_cli == from_cli + + with SyncClient(preview_env["urls"]["sync_api"], preview_env["bearer_token"], timeout=30.0) as client: + request = ConfigMutationRequest(package=package, reason=REASON) + python_key = idempotency_headers("preview-netbox")["Idempotency-Key"] + from_python = client.register_config(request, python_key) + replayed_python = client.register_config(request, python_key) + assert replayed_python == from_python + + transcript = evidence_dir / "netbox-package-register-http.jsonl" + with authenticated_client(preview_env, transcript=transcript) as api: + http_key = idempotency_headers("preview-netbox") + body = {"package": package, "reason": REASON} + first = api.post("/configs", headers=http_key, json=body) + assert first.status_code == 201, first.text + replayed_http = api.post("/configs", headers=http_key, json=body) + assert replayed_http.json() == first.json() + from_http = first.json()["version"] + + # Three independent registrations of one declared file: different configurations, the + # same content, so the checksum the registry computed has to be one value. + assert from_cli["package_checksum"] == from_python.version.package_checksum == from_http["package_checksum"] + assert len({from_cli["config_id"], from_python.version.config_id, from_http["config_id"]}) == 3 + captured = transcript.read_text(encoding="utf-8") + exchanges = [ + (record["method"], record["path"], record["status"]) for record in map(json.loads, captured.splitlines()) + ] + assert exchanges == [("POST", "/configs", 201), ("POST", "/configs", 201)] + artifacts.update( + { + "NetBox Python register resource": from_python, + "NetBox Python register replay resource": replayed_python, + str(transcript): captured, + } + ) + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + + +def test_the_netbox_package_validates_identically_through_every_interface( + preview_env: dict[str, Any], evidence_dir: Path +) -> None: + """Findings are byte-identical across the interfaces, both when empty and when not.""" + transcript = evidence_dir / "netbox-package-validate-http.jsonl" + artifacts: dict[str, object] = {} + python_reports: list[object] = [] + http_bodies: list[bytes] = [] + with authenticated_client(preview_env, transcript=transcript) as api: + clean = _register_over_http(api, shipped_package()) + warned = _register_over_http(api, _with_omission()) + expected = { + (clean["config_id"], clean["registry_version"]): [], + (warned["config_id"], warned["registry_version"]): [ + _rendered( + "intentional-omission", + "warning", + "/omissions/0", + f"declared content is intentionally omitted from synchronization: {OMISSION_REASON}", + ) + ], + } + + for index, ((config_id, registry_version), findings) in enumerate(expected.items(), start=1): + version = str(registry_version) + cli = run_cli_command( + preview_env, + "configs", + "validate", + config_id, + version, + artifacts=artifacts, + artifact_name=f"NetBox package validate {index}", + ) + assert cli.returncode == 0, cli.stderr + + with SyncClient(preview_env["urls"]["sync_api"], preview_env["bearer_token"], timeout=30.0) as client: + report = client.validate_config(config_id, registry_version) + python_reports.append(report) + body = api.post(f"/configs/{config_id}/versions/{version}/validate") + assert body.status_code == 200, body.text + http_bodies.append(body.content) + + python_findings = [ + _rendered(item.code, item.severity, item.location, item.message) for item in report.findings + ] + assert _cli_findings(cli.stdout) == python_findings == _http_findings(body.json()) == findings + assert report.total_findings == len(findings) + assert body.json()["total_findings"] == len(findings) + # Decision: no interface exposes the destination-schema opt-in, so this stays absent. + assert report.destination_schema_fingerprint is None + assert body.json()["destination_schema_fingerprint"] is None + + captured = transcript.read_text(encoding="utf-8") + exchanges = [ + (record["method"], record["path"], record["status"]) for record in map(json.loads, captured.splitlines()) + ] + assert exchanges == [ + ("POST", "/configs", 201), + ("POST", "/configs", 201), + ("POST", f"/configs/{clean['config_id']}/versions/{clean['registry_version']}/validate", 200), + ("POST", f"/configs/{warned['config_id']}/versions/{warned['registry_version']}/validate", 200), + ] + artifacts[str(transcript)] = captured + artifacts.update( + {f"NetBox Python validation resource {index}": report for index, report in enumerate(python_reports)} + ) + artifacts.update({f"NetBox HTTP validation body {index}": body for index, body in enumerate(http_bodies)}) + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + + +def test_validating_the_netbox_package_reads_no_source( + preview_env: dict[str, Any], source_sink: str, evidence_dir: Path +) -> None: + """Default validation judges declared content, so the source endpoint is never called.""" + package = shipped_package() + package["configuration"]["source"]["settings"]["url"] = source_sink + # Prove the listener records before relying on it recording nothing; an empty log from + # a listener nothing could have reached would be evidence of the harness, not the route. + assert httpx.get(source_sink, timeout=5).status_code == 200 + assert _sink_requests == ["GET /"] + _sink_requests.clear() + + transcript = evidence_dir / "netbox-package-zero-source-http.jsonl" + with authenticated_client(preview_env, transcript=transcript) as api: + version = _register_over_http(api, package) + assert version["declared_content"]["configuration"]["source"]["settings"]["url"] == source_sink + report = api.post(f"/configs/{version['config_id']}/versions/{version['registry_version']}/validate") + assert report.status_code == 200, report.text + assert report.json()["findings"] == [], report.text + + assert _sink_requests == [] + captured = transcript.read_text(encoding="utf-8") + exchanges = [ + (record["method"], record["path"], record["status"]) for record in map(json.loads, captured.splitlines()) + ] + assert exchanges == [ + ("POST", "/configs", 201), + ("POST", f"/configs/{version['config_id']}/versions/{version['registry_version']}/validate", 200), + ] + assert ( + canary_leaks( + preview_env["infrahub_token"], + {str(transcript): captured, "NetBox zero-source validation body": report.content}, + ) + == [] + ) diff --git a/tests/preview/test_python_client.py b/tests/preview/test_python_client.py index efad5bf7..343c7f1f 100644 --- a/tests/preview/test_python_client.py +++ b/tests/preview/test_python_client.py @@ -21,6 +21,7 @@ from infrahub_sync.client import SyncClient from infrahub_sync.client.models import ApplyRunRequest, ConfigMutationRequest, CreateRunRequest from tasks.preview import SHARED_DEVICE_NAME, SMOKE_BRANCH +from tests.preview.evidence import canary_leaks from tests.preview.test_service_api import ( device_types, infrahub_client, @@ -99,4 +100,21 @@ def test_sync_client_registers_plans_and_applies_against_the_service(preview_env assert applied_summary.get("update", 0) > 0, applied_summary assert applied_summary.get("delete", 0) == 0, applied_summary + assert ( + canary_leaks( + preview_env["infrahub_token"], + { + "Python lifecycle register resource": registered, + "Python lifecycle validation resource": report, + "Python lifecycle plan accepted resource": accepted, + "Python lifecycle planned resource": planned, + "Python lifecycle plan resource": plan, + "Python lifecycle apply accepted resource": apply_accepted, + "Python lifecycle applied resource": applied, + "Python lifecycle results resource": results, + }, + ) + == [] + ) + assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] == mutated_type diff --git a/tests/preview/test_run_completion.py b/tests/preview/test_run_completion.py new file mode 100644 index 00000000..2c8a889a --- /dev/null +++ b/tests/preview/test_run_completion.py @@ -0,0 +1,350 @@ +"""Completion parity: the one-run synchronization, and cancelling a run that is in flight. + +`sync` plans, verifies and applies inside a single admitted run, so it is the one shipped +operation whose writes are never gated on a separate reviewed checksum — each interface +that offers it drives it here against the running service, with the destination read back +through the Infrahub SDK. + +Cancellation races the worker by nature: a small plan can finish before the request lands. +The rows below assert the two outcomes the service defines for that race and record which +one happened, rather than retrying until one of them appears. +""" + +from __future__ import annotations + +import json +import time +from typing import TYPE_CHECKING, Any + +import httpx +import pytest + +from infrahub_sync.client import APIError, SyncClient +from infrahub_sync.client.models import CancelRunRequest, ConfigMutationRequest, CreateRunRequest +from tasks.preview import SHARED_DEVICE_NAME, SMOKE_BRANCH +from tests.preview.evidence import canary_leaks +from tests.preview.test_cli_client import package_file, run_cli +from tests.preview.test_service_api import ( + authenticated_client, + create_run_request, + device_types, + idempotency_headers, + infrahub_client, + register_request, + seed_source_branch, + smoke_package, + wait_for_phase, +) + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + +pytestmark = pytest.mark.preview + +WAIT_TIMEOUT_SECONDS = 240.0 +POLL_INTERVAL_SECONDS = 3.0 +SYNC_REASON = "preview qualification: synchronize in one run" +CANCEL_REASON = "preview qualification: cancel a run in flight" +CLAIM_TIMEOUT_SECONDS = 120.0 +# The states Prefect reports once an execution can no longer be cancelled. +TERMINAL_EXECUTION_STATES = frozenset({"cancelled", "completed", "crashed", "failed"}) + + +def _sync_request(config_id: str, registry_version: int) -> dict[str, Any]: + """The `POST /runs` body for the confirmed one-run synchronization.""" + return { + **create_run_request(config_id, registry_version), + "operation": "sync", + "confirm_writes": True, + "reason": SYNC_REASON, + } + + +def _record_cancellation(evidence_dir: Path, interface: str, outcome: str) -> None: + """Write which of the two defined cancellation outcomes this run produced.""" + (evidence_dir / f"cancellation-{interface}.txt").write_text(f"{outcome}\n", encoding="utf-8") + + +def _await_claimed_execution(read_state: Callable[[], str | None], run_id: str) -> str | None: + """Poll until the admitted execution is claimed or already terminal.""" + deadline = time.monotonic() + CLAIM_TIMEOUT_SECONDS + state: str | None = None + while time.monotonic() < deadline: + state = read_state() + if state == "running" or state in TERMINAL_EXECUTION_STATES: + return state + time.sleep(0.25) + return pytest.fail(f"run {run_id} was not claimed within {CLAIM_TIMEOUT_SECONDS}s (last state {state!r})") + + +def _await_prefect_terminal_state(preview_env: dict[str, Any], flow_run_id: str) -> str: + """Read the row's own Prefect run until it reaches a terminal state.""" + deadline = time.monotonic() + WAIT_TIMEOUT_SECONDS + state = "" + while time.monotonic() < deadline: + response = httpx.get(f"{preview_env['urls']['prefect']}/api/flow_runs/{flow_run_id}", timeout=15) + assert response.status_code == 200, response.text + state = response.json()["state_type"] + if state not in {"PENDING", "RUNNING", "SCHEDULED", "CANCELLING"}: + return state + time.sleep(POLL_INTERVAL_SECONDS) + return pytest.fail(f"Prefect flow run {flow_run_id} stayed non-terminal ({state})") + + +def test_the_cli_synchronizes_in_one_run(preview_env: dict[str, Any], tmp_path: Path, evidence_dir: Path) -> None: + """`sync` plans and applies under one run id, and the destination carries the value.""" + mutated_type = seed_source_branch(preview_env) + artifacts: dict[str, object] = {} + registered = run_cli( + preview_env, + "configs", + "register", + str(package_file(preview_env, tmp_path)), + "--reason", + SYNC_REASON, + artifacts=artifacts, + artifact_name="CLI sync configs register", + ) + + completed = run_cli( + preview_env, + "sync", + "--config-id", + registered["config_id"], + "--version", + registered["registry_version"], + "--branch", + SMOKE_BRANCH, + "--reason", + SYNC_REASON, + "--wait-timeout", + str(int(WAIT_TIMEOUT_SECONDS)), + "--poll-interval", + str(int(POLL_INTERVAL_SECONDS)), + artifacts=artifacts, + artifact_name="CLI sync", + ) + assert completed["operation"] == "sync" + assert completed["phase"] == "applied" + + oracle_transcript = evidence_dir / "run-sync-cli-oracle-http.jsonl" + with authenticated_client(preview_env, transcript=oracle_transcript) as client: + results = client.get(f"/runs/{completed['run_id']}/results") + assert results.status_code == 200, results.text + recorded = results.json()["results"] + assert recorded["operation"] == "sync" + assert recorded["summary"]["update"] > 0, recorded + + assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] == mutated_type + artifacts[str(oracle_transcript)] = oracle_transcript.read_text(encoding="utf-8") + artifacts["CLI sync results body"] = results.content + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + + +def test_the_python_client_synchronizes_in_one_run(preview_env: dict[str, Any]) -> None: + """`SyncClient.sync` completes the confirmed one-run synchronization and records it.""" + mutated_type = seed_source_branch(preview_env) + + with SyncClient(preview_env["urls"]["sync_api"], preview_env["bearer_token"], timeout=30.0) as client: + registered = client.register_config( + ConfigMutationRequest(package=smoke_package(preview_env["urls"]["infrahub"]), reason=SYNC_REASON), + idempotency_headers("preview-completion")["Idempotency-Key"], + ) + accepted = client.sync( + CreateRunRequest( + operation="sync", + config_id=registered.version.config_id, + registry_version=registered.version.registry_version, + branch=SMOKE_BRANCH, + confirm_writes=True, + reason=SYNC_REASON, + ), + idempotency_headers("preview-completion")["Idempotency-Key"], + ) + applied = client.wait_for_run(accepted, timeout=WAIT_TIMEOUT_SECONDS, poll_interval=POLL_INTERVAL_SECONDS) + assert applied.run.phase == "applied", applied.run + + results = client.get_results(accepted.run.run_id) + assert results.results["operation"] == "sync" + assert results.results["summary"]["update"] > 0, results.results + + assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] == mutated_type + assert ( + canary_leaks( + preview_env["infrahub_token"], + { + "sync register resource": registered, + "sync accepted resource": accepted, + "sync run resource": applied, + "sync results resource": results, + }, + ) + == [] + ) + + +def test_raw_http_synchronizes_in_one_run(preview_env: dict[str, Any], evidence_dir: Path) -> None: + """The confirmed one-run synchronization over the wire, with the exchange captured.""" + mutated_type = seed_source_branch(preview_env) + transcript = evidence_dir / "run-sync-http.jsonl" + + with authenticated_client(preview_env, transcript=transcript) as client: + registered = client.post( + "/configs", + headers=idempotency_headers("preview-completion"), + json=register_request(preview_env["urls"]["infrahub"]), + ) + assert registered.status_code == 201, registered.text + version = registered.json()["version"] + + created = client.post( + "/runs", + headers=idempotency_headers("preview-completion"), + json=_sync_request(version["config_id"], version["registry_version"]), + ) + assert created.status_code == 202, created.text + run_id = created.json()["run"]["run_id"] + + applied = wait_for_phase(client, run_id, "applied") + assert applied["run"]["summary"]["update"] > 0, applied["run"] + assert applied["run"]["summary"]["delete"] == 0, applied["run"] + + assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] == mutated_type + captured = transcript.read_text(encoding="utf-8") + assert ("POST", "/runs", 202) in { + (json.loads(line)["method"], json.loads(line)["path"], json.loads(line)["status"]) + for line in captured.splitlines() + } + assert canary_leaks(preview_env["infrahub_token"], {str(transcript): captured}) == [] + + +def test_the_python_client_cancels_a_run_it_has_just_admitted(preview_env: dict[str, Any], evidence_dir: Path) -> None: + """Either the run reaches the cancelled terminal state, or the cancel is refused as late.""" + seed_source_branch(preview_env) + + with SyncClient(preview_env["urls"]["sync_api"], preview_env["bearer_token"], timeout=30.0) as client: + registered = client.register_config( + ConfigMutationRequest(package=smoke_package(preview_env["urls"]["infrahub"]), reason=CANCEL_REASON), + idempotency_headers("preview-completion")["Idempotency-Key"], + ) + accepted = client.plan( + CreateRunRequest( + operation="plan", + config_id=registered.version.config_id, + registry_version=registered.version.registry_version, + branch=SMOKE_BRANCH, + reason=CANCEL_REASON, + ), + idempotency_headers("preview-completion")["Idempotency-Key"], + ) + run_id = accepted.run.run_id + flow_run_id = accepted.orchestration[-1].flow_run_id + claimed_state = _await_claimed_execution(lambda: client.get_run(run_id).orchestration[-1].state, run_id) + + refusal: APIError | None = None + cancel_accepted: object | None = None + try: + cancel_accepted = client.cancel_run( + run_id, + CancelRunRequest(reason=CANCEL_REASON), + idempotency_headers("preview-completion")["Idempotency-Key"], + ) + except APIError as error: + refusal = error + + evidence = evidence_dir / "cancellation-python.txt" + artifacts: dict[str, object] = { + "Python cancellation register resource": registered, + "Python cancellation accepted resource": accepted, + "Python cancellation claimed state": claimed_state, + } + if refusal is not None: + # The execution finished between admission and this request. The refusal, not a + # retry, is the contract for that: the service will not cancel a terminal run. + assert refusal.status == 409, refusal + assert refusal.code == "execution-terminal", refusal + _record_cancellation(evidence_dir, "python", "refused: execution-terminal") + artifacts["Python cancellation refusal"] = refusal + assert _await_prefect_terminal_state(preview_env, flow_run_id) in { + "CANCELLED", + "COMPLETED", + "CRASHED", + "FAILED", + } + else: + transcript = evidence_dir / "run-cancel-python-oracle-http.jsonl" + with authenticated_client(preview_env, transcript=transcript) as api: + cancelled = wait_for_phase(api, run_id, "cancelled") + assert cancelled["run"]["outcome"] == "cancelled", cancelled["run"] + assert _await_prefect_terminal_state(preview_env, flow_run_id) == "CANCELLED" + _record_cancellation(evidence_dir, "python", "accepted: run cancelled") + artifacts.update( + { + "Python cancellation response resource": cancel_accepted, + str(transcript): transcript.read_text(encoding="utf-8"), + } + ) + artifacts[str(evidence)] = evidence.read_bytes() + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + + +def test_raw_http_cancels_a_run_it_has_just_admitted(preview_env: dict[str, Any], evidence_dir: Path) -> None: + """The same two outcomes over the wire: 202 then cancelled, or 409 `execution-terminal`.""" + seed_source_branch(preview_env) + transcript = evidence_dir / "run-cancel-http.jsonl" + + with authenticated_client(preview_env, transcript=transcript) as client: + registered = client.post( + "/configs", + headers=idempotency_headers("preview-completion"), + json=register_request(preview_env["urls"]["infrahub"]), + ) + assert registered.status_code == 201, registered.text + version = registered.json()["version"] + + created = client.post( + "/runs", + headers=idempotency_headers("preview-completion"), + json=create_run_request(version["config_id"], version["registry_version"]), + ) + assert created.status_code == 202, created.text + admitted = created.json() + run_id = admitted["run"]["run_id"] + flow_run_id = admitted["orchestration"][-1]["flow_run_id"] + _await_claimed_execution(lambda: client.get(f"/runs/{run_id}").json()["orchestration"][-1]["state"], run_id) + + cancel = client.post( + f"/runs/{run_id}/cancel", + headers=idempotency_headers("preview-completion"), + json={"reason": CANCEL_REASON}, + ) + assert cancel.status_code in {202, 409}, cancel.text + if cancel.status_code == 409: + assert cancel.json()["error"]["code"] == "execution-terminal", cancel.text + assert _await_prefect_terminal_state(preview_env, flow_run_id) in { + "CANCELLED", + "COMPLETED", + "CRASHED", + "FAILED", + } + _record_cancellation(evidence_dir, "http", "refused: execution-terminal") + else: + cancelled = wait_for_phase(client, run_id, "cancelled") + assert cancelled["run"]["outcome"] == "cancelled", cancelled["run"] + assert _await_prefect_terminal_state(preview_env, flow_run_id) == "CANCELLED" + _record_cancellation(evidence_dir, "http", "accepted: run cancelled") + + captured = transcript.read_text(encoding="utf-8") + assert ("POST", f"/runs/{run_id}/cancel", cancel.status_code) in { + (json.loads(line)["method"], json.loads(line)["path"], json.loads(line)["status"]) + for line in captured.splitlines() + } + evidence = evidence_dir / "cancellation-http.txt" + assert ( + canary_leaks( + preview_env["infrahub_token"], + {str(transcript): captured, str(evidence): evidence.read_bytes()}, + ) + == [] + ) diff --git a/tests/preview/test_run_review.py b/tests/preview/test_run_review.py new file mode 100644 index 00000000..b90b44e7 --- /dev/null +++ b/tests/preview/test_run_review.py @@ -0,0 +1,360 @@ +"""Review parity: what an operator can read about a run before approving its writes. + +Everything here stops short of an apply. The saved plan is reviewed through the CLI in +summary and in detail, the bounded wait is proved to expire without touching the remote +run, and the two review resources the CLI does not ship — verification, and the saved-plan +artifact with its digest — are driven through the typed client and over raw HTTP. + +Each test registers its own configuration and creates its own run, because a review row +has to observe a plan whose operations it set up: the shared smoke branch converges as the +other modules apply to it, and a plan with nothing in it renders no operation to filter. +""" + +from __future__ import annotations + +import json +import time +from hashlib import sha256 +from typing import TYPE_CHECKING, Any + +import pytest + +from infrahub_sync.client import SyncClient +from infrahub_sync.client.models import ConfigMutationRequest, CreateRunRequest, VerifyRunRequest +from tasks.preview import SHARED_DEVICE_NAME, SMOKE_BRANCH, SMOKE_KIND +from tests.preview.evidence import canary_leaks +from tests.preview.test_cli_client import ANSI, fields, package_file, run_cli, run_cli_command +from tests.preview.test_service_api import ( + authenticated_client, + create_run_request, + idempotency_headers, + register_request, + seed_source_branch, + smoke_package, + unwritten_plan_reasons, + wait_for_phase, +) + +if TYPE_CHECKING: + from pathlib import Path + +pytestmark = pytest.mark.preview + +WAIT_TIMEOUT_SECONDS = 240.0 +POLL_INTERVAL_SECONDS = 3.0 +PLAN_ARTIFACT_ID = "plan-review" +REASON = "preview qualification: review a saved plan" +# A kind the destination schema does not define, so `--kind` filters the plan to nothing. +ABSENT_KIND = "InfraNotAKind" + + +def _operation_lines(output: str) -> list[str]: + """The `runs plan --detail` operation lines, which are the only ones naming an action.""" + return [line for line in ANSI.sub("", output).splitlines() if f" {SMOKE_KIND} " in line] + + +def _await_verification(client: Any, run_id: str) -> dict[str, Any]: # noqa: ANN401 — the raw httpx client + """Poll the results resource until the verification stage has recorded itself. + + Verification merges into a run that is already planned instead of moving it to a new + phase, so the phase poll the other stages use has nothing here to observe. + """ + deadline = time.monotonic() + WAIT_TIMEOUT_SECONDS + while time.monotonic() < deadline: + results = client.get(f"/runs/{run_id}/results") + assert results.status_code == 200, results.text + verification = results.json()["results"].get("verification") + if verification is not None: + return dict(verification) + time.sleep(POLL_INTERVAL_SECONDS) + pytest.fail(f"run {run_id} recorded no verification result within {WAIT_TIMEOUT_SECONDS}s") + + +def test_the_cli_admits_a_run_without_waiting_then_reviews_its_saved_plan( + preview_env: dict[str, Any], tmp_path: Path, evidence_dir: Path +) -> None: + """`--no-wait`, then `runs plan` in summary, in detail, filtered, and refused.""" + seed_source_branch(preview_env) + artifacts: dict[str, object] = {} + registered = run_cli( + preview_env, + "configs", + "register", + str(package_file(preview_env, tmp_path)), + "--reason", + REASON, + artifacts=artifacts, + artifact_name="run review configs register", + ) + + accepted = run_cli( + preview_env, + "diff", + "--config-id", + registered["config_id"], + "--version", + registered["registry_version"], + "--branch", + SMOKE_BRANCH, + "--reason", + REASON, + "--no-wait", + artifacts=artifacts, + artifact_name="run review diff no-wait", + ) + run_id = accepted["run_id"] + # `--no-wait` renders the accepted run and stops. The plan review the waiting form + # prints afterwards is the field that separates the two, so its absence is the row. + assert "plan_checksum" not in accepted, accepted + assert accepted["operation"] == "plan" + + oracle_transcript = evidence_dir / "run-review-cli-oracle-http.jsonl" + with authenticated_client(preview_env, transcript=oracle_transcript) as client: + planned = wait_for_phase(client, run_id, "planned") + assert planned["run"]["outcome"] is not None, planned["run"] + summary = client.get(f"/runs/{run_id}/plan").json()["summary"] + assert unwritten_plan_reasons(summary) == [], summary + + reviewed = run_cli( + preview_env, + "runs", + "plan", + run_id, + artifacts=artifacts, + artifact_name="run review summary", + ) + assert reviewed["checksum_ok"] == "true" + assert reviewed["operations"] == str(summary["total"]) + + detailed = run_cli_command( + preview_env, + "runs", + "plan", + run_id, + "--detail", + artifacts=artifacts, + artifact_name="run review detail", + ) + assert detailed.returncode == 0, detailed.stderr + assert _operation_lines(detailed.stdout), detailed.stdout + assert any(f"update {SMOKE_KIND} name={SHARED_DEVICE_NAME}" in line for line in _operation_lines(detailed.stdout)) + + filtered = run_cli_command( + preview_env, + "runs", + "plan", + run_id, + "--detail", + "--kind", + SMOKE_KIND, + artifacts=artifacts, + artifact_name="run review filtered detail", + ) + assert filtered.returncode == 0, filtered.stderr + assert _operation_lines(filtered.stdout) == _operation_lines(detailed.stdout) + + # `--kind` is a filter over the detailed list, so it is refused without it rather than + # silently ignored; a kind that matches nothing is refused the same way. + for index, arguments in enumerate((("--kind", SMOKE_KIND), ("--detail", "--kind", ABSENT_KIND)), start=1): + refused = run_cli_command( + preview_env, + "runs", + "plan", + run_id, + *arguments, + artifacts=artifacts, + artifact_name=f"run review refusal {index}", + ) + assert refused.returncode == 2, refused.stdout + assert "error: client-input" in ANSI.sub("", refused.stderr) + assert "argument: kind" in ANSI.sub("", refused.stderr) + + artifacts[str(oracle_transcript)] = oracle_transcript.read_text(encoding="utf-8") + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + + +def test_the_cli_bounded_wait_expires_without_cancelling_the_run( + preview_env: dict[str, Any], tmp_path: Path, evidence_dir: Path +) -> None: + """An expired local wait is a non-zero exit; the remote run keeps going and completes.""" + seed_source_branch(preview_env) + artifacts: dict[str, object] = {} + registered = run_cli( + preview_env, + "configs", + "register", + str(package_file(preview_env, tmp_path)), + "--reason", + REASON, + artifacts=artifacts, + artifact_name="wait timeout configs register", + ) + + # Shorter than the worker's query interval, so the run cannot have finished: what the + # command reports is the wait expiring, never the run failing. + expired = run_cli_command( + preview_env, + "diff", + "--config-id", + registered["config_id"], + "--version", + registered["registry_version"], + "--branch", + SMOKE_BRANCH, + "--reason", + REASON, + "--wait-timeout", + "1", + "--poll-interval", + "1", + artifacts=artifacts, + artifact_name="wait timeout diff", + ) + assert expired.returncode == 1, expired.stdout + assert "error: run-wait-timeout" in ANSI.sub("", expired.stderr) + run_id = fields(expired.stdout)["run_id"] + + oracle_transcript = evidence_dir / "wait-timeout-cli-oracle-http.jsonl" + with authenticated_client(preview_env, transcript=oracle_transcript) as client: + # The bound is local. Nothing was cancelled, so the service still completes it. + planned = wait_for_phase(client, run_id, "planned") + + assert planned["run"]["outcome"] is not None, planned["run"] + artifacts[str(oracle_transcript)] = oracle_transcript.read_text(encoding="utf-8") + assert canary_leaks(preview_env["infrahub_token"], artifacts) == [] + + +def test_the_python_client_verifies_a_plan_and_reads_its_artifact_and_results( + preview_env: dict[str, Any], +) -> None: + """`verify`, `list_artifacts`, `get_artifact` with its digest, and `get_results`.""" + seed_source_branch(preview_env) + + with SyncClient(preview_env["urls"]["sync_api"], preview_env["bearer_token"], timeout=30.0) as client: + registered = client.register_config( + ConfigMutationRequest(package=smoke_package(preview_env["urls"]["infrahub"]), reason=REASON), + idempotency_headers("preview-review")["Idempotency-Key"], + ) + accepted = client.plan( + CreateRunRequest( + operation="plan", + config_id=registered.version.config_id, + registry_version=registered.version.registry_version, + branch=SMOKE_BRANCH, + reason=REASON, + ), + idempotency_headers("preview-review")["Idempotency-Key"], + ) + run_id = accepted.run.run_id + planned = client.wait_for_run(accepted, timeout=WAIT_TIMEOUT_SECONDS, poll_interval=POLL_INTERVAL_SECONDS) + assert planned.run.phase == "planned" + plan = client.get_plan(run_id) + + verified = client.verify_run( + run_id, + VerifyRunRequest(reason=REASON), + idempotency_headers("preview-review")["Idempotency-Key"], + ) + verification_completed = client.wait_for_run( + verified, timeout=WAIT_TIMEOUT_SECONDS, poll_interval=POLL_INTERVAL_SECONDS + ) + + results = client.get_results(run_id) + verification = results.results["verification"] + assert verification["outcome"] == "verified" + assert verification["checksum"] == plan.checksum + assert verification["checksum_ok"] is True + + artifacts = client.list_artifacts(run_id) + assert [reference.artifact_id for reference in artifacts.artifacts] == [PLAN_ARTIFACT_ID] + reference = artifacts.artifacts[0] + content = client.get_artifact(run_id, PLAN_ARTIFACT_ID) + # The client verifies the declared digest itself; recomputing here is what proves + # the bytes it returned are the bytes that digest describes. + assert content.digest == reference.digest + assert sha256(content.data).hexdigest() == reference.digest + assert json.loads(content.data)["checksum"] == plan.checksum + + assert ( + canary_leaks( + preview_env["infrahub_token"], + { + "get_plan resource": plan, + "register_config resource": registered, + "plan accepted resource": accepted, + "planned resource": planned, + "verify accepted resource": verified, + "verification completed resource": verification_completed, + "get_results resource": results, + "list_artifacts resource": artifacts, + "plan-review artifact bytes": content.data, + }, + ) + == [] + ) + + +def test_raw_http_verifies_a_plan_and_reads_its_artifact_and_results( + preview_env: dict[str, Any], evidence_dir: Path +) -> None: + """The same three review resources over the wire, with the exchange captured.""" + seed_source_branch(preview_env) + transcript = evidence_dir / "run-review-http.jsonl" + + with authenticated_client(preview_env, transcript=transcript) as client: + registered = client.post( + "/configs", + headers=idempotency_headers("preview-review"), + json=register_request(preview_env["urls"]["infrahub"]), + ) + assert registered.status_code == 201, registered.text + version = registered.json()["version"] + + created = client.post( + "/runs", + headers=idempotency_headers("preview-review"), + json=create_run_request(version["config_id"], version["registry_version"]), + ) + assert created.status_code == 202, created.text + run_id = created.json()["run"]["run_id"] + wait_for_phase(client, run_id, "planned") + checksum = client.get(f"/runs/{run_id}/plan").json()["checksum"] + + verify = client.post( + f"/runs/{run_id}/verify", + headers=idempotency_headers("preview-review"), + json={"reason": REASON}, + ) + assert verify.status_code == 202, verify.text + verification = _await_verification(client, run_id) + assert verification["outcome"] == "verified" + assert verification["checksum"] == checksum + + listed = client.get(f"/runs/{run_id}/artifacts") + assert listed.status_code == 200, listed.text + reference = listed.json()["artifacts"][0] + assert reference["artifact_id"] == PLAN_ARTIFACT_ID + + artifact = client.get(f"/runs/{run_id}/artifacts/{PLAN_ARTIFACT_ID}") + assert artifact.status_code == 200, artifact.text + # The digest is a response header on this route, and it is what a client without + # the typed helper has to check the bytes against. + assert artifact.headers["Digest"] == f"sha-256={reference['digest']}" + assert sha256(artifact.content).hexdigest() == reference["digest"] + assert json.loads(artifact.content)["checksum"] == checksum + + captured = transcript.read_text(encoding="utf-8") + recorded = {(json.loads(line)["method"], json.loads(line)["path"]) for line in captured.splitlines()} + assert recorded >= { + ("POST", f"/runs/{run_id}/verify"), + ("GET", f"/runs/{run_id}/results"), + ("GET", f"/runs/{run_id}/artifacts"), + ("GET", f"/runs/{run_id}/artifacts/{PLAN_ARTIFACT_ID}"), + } + assert ( + canary_leaks( + preview_env["infrahub_token"], + {str(transcript): captured, "raw plan-review artifact bytes": artifact.content}, + ) + == [] + ) diff --git a/tests/preview/test_schema_drift.py b/tests/preview/test_schema_drift.py new file mode 100644 index 00000000..b2838ab3 --- /dev/null +++ b/tests/preview/test_schema_drift.py @@ -0,0 +1,179 @@ +"""The pre-write gate: an apply refuses a plan the destination schema has moved under. + +A saved plan records the destination schema semantics it was computed against, and the +apply recomputes them from one live read before any adapter is constructed. This module +makes that happen for real — plan, change a consumed attribute's kind on the destination +branch, apply — and reads the refusal back through all three interfaces: the CLI's +rendering, the typed client's recorded apply failure, and the raw results body. + +The change is scoped to the disposable smoke branch and is reversed in a `finally`, with +the original kind read from the running destination first rather than assumed, so a +failure mid-test cannot leave the branch on a schema the other modules do not expect. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +import pytest +import yaml + +from infrahub_sync.client import SyncClient +from tasks.preview import SCHEMA_FILE, SHARED_DEVICE_NAME, SMOKE_BRANCH, SMOKE_KIND +from tests.preview.evidence import canary_leaks +from tests.preview.test_cli_client import ANSI, package_file, run_cli, run_cli_command +from tests.preview.test_run_completion import _await_prefect_terminal_state +from tests.preview.test_service_api import ( + authenticated_client, + device_types, + infrahub_client, + seed_source_branch, +) + +if TYPE_CHECKING: + from pathlib import Path + +pytestmark = pytest.mark.preview + +REASON = "preview qualification: refuse an apply whose destination schema moved" +# The plan maps this attribute, so its kind is part of the semantics the fingerprint +# covers. Text and TextArea both hold the seeded string values, so the conversion is +# reversible in either direction and no device loses its `type`. +DRIFTED_ATTRIBUTE = "type" +REVERSIBLE_KINDS = {"Text": "TextArea", "TextArea": "Text"} + + +def _reversible_kind(original: str) -> str: + """Return the supported temporary counterpart for the live schema kind.""" + assert original in REVERSIBLE_KINDS, f"{DRIFTED_ATTRIBUTE} has unsupported live kind {original!r}" + return REVERSIBLE_KINDS[original] + + +def _attribute_kind(client: Any, attribute: str) -> str: # noqa: ANN401 — the SDK's sync client + """The destination branch's current kind for one attribute of the smoke node.""" + node = client.schema.get(kind=SMOKE_KIND, branch=SMOKE_BRANCH, refresh=True) + return next(declared.kind for declared in node.attributes if declared.name == attribute) + + +def _load_attribute_kind(client: Any, kind: str) -> None: # noqa: ANN401 — the SDK's sync client + """Load the seeded node schema onto the smoke branch with one attribute kind changed. + + Infrahub applies a loaded schema asynchronously, so the load waits for convergence: + the apply under test reads the destination schema live, and an unconverged read would + make the refusal depend on timing rather than on the change. + """ + schema = yaml.safe_load(SCHEMA_FILE.read_text(encoding="utf-8")) + for node in schema["nodes"]: + for attribute in node["attributes"]: + if attribute["name"] == DRIFTED_ATTRIBUTE: + attribute["kind"] = kind + client.schema.load(schemas=[schema], branch=SMOKE_BRANCH, wait_until_converged=True) + assert _attribute_kind(client, DRIFTED_ATTRIBUTE) == kind + + +def test_an_apply_refuses_a_plan_whose_destination_schema_changed( # noqa: PLR0914 + preview_env: dict[str, Any], tmp_path: Path, evidence_dir: Path +) -> None: + """`PlanSchemaChangedError` through CLI rendering, `get_results`, and the raw body.""" + seed_source_branch(preview_env) + sdk = infrahub_client(preview_env) + original_kind = _attribute_kind(sdk, DRIFTED_ATTRIBUTE) + drifted_kind = _reversible_kind(original_kind) + artifacts: dict[str, object] = {} + + registered = run_cli( + preview_env, + "configs", + "register", + str(package_file(preview_env, tmp_path)), + "--reason", + REASON, + artifacts=artifacts, + artifact_name="schema drift configs register", + ) + planned = run_cli( + preview_env, + "diff", + "--config-id", + registered["config_id"], + "--version", + registered["registry_version"], + "--branch", + SMOKE_BRANCH, + "--reason", + REASON, + artifacts=artifacts, + artifact_name="schema drift diff", + ) + run_id, checksum = planned["run_id"], planned["plan_checksum"] + assert planned["schema_fingerprint"], planned + before = device_types(sdk, SMOKE_BRANCH)[SHARED_DEVICE_NAME] + + try: + _load_attribute_kind(sdk, drifted_kind) + refused = run_cli_command( + preview_env, + "apply", + run_id, + "--expected-checksum", + checksum, + "--branch", + SMOKE_BRANCH, + "--reason", + REASON, + artifacts=artifacts, + artifact_name="schema drift apply refusal", + ) + finally: + _load_attribute_kind(sdk, original_kind) + + assert refused.returncode == 1, refused.stdout + rendered = ANSI.sub("", refused.stderr) + assert "apply failed: PlanSchemaChangedError" in rendered, rendered + assert "hint: create and review a new plan before applying again" in rendered, rendered + + with SyncClient(preview_env["urls"]["sync_api"], preview_env["bearer_token"], timeout=30.0) as client: + results = client.get_results(run_id) + failure = results.results["apply_failure"] + assert failure["stage"] == "apply" + assert failure["error_type"] == "PlanSchemaChangedError" + # The gate runs before any adapter is constructed, so a refusal cannot have written. + assert failure.get("may_have_partially_written") in {None, False}, failure + + transcript = evidence_dir / "schema-drift-results-http.jsonl" + with authenticated_client(preview_env, transcript=transcript) as api: + recorded = api.get(f"/runs/{run_id}") + assert recorded.status_code == 200, recorded.text + assert recorded.json()["run"]["phase"] == "apply-failed", recorded.text + body = api.get(f"/runs/{run_id}/results") + assert body.status_code == 200, body.text + assert body.json()["results"]["apply_failure"]["error_type"] == "PlanSchemaChangedError", body.text + flow_run_id = recorded.json()["orchestration"][-1]["flow_run_id"] + + assert _await_prefect_terminal_state(preview_env, flow_run_id) == "FAILED" + + assert device_types(sdk, SMOKE_BRANCH)[SHARED_DEVICE_NAME] == before + assert _attribute_kind(sdk, DRIFTED_ATTRIBUTE) == original_kind + captured = transcript.read_text(encoding="utf-8") + exchanges = { + (record["method"], record["path"], record["status"]) for record in map(json.loads, captured.splitlines()) + } + assert exchanges >= { + ("GET", f"/runs/{run_id}", 200), + ("GET", f"/runs/{run_id}/results", 200), + } + artifacts.update( + { + "schema drift get_results resource": results, + str(transcript): captured, + "schema drift raw results body": body.content, + } + ) + assert ( + canary_leaks( + preview_env["infrahub_token"], + artifacts, + ) + == [] + ) diff --git a/tests/preview/test_service_api.py b/tests/preview/test_service_api.py index 736505c3..fb4f529b 100644 --- a/tests/preview/test_service_api.py +++ b/tests/preview/test_service_api.py @@ -30,15 +30,20 @@ from __future__ import annotations +import json import time import uuid from collections.abc import Iterable, Mapping -from typing import Any +from typing import TYPE_CHECKING, Any import httpx import pytest from tasks.preview import SHARED_DEVICE_NAME, SMOKE_BRANCH, SMOKE_KIND +from tests.preview.evidence import canary_leaks, transcript_hooks + +if TYPE_CHECKING: + from pathlib import Path pytestmark = pytest.mark.preview @@ -140,12 +145,19 @@ def _client(preview_env: dict[str, Any], token: str | None) -> httpx.Client: return httpx.Client(base_url=preview_env["urls"]["sync_api"], headers=headers, timeout=30) -def _idempotency() -> dict[str, str]: +def authenticated_client(preview_env: dict[str, Any], *, transcript: Path) -> httpx.Client: + """The bearer-authenticated raw client, recording every exchange.""" + client = _client(preview_env, preview_env["bearer_token"]) + client.event_hooks = transcript_hooks(transcript) + return client + + +def idempotency_headers(prefix: str = "preview-smoke") -> dict[str, str]: """A fresh key per mutation, so a re-run never replays an earlier smoke's response.""" - return {"Idempotency-Key": f"preview-smoke-{uuid.uuid4()}"} + return {"Idempotency-Key": f"{prefix}-{uuid.uuid4()}"} -def _wait_for_phase(client: httpx.Client, run_id: str, target_phase: str) -> dict[str, Any]: +def wait_for_phase(client: httpx.Client, run_id: str, target_phase: str) -> dict[str, Any]: """Poll until the durable record reaches the target phase. Polling ``finished_at`` is not enough: an admitted apply continues the @@ -169,7 +181,9 @@ def _wait_for_phase(client: httpx.Client, run_id: str, target_phase: str) -> dic def _registered_version(client: httpx.Client, preview_env: dict[str, Any]) -> tuple[str, int]: """Register the smoke package and prove the returned version validates cleanly.""" - registered = client.post("/configs", headers=_idempotency(), json=register_request(preview_env["urls"]["infrahub"])) + registered = client.post( + "/configs", headers=idempotency_headers(), json=register_request(preview_env["urls"]["infrahub"]) + ) assert registered.status_code == 201, registered.text version = registered.json()["version"] config_id, registry_version = version["config_id"], version["registry_version"] @@ -258,18 +272,21 @@ def test_requests_without_a_bearer_token_are_refused(preview_env: dict[str, Any] assert response.status_code == 401 -def test_service_plan_and_apply_lifecycle(preview_env: dict[str, Any]) -> None: +def test_service_plan_and_apply_lifecycle(preview_env: dict[str, Any], evidence_dir: Path) -> None: # noqa: PLR0914 — the row scans every direct-HTTP artifact mutated_type = seed_source_branch(preview_env) assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] != mutated_type - with _client(preview_env, token=preview_env["bearer_token"]) as client: + transcript = evidence_dir / "service-lifecycle-http.jsonl" + with authenticated_client(preview_env, transcript=transcript) as client: config_id, registry_version = _registered_version(client, preview_env) - created = client.post("/runs", headers=_idempotency(), json=create_run_request(config_id, registry_version)) + created = client.post( + "/runs", headers=idempotency_headers(), json=create_run_request(config_id, registry_version) + ) assert created.status_code == 202, created.text run_id = created.json()["run"]["run_id"] - planned = _wait_for_phase(client, run_id, "planned") + planned = wait_for_phase(client, run_id, "planned") assert planned["run"]["outcome"] is not None, planned["run"] plan_view = client.get(f"/runs/{run_id}/plan") @@ -292,10 +309,12 @@ def test_service_plan_and_apply_lifecycle(preview_env: dict[str, Any]) -> None: assert summary.get("deletes_not_executed", 0) == 0, summary checksum = plan_payload["checksum"] - apply_accepted = client.post(f"/runs/{run_id}/apply", headers=_idempotency(), json=apply_run_request(checksum)) + apply_accepted = client.post( + f"/runs/{run_id}/apply", headers=idempotency_headers(), json=apply_run_request(checksum) + ) assert apply_accepted.status_code == 202, apply_accepted.text - applied = _wait_for_phase(client, run_id, "applied") + applied = wait_for_phase(client, run_id, "applied") assert applied["run"]["outcome"] is not None, applied["run"] # What the apply actually did, not merely that it finished. applied_summary = applied["run"]["summary"] @@ -309,5 +328,23 @@ def test_service_plan_and_apply_lifecycle(preview_env: dict[str, Any]) -> None: # the recorded action counts are the positive-applied evidence. assert results.json()["results"]["summary"]["update"] > 0, results.text + captured = transcript.read_text(encoding="utf-8") + exchanges = {(entry["method"], entry["path"], entry["status"]) for entry in map(json.loads, captured.splitlines())} + assert exchanges >= { + ("POST", "/configs", 201), + ("POST", f"/configs/{config_id}/versions/{registry_version}/validate", 200), + ("POST", "/runs", 202), + ("GET", f"/runs/{run_id}/plan", 200), + ("POST", f"/runs/{run_id}/apply", 202), + ("GET", f"/runs/{run_id}/results", 200), + } + assert ( + canary_leaks( + preview_env["infrahub_token"], + {str(transcript): captured, "service lifecycle results body": results.content}, + ) + == [] + ) + # The destination now carries the value the source was mutated to. assert device_types(infrahub_client(preview_env), SMOKE_BRANCH)[SHARED_DEVICE_NAME] == mutated_type