From a4f86d7ba9d9c448e5f43181a0df80c8bf41e484 Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:18:50 +0200 Subject: [PATCH 1/4] Accept a command health check in the component schema (#170) A component can declare healthCheck type: command (command argv, timeoutSeconds 1-300, attempts 1-20, no other key). The SDK parses it into value objects and the deploy, restart and rollback plans name the check and its command in wouldRun; the plan JSON gains no field (S-009-1, S-009-13). The engine does not run the check yet (#164), so the deploy role's entry points refuse a command check on their first task instead of passing an unrun gate. Fixes #161 Co-authored-by: shipmill-romamo[bot] <338557544+shipmill-romamo[bot]@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 --- .../161-command-health-check-schema.md | 3 + config/schemas/v1/component.schema.json | 34 ++ .../cloudfall_deploy/meta/argument_specs.yml | 4 + .../roles/cloudfall_deploy/tasks/health.yml | 10 + .../roles/cloudfall_deploy/tasks/main.yml | 10 + .../roles/cloudfall_deploy/tasks/restart.yml | 10 + .../roles/cloudfall_deploy/tasks/rollback.yml | 10 + engine/tests/test_ansible_inventory.py | 29 ++ engine/tests/test_deploy_role.py | 60 +++ sdk/src/cloudfall/domain.py | 59 +++ sdk/src/cloudfall/inventory.py | 42 +- sdk/src/cloudfall/lifecycle.py | 15 +- sdk/tests/test_command_health_check.py | 365 ++++++++++++++++++ 13 files changed, 649 insertions(+), 2 deletions(-) create mode 100644 changelog.d/161-command-health-check-schema.md create mode 100644 sdk/tests/test_command_health_check.py diff --git a/changelog.d/161-command-health-check-schema.md b/changelog.d/161-command-health-check-schema.md new file mode 100644 index 0000000..a31cc82 --- /dev/null +++ b/changelog.d/161-command-health-check-schema.md @@ -0,0 +1,3 @@ +### Added + +- Components: a `healthCheck` of `type: command` (a required non-empty `command` argv, `timeoutSeconds` 1 to 300 and `attempts` 1 to 20, no other key) validates, and `deploy`, `restart` and `rollback` without `--yes` name the check and its command in their plan's `wouldRun`. The engine does not run the check yet: until it does, `deploy`, `restart`, `rollback` and `health` of such a component fail on their first task, before anything changes on the server, instead of passing an unrun gate (#161) diff --git a/config/schemas/v1/component.schema.json b/config/schemas/v1/component.schema.json index 65d6fbe..6fe6505 100644 --- a/config/schemas/v1/component.schema.json +++ b/config/schemas/v1/component.schema.json @@ -218,6 +218,40 @@ "type" ], "additionalProperties": false + }, + { + "type": "object", + "properties": { + "type": { + "const": "command" + }, + "command": { + "type": "array", + "items": { + "type": "string", + "minLength": 1, + "pattern": "^[^\\u0000]+$" + }, + "minItems": 1 + }, + "timeoutSeconds": { + "type": "integer", + "minimum": 1, + "maximum": 300 + }, + "attempts": { + "type": "integer", + "minimum": 1, + "maximum": 20 + } + }, + "required": [ + "type", + "command", + "timeoutSeconds", + "attempts" + ], + "additionalProperties": false } ] }, diff --git a/engine/ansible/roles/cloudfall_deploy/meta/argument_specs.yml b/engine/ansible/roles/cloudfall_deploy/meta/argument_specs.yml index 27b90fd..3816e06 100644 --- a/engine/ansible/roles/cloudfall_deploy/meta/argument_specs.yml +++ b/engine/ansible/roles/cloudfall_deploy/meta/argument_specs.yml @@ -60,7 +60,11 @@ argument_specs: choices: - http - none + - command required: true + command: + type: list + elements: str scheme: type: str choices: diff --git a/engine/ansible/roles/cloudfall_deploy/tasks/health.yml b/engine/ansible/roles/cloudfall_deploy/tasks/health.yml index 7a8b453..8c15bf8 100644 --- a/engine/ansible/roles/cloudfall_deploy/tasks/health.yml +++ b/engine/ansible/roles/cloudfall_deploy/tasks/health.yml @@ -1,3 +1,13 @@ --- +# Until cloudfall-dev/cloudfall#164 runs it, a command health check fails here +# rather than reporting healthy without having run. +- name: Refuse a command health check this engine does not run yet + ansible.builtin.fail: + msg: >- + Component {{ cloudfall_deploy_component.id }} declares a command health + check, which this engine does not run yet + (cloudfall-dev/cloudfall#164); nothing was changed + when: cloudfall_deploy_component.healthCheck.type == "command" + - name: Probe the component's declared health check ansible.builtin.include_tasks: health_gate.yml diff --git a/engine/ansible/roles/cloudfall_deploy/tasks/main.yml b/engine/ansible/roles/cloudfall_deploy/tasks/main.yml index c4e5381..1efdfc9 100644 --- a/engine/ansible/roles/cloudfall_deploy/tasks/main.yml +++ b/engine/ansible/roles/cloudfall_deploy/tasks/main.yml @@ -1,4 +1,14 @@ --- +# Until cloudfall-dev/cloudfall#164 runs it, a command health check fails here, +# before anything changes, rather than letting the deploy pass an unrun gate. +- name: Refuse a command health check this engine does not run yet + ansible.builtin.fail: + msg: >- + Component {{ cloudfall_deploy_component.id }} declares a command health + check, which this engine does not run yet + (cloudfall-dev/cloudfall#164); nothing was changed + when: cloudfall_deploy_component.healthCheck.type == "command" + - name: Validate deployment target and declaration ansible.builtin.assert: that: diff --git a/engine/ansible/roles/cloudfall_deploy/tasks/restart.yml b/engine/ansible/roles/cloudfall_deploy/tasks/restart.yml index 89b9829..f72ae81 100644 --- a/engine/ansible/roles/cloudfall_deploy/tasks/restart.yml +++ b/engine/ansible/roles/cloudfall_deploy/tasks/restart.yml @@ -1,4 +1,14 @@ --- +# Until cloudfall-dev/cloudfall#164 runs it, a command health check fails here, +# before anything changes, rather than letting the restart pass an unrun gate. +- name: Refuse a command health check this engine does not run yet + ansible.builtin.fail: + msg: >- + Component {{ cloudfall_deploy_component.id }} declares a command health + check, which this engine does not run yet + (cloudfall-dev/cloudfall#164); nothing was changed + when: cloudfall_deploy_component.healthCheck.type == "command" + - name: Validate restart target and declaration ansible.builtin.assert: that: diff --git a/engine/ansible/roles/cloudfall_deploy/tasks/rollback.yml b/engine/ansible/roles/cloudfall_deploy/tasks/rollback.yml index fc7deb4..d17bbff 100644 --- a/engine/ansible/roles/cloudfall_deploy/tasks/rollback.yml +++ b/engine/ansible/roles/cloudfall_deploy/tasks/rollback.yml @@ -1,4 +1,14 @@ --- +# Until cloudfall-dev/cloudfall#164 runs it, a command health check fails here, +# before anything changes, rather than letting the rollback pass an unrun gate. +- name: Refuse a command health check this engine does not run yet + ansible.builtin.fail: + msg: >- + Component {{ cloudfall_deploy_component.id }} declares a command health + check, which this engine does not run yet + (cloudfall-dev/cloudfall#164); nothing was changed + when: cloudfall_deploy_component.healthCheck.type == "command" + - name: Validate rollback target and declaration ansible.builtin.assert: that: diff --git a/engine/tests/test_ansible_inventory.py b/engine/tests/test_ansible_inventory.py index 722a544..f250cbd 100644 --- a/engine/tests/test_ansible_inventory.py +++ b/engine/tests/test_ansible_inventory.py @@ -311,6 +311,35 @@ def test_s005_6_a_release_renders_with_its_release_server(tmp_path: Path) -> Non } +def test_s009_1_a_command_health_check_renders_as_declared(tmp_path: Path) -> None: + project = tmp_path / "project" + shutil.copytree(EXAMPLES, project) + path = project / "components" / "crm-backend.yaml" + text = path.read_text(encoding="utf-8") + declared = text[text.index(" healthCheck:\n") : text.index(" secretRefs:\n")] + check = { + "type": "command", + "command": [".venv/bin/python", "-m", "app.smoke"], + "timeoutSeconds": 60, + "attempts": 3, + } + path.write_text( + text.replace(declared, f" healthCheck: {json.dumps(check)}\n"), + encoding="utf-8", + ) + rendered = render_ansible_inventory( + PlatformInventory.from_state(validate_config(project, SCHEMAS)) + ) + hosts = _mapping_entry( + _mapping_entry(_inventory_children(rendered), "cloudfall_servers"), "hosts" + ) + + for host in ("h1", "h2"): + components = _mapping_entry(hosts, host)["cloudfall_components"] + assert isinstance(components, list) + assert components[0]["healthCheck"] == check + + def test_engine_cli_emits_ansible_json( capsys: pytest.CaptureFixture[str], ) -> None: diff --git a/engine/tests/test_deploy_role.py b/engine/tests/test_deploy_role.py index fda5b8a..242da59 100644 --- a/engine/tests/test_deploy_role.py +++ b/engine/tests/test_deploy_role.py @@ -12,6 +12,7 @@ from pathlib import Path import pytest +from ansible.module_utils.common.arg_spec import ArgumentSpecValidator from ansible.parsing.dataloader import DataLoader from ansible.template import Templar, trust_as_template @@ -479,3 +480,62 @@ def test_the_failure_of_a_first_release_without_a_kept_unit_is_unchanged() -> No "Release r2 failed its health check and was stopped because no previous" " release exists" ) + + +COMMAND_CHECK: dict[str, object] = { + "type": "command", + "command": [".venv/bin/python", "-m", "app.smoke"], + "timeoutSeconds": 60, + "attempts": 3, +} +REFUSE_COMMAND = "Refuse a command health check this engine does not run yet" + + +@pytest.mark.parametrize( + "entry_point", ["main.yml", "restart.yml", "rollback.yml", "health.yml"] +) +def test_the_engine_refuses_a_command_check_it_does_not_run_yet( + entry_point: str, +) -> None: + """Until #164 runs it, a command check fails first, never passes unrun.""" + tasks = _load(ROLE / "tasks" / entry_point) + assert isinstance(tasks, list) + refusal = tasks[0] + assert refusal["name"] == REFUSE_COMMAND + fail = refusal["ansible.builtin.fail"] + assert isinstance(fail, dict) + + def variables(health_check: dict[str, object]) -> dict[str, object]: + return { + "cloudfall_deploy_component": { + "id": "crm-backend", + "healthCheck": health_check, + } + } + + assert all(_holds(c, variables(COMMAND_CHECK)) for c in _conditions(refusal)) + for other in (HTTP_CHECK, NO_CHECK): + assert not all(_holds(c, variables(other)) for c in _conditions(refusal)) + message = str( + _templar(variables(COMMAND_CHECK)).template(trust_as_template(fail["msg"])) + ) + assert message.startswith( + "Component crm-backend declares a command health check" + ) + assert "cloudfall-dev/cloudfall#164" in message + + +@pytest.mark.parametrize("health_check", [HTTP_CHECK, NO_CHECK, COMMAND_CHECK]) +def test_s009_1_the_role_argument_spec_accepts_every_declared_check( + health_check: dict[str, object], +) -> None: + specs = _load(ROLE / "meta" / "argument_specs.yml") + assert isinstance(specs, dict) + component_spec = specs["argument_specs"]["main"]["options"][ + "cloudfall_deploy_component" + ]["options"]["healthCheck"] + validator = ArgumentSpecValidator(component_spec["options"]) + + result = validator.validate(dict(health_check)) + + assert result.error_messages == [] diff --git a/sdk/src/cloudfall/domain.py b/sdk/src/cloudfall/domain.py index 10d30eb..42e3b17 100644 --- a/sdk/src/cloudfall/domain.py +++ b/sdk/src/cloudfall/domain.py @@ -621,6 +621,7 @@ class HealthCheckType(StrEnum): HTTP = "http" NONE = "none" + COMMAND = "command" @classmethod def from_boundary(cls, value: object) -> HealthCheckType: @@ -628,6 +629,64 @@ def from_boundary(cls, value: object) -> HealthCheckType: return cls(_required_string(value, "health check type")) +_HEALTH_CHECK_TIMEOUT_MAXIMUM = 300 +_HEALTH_CHECK_ATTEMPTS_MAXIMUM = 20 + + +def _required_bounded_integer(value: object, label: str, maximum: int) -> int: + if isinstance(value, bool) or not isinstance(value, int): + message = f"{label} must be an integer, got {type(value).__name__}" + raise TypeError(message) + if not 1 <= value <= maximum: + message = f"{label} must be 1 to {maximum}, got {value}" + raise ValueError(message) + return value + + +@dataclass(frozen=True, slots=True) +class HealthCheckTimeout: + """How long one run of a command health check may take: 1 to 300 seconds.""" + + value: int + + def __post_init__(self) -> None: + """Reject booleans and values outside 1..300.""" + _required_bounded_integer( + self.value, "health check timeout", _HEALTH_CHECK_TIMEOUT_MAXIMUM + ) + + @classmethod + def from_boundary(cls, value: object) -> HealthCheckTimeout: + """Coerce a required boundary value into a health-check timeout.""" + return cls( + _required_bounded_integer( + value, "health check timeout", _HEALTH_CHECK_TIMEOUT_MAXIMUM + ) + ) + + +@dataclass(frozen=True, slots=True) +class HealthCheckAttempts: + """How many times a command health check runs before it fails: 1 to 20.""" + + value: int + + def __post_init__(self) -> None: + """Reject booleans and values outside 1..20.""" + _required_bounded_integer( + self.value, "health check attempts", _HEALTH_CHECK_ATTEMPTS_MAXIMUM + ) + + @classmethod + def from_boundary(cls, value: object) -> HealthCheckAttempts: + """Coerce a required boundary value into a health-check attempt count.""" + return cls( + _required_bounded_integer( + value, "health check attempts", _HEALTH_CHECK_ATTEMPTS_MAXIMUM + ) + ) + + class RuntimeType(StrEnum): """Component runtimes supported by the v1 contract.""" diff --git a/sdk/src/cloudfall/inventory.py b/sdk/src/cloudfall/inventory.py index 038680e..4ecf56b 100644 --- a/sdk/src/cloudfall/inventory.py +++ b/sdk/src/cloudfall/inventory.py @@ -27,6 +27,8 @@ FileMode, FilesystemName, FirewallPolicy, + HealthCheckAttempts, + HealthCheckTimeout, HealthCheckType, Hostname, HttpScheme, @@ -477,22 +479,48 @@ class ComponentHttpHealthCheck: attempts: PositiveCount +@dataclass(frozen=True, slots=True) +class ComponentCommandHealthCheck: + """Command contract for a component health gate: an argv run on the server.""" + + command: CommandArgv + timeout_seconds: HealthCheckTimeout + attempts: HealthCheckAttempts + + @property + def described(self) -> str: + """The command as one line, for a plan's ``wouldRun``.""" + return " ".join(self.command.value) + + @dataclass(frozen=True, slots=True) class ComponentHealthCheck: """Health gate for a deployed component.""" check_type: HealthCheckType http: ComponentHttpHealthCheck | None + command: ComponentCommandHealthCheck | None = None def __post_init__(self) -> None: - """Require the HTTP contract exactly for HTTP health checks.""" + """Require each type's contract exactly for checks of that type.""" if (self.check_type is HealthCheckType.HTTP) != (self.http is not None): message = "HTTP health checks require an HTTP contract" raise ValueError(message) + if (self.check_type is HealthCheckType.COMMAND) != (self.command is not None): + message = "command health checks require a command contract" + raise ValueError(message) def as_dict(self) -> dict[str, object]: """Serialize the health contract.""" result: dict[str, object] = {"type": self.check_type.value} + if self.command is not None: + result.update( + { + "command": list(self.command.command.value), + "timeoutSeconds": self.command.timeout_seconds.value, + "attempts": self.command.attempts.value, + } + ) if self.http is not None: result.update( { @@ -1734,6 +1762,18 @@ def _component_health_check( check_type = HealthCheckType.from_boundary(health.get("type")) if check_type is HealthCheckType.NONE: return ComponentHealthCheck(check_type=check_type, http=None) + if check_type is HealthCheckType.COMMAND: + return ComponentHealthCheck( + check_type=check_type, + http=None, + command=ComponentCommandHealthCheck( + command=CommandArgv.from_boundary(health.get("command")), + timeout_seconds=HealthCheckTimeout.from_boundary( + health.get("timeoutSeconds") + ), + attempts=HealthCheckAttempts.from_boundary(health.get("attempts")), + ), + ) return ComponentHealthCheck( check_type=check_type, http=ComponentHttpHealthCheck( diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index 126a5a3..d3a0498 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -1143,6 +1143,7 @@ def preview_deploy( f"once on {release_command.server}, before any server switches " "to the release" ) + effect += _command_health_check_effect(component) return LifecyclePreview( action="deploy", subject=PreviewSubject.COMPONENT, @@ -1174,6 +1175,7 @@ def preview_rollback( "; its release command " f"{' '.join(component.release.command.value)} is not run" ) + effect += _command_health_check_effect(component) return LifecyclePreview( action="rollback", subject=PreviewSubject.COMPONENT, @@ -1199,11 +1201,22 @@ def preview_restart( effect=( f"restart component {component_id} on " f"{_server_list(component.server_ids)} and require its declared " - "health check" + "health check" + _command_health_check_effect(component) ), ) +def _command_health_check_effect(component: ComponentInventory) -> str: + """Name a component's command health check for a plan's ``wouldRun``.""" + command = component.health_check.command + if command is None: + return "" + return ( + f"; its command health check {command.described} runs on each server " + "after its restart" + ) + + def rollback( context: EngineContext, component_id: ResourceId, diff --git a/sdk/tests/test_command_health_check.py b/sdk/tests/test_command_health_check.py new file mode 100644 index 0000000..f91731a --- /dev/null +++ b/sdk/tests/test_command_health_check.py @@ -0,0 +1,365 @@ +"""A component's ``healthCheck`` of ``type: command`` (S-009-1, S-009-13). + +The config contract and the plans only: the engine running the check is +cloudfall-dev/cloudfall#164. The engine directory the CLI is pointed at is a +stand-in whose playbooks only leave a marker file, so a test sees whether +``ansible-playbook`` ran at all without contacting any host. +""" + +from __future__ import annotations + +import hashlib +import json +import shutil +import tarfile +from pathlib import Path +from typing import Any + +import pytest +from cloudfall.cli import main +from cloudfall.domain import ( + CommandArgv, + HealthCheckAttempts, + HealthCheckTimeout, + HealthCheckType, +) +from cloudfall.inventory import ( + ComponentCommandHealthCheck, + ComponentHealthCheck, + PlatformInventory, +) +from cloudfall.validation import validate_config + +ROOT = Path(__file__).parents[2] +SCHEMAS = ROOT / "config" / "schemas" / "v1" +EXAMPLES = ROOT / "config" / "examples" +ENGINE = ROOT / "engine" +RELEASE = "20260101T000000Z-abcdef0" +DECLARED_HTTP_CHECK = ( + " healthCheck:\n" + " type: http\n" + " scheme: http\n" + " port: 8100\n" + " path: /health\n" + " expectedStatuses:\n" + " - 200\n" + " timeoutSeconds: 5\n" + " attempts: 5\n" +) +COMMAND_CHECK: dict[str, object] = { + "type": "command", + "command": [".venv/bin/python", "-m", "app.smoke"], + "timeoutSeconds": 60, + "attempts": 3, +} + + +def _project(tmp_path: Path, health_check: object) -> Path: + """Copy the examples with crm-backend's ``healthCheck`` replaced.""" + project = tmp_path / "config" + shutil.copytree(EXAMPLES, project) + path = project / "components" / "crm-backend.yaml" + text = path.read_text(encoding="utf-8") + assert DECLARED_HTTP_CHECK in text + path.write_text( + text.replace( + DECLARED_HTTP_CHECK, f" healthCheck: {json.dumps(health_check)}\n" + ), + encoding="utf-8", + ) + return project + + +def _validate(project: Path, capsys: pytest.CaptureFixture[str]) -> tuple[int, Any]: + exit_code = main( + ["config", "validate", "--project", str(project), "--schemas", str(SCHEMAS)] + ) + return exit_code, json.loads(capsys.readouterr().out) + + +def test_s009_1_config_validate_accepts_a_command_health_check( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + exit_code, payload = _validate(_project(tmp_path, COMMAND_CHECK), capsys) + + assert exit_code == 0, payload + assert payload["ok"] is True + + +@pytest.mark.parametrize( + "health_check", + [ + { + **COMMAND_CHECK, + "command": ["/usr/bin/true"], + "timeoutSeconds": 1, + "attempts": 1, + }, + {**COMMAND_CHECK, "timeoutSeconds": 300, "attempts": 20}, + ], + ids=["lower-bounds", "upper-bounds"], +) +def test_s009_1_config_validate_accepts_the_bounds( + tmp_path: Path, capsys: pytest.CaptureFixture[str], health_check: object +) -> None: + exit_code, payload = _validate(_project(tmp_path, health_check), capsys) + + assert exit_code == 0, payload + + +def _without(key: str) -> dict[str, object]: + return {name: value for name, value in COMMAND_CHECK.items() if name != key} + + +@pytest.mark.parametrize( + "health_check", + [ + {**COMMAND_CHECK, "command": []}, + {**COMMAND_CHECK, "command": [""]}, + {**COMMAND_CHECK, "command": [".venv/bin/python", ""]}, + {**COMMAND_CHECK, "command": "app.smoke"}, + {**COMMAND_CHECK, "command": ["bad\u0000arg"]}, + _without("command"), + _without("timeoutSeconds"), + _without("attempts"), + {**COMMAND_CHECK, "timeoutSeconds": 0}, + {**COMMAND_CHECK, "timeoutSeconds": 301}, + {**COMMAND_CHECK, "timeoutSeconds": "60"}, + {**COMMAND_CHECK, "attempts": 0}, + {**COMMAND_CHECK, "attempts": 21}, + {**COMMAND_CHECK, "attempts": True}, + {**COMMAND_CHECK, "port": 8100}, + {**COMMAND_CHECK, "scheme": "http"}, + {**COMMAND_CHECK, "path": "/health"}, + {**COMMAND_CHECK, "expectedStatuses": [200]}, + {**COMMAND_CHECK, "workingDirectory": "services/api"}, + ], + ids=[ + "empty-command", + "empty-argument", + "empty-later-argument", + "string-command", + "nul-in-argument", + "missing-command", + "missing-timeout", + "missing-attempts", + "timeout-0", + "timeout-301", + "timeout-string", + "attempts-0", + "attempts-21", + "attempts-boolean", + "http-port", + "http-scheme", + "http-path", + "http-expected-statuses", + "unknown-key", + ], +) +def test_s009_1_config_validate_refuses_a_malformed_command_check( + tmp_path: Path, capsys: pytest.CaptureFixture[str], health_check: object +) -> None: + exit_code, payload = _validate(_project(tmp_path, health_check), capsys) + + assert exit_code == 80, payload + assert payload["error"]["code"] == "CONFIG_INVALID" + assert payload["error"]["context"]["code"] == "schema_validation_failed" + + +def test_s009_1_the_command_check_parses_into_value_objects(tmp_path: Path) -> None: + inventory = PlatformInventory.from_state( + validate_config(_project(tmp_path, COMMAND_CHECK), SCHEMAS) + ) + health_check = inventory.components[0].health_check + + assert health_check == ComponentHealthCheck( + check_type=HealthCheckType.COMMAND, + http=None, + command=ComponentCommandHealthCheck( + command=CommandArgv((".venv/bin/python", "-m", "app.smoke")), + timeout_seconds=HealthCheckTimeout(60), + attempts=HealthCheckAttempts(3), + ), + ) + assert health_check.as_dict() == COMMAND_CHECK + + +@pytest.mark.parametrize("value", [0, 301, True]) +def test_s009_1_the_timeout_refuses_values_outside_1_to_300(value: object) -> None: + with pytest.raises((ValueError, TypeError)): + HealthCheckTimeout.from_boundary(value) + + +@pytest.mark.parametrize("value", [0, 21, True, None]) +def test_s009_1_attempts_refuse_values_outside_1_to_20(value: object) -> None: + with pytest.raises((ValueError, TypeError)): + HealthCheckAttempts.from_boundary(value) + + +def test_s009_1_a_check_carries_exactly_the_contract_of_its_type() -> None: + command = ComponentCommandHealthCheck( + command=CommandArgv(("/usr/bin/true",)), + timeout_seconds=HealthCheckTimeout(1), + attempts=HealthCheckAttempts(1), + ) + with pytest.raises(ValueError, match="command contract"): + ComponentHealthCheck(check_type=HealthCheckType.COMMAND, http=None) + with pytest.raises(ValueError, match="command contract"): + ComponentHealthCheck( + check_type=HealthCheckType.NONE, http=None, command=command + ) + + +def test_s009_1_http_and_none_checks_keep_their_shape(tmp_path: Path) -> None: + """Q-3: the existing checks parse and serialize exactly as before.""" + examples = PlatformInventory.from_state(validate_config(EXAMPLES, SCHEMAS)) + http = examples.components[0].health_check + none = PlatformInventory.from_state( + validate_config(_project(tmp_path, {"type": "none"}), SCHEMAS) + ).components[0].health_check + + assert http.command is None + assert http.as_dict() == { + "type": "http", + "scheme": "http", + "port": 8100, + "path": "/health", + "expectedStatuses": [200], + "timeoutSeconds": 5, + "attempts": 5, + } + assert none.command is None + assert none.as_dict() == {"type": "none"} + + +def _marker_engine(tmp_path: Path) -> Path: + """Lay out an engine whose playbooks only write a marker.""" + engine = tmp_path / "engine" + playbooks = engine / "ansible" / "playbooks" + playbooks.mkdir(parents=True) + shutil.copy(ENGINE / "ansible" / "ansible.cfg", engine / "ansible") + for name in ("deploy.yml", "rollback.yml", "restart.yml"): + marker = tmp_path / f"ran-{name}" + (playbooks / name).write_text( + "---\n- name: Leave a marker\n hosts: localhost\n" + " connection: local\n gather_facts: false\n tasks:\n" + " - name: Write the marker\n ansible.builtin.copy:\n" + f" dest: {marker}\n content: ran\n mode: '0600'\n", + encoding="utf-8", + ) + return engine + + +def _write_artifact(tmp_path: Path) -> Path: + artifact_directory = tmp_path / "artifacts" + component_directory = artifact_directory / "crm-backend" + component_directory.mkdir(parents=True) + source = tmp_path / "source" + source.mkdir() + (source / "app.py").write_text("print('crm')\n", encoding="utf-8") + archive_path = component_directory / f"{RELEASE}.tar.gz" + with tarfile.open(archive_path, "w:gz") as archive: + archive.add(source, arcname=".") + metadata = { + "apiVersion": "cloudfall/v1", + "kind": "Artifact", + "metadata": {"id": "crm-backend"}, + "spec": { + "component": "crm-backend", + "release": RELEASE, + "gitRef": "main", + "gitCommit": "a" * 40, + "builtAt": "2026-01-01T00:00:00Z", + "archive": archive_path.name, + "archiveSha256": hashlib.sha256(archive_path.read_bytes()).hexdigest(), + "sizeBytes": archive_path.stat().st_size, + }, + } + (component_directory / f"{RELEASE}.json").write_text( + json.dumps(metadata), encoding="utf-8" + ) + return artifact_directory + + +def _plan( + action: str, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + health_check: object, +) -> dict[str, Any]: + arguments = { + "deploy": [ + "--release", + RELEASE, + "--artifacts", + str(_write_artifact(tmp_path)), + ], + "rollback": ["--release", RELEASE], + "restart": [], + }[action] + exit_code = main( + [ + action, + "crm-backend", + *arguments, + "--project", + str(_project(tmp_path, health_check)), + "--schemas", + str(SCHEMAS), + "--engine", + str(_marker_engine(tmp_path)), + "--inventory-file", + str(tmp_path / "inventory.json"), + ] + ) + payload = json.loads(capsys.readouterr().out) + assert exit_code == 0, payload + assert not any(tmp_path.glob("ran-*")) + assert not (tmp_path / "inventory.json").exists() + data = payload["data"] + assert isinstance(data, dict) + assert data["status"] == "plan" + return data + + +@pytest.mark.parametrize("action", ["deploy", "rollback", "restart"]) +def test_s009_13_the_plan_names_the_command_health_check( + action: str, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + data = _plan(action, tmp_path, capsys, COMMAND_CHECK) + + assert ( + "command health check .venv/bin/python -m app.smoke runs on each server " + "after its restart" + ) in data["would_run"] + + +@pytest.mark.parametrize("action", ["deploy", "rollback", "restart"]) +@pytest.mark.parametrize( + "health_check", + [ + { + "type": "http", + "scheme": "http", + "port": 8100, + "path": "/health", + "expectedStatuses": [200], + "timeoutSeconds": 5, + "attempts": 5, + }, + {"type": "none"}, + ], + ids=["http", "none"], +) +def test_s009_13_the_plan_gains_no_field( + action: str, + health_check: object, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """The command check changes only ``would_run``; other checks change nothing.""" + other = _plan(action, tmp_path / "other", capsys, health_check) + command = _plan(action, tmp_path / "command", capsys, COMMAND_CHECK) + + assert set(command) == set(other) + assert "command health check" not in other["would_run"] From de171142d3f639708a2c4ab3e207399045bd7fa1 Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:23:26 +0200 Subject: [PATCH 2/4] Refuse a bad --source-schema with PRECONDITION and add migrate --data-schema (S-007-12, S-007-13) (#169) * Refuse a bad --source-schema with PRECONDITION and add migrate --data-schema A schema name must match ^[a-z_][a-z0-9_]*$ and be at most 63 characters, and may be named once; a malformed or repeated name now exits PRECONDITION from the command before any playbook, as spec 007 says, where it was a usage error and a repeat was silently dropped. cloudfall migrate takes --data-schema DATABASE=NAME beside --data, so the data step copies only the named schemas as data copy --source-schema does; one for a database with no --data is refused before any step (S-007-12, S-007-13). Fixes #126 Co-Authored-By: Claude Opus 5.5 * Describe a resumed migrate step by this run's options A pending step of a persisted plan kept the description it was saved with, so a data step saved with --data-schema crm=public and resumed without it reported "schema public" while it copied every schema (#126, S-007-13). Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: shipmill-romamo[bot] <338557544+shipmill-romamo[bot]@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 --- changelog.d/126-source-schema-checks.md | 3 + docs/neon-migration-guide.md | 8 +- docs/render-migration-guide.md | 6 +- sdk/src/cloudfall/app.py | 61 +++++++--- sdk/src/cloudfall/domain.py | 28 +++-- sdk/src/cloudfall/lifecycle.py | 30 ++++- sdk/src/cloudfall/migrate.py | 35 +++++- sdk/tests/test_agent_server.py | 41 +++++++ sdk/tests/test_app.py | 16 +++ sdk/tests/test_data_migration.py | 146 +++++++++++++++++++++++- sdk/tests/test_lifecycle.py | 89 ++++++++++++++- sdk/tests/test_migrate.py | 59 ++++++++++ 12 files changed, 474 insertions(+), 48 deletions(-) create mode 100644 changelog.d/126-source-schema-checks.md diff --git a/changelog.d/126-source-schema-checks.md b/changelog.d/126-source-schema-checks.md new file mode 100644 index 0000000..93d6999 --- /dev/null +++ b/changelog.d/126-source-schema-checks.md @@ -0,0 +1,3 @@ +### Added + +- `cloudfall migrate` takes a repeatable `--data-schema DATABASE=NAME` beside `--data DATABASE=PATH` (and the `migrate` MCP tool a `data_schema` list): the `data:` step copies only the named schemas, as `data copy --source-schema` does, and the plan names them. A `--source-schema` or `--data-schema` name must now match `^[a-z_][a-z0-9_]*$` and be at most 63 characters, and may be given once per copy; a name that doesn't, a repeated name, or a `--data-schema` for a database with no `--data` exits `PRECONDITION` (code `data_copy_schema_invalid` or `migrate_data_schema_without_data`) before any playbook runs, where a malformed name was a usage error and a repeat was silently dropped (#126) diff --git a/docs/neon-migration-guide.md b/docs/neon-migration-guide.md index dc94f24..e617441 100644 --- a/docs/neon-migration-guide.md +++ b/docs/neon-migration-guide.md @@ -181,7 +181,10 @@ By default the copy takes every non-system schema of the source, and the row-count check and the receipt cover all of them. To take only some, name each with a repeatable `--source-schema`, such as `--source-schema public`; the plan names the schemas it will copy, and the -receipt lists them under `spec.schemas`, each table with its `schema`. +receipt lists them under `spec.schemas`, each table with its `schema`. A +name must match `^[a-z_][a-z0-9_]*$`, be at most 63 characters, and be +named once; anything else stops the command with `PRECONDITION` before +any playbook runs. The row-count check is the verify step and the receipt is the record's entry for the migration. The restore is not done because `pg_restore` @@ -193,7 +196,8 @@ refusal guards, and cannot skip either. One database at a time; repeat per declared database. If the application is also moving in the same operation, `cloudfall migrate --data app=tmp/neon-app-url` inserts this step into the orchestrated plan -between services and deploys. +between services and deploys; `--data-schema app=public` (repeatable) +narrows it to the named schemas, as `--source-schema` does. Note the one-time egress: Neon meters the dump like any other transfer, so a 14 GB dump costs about a dollar — the last transfer bill this diff --git a/docs/render-migration-guide.md b/docs/render-migration-guide.md index 77e7f67..4b6a9e1 100644 --- a/docs/render-migration-guide.md +++ b/docs/render-migration-guide.md @@ -152,8 +152,10 @@ database that already has tables in the copied schemas, verifies per-table row counts, and writes a `DataMigrationReceipt`. The row-count check is the verify step: the migration is not done because the restore exited zero, it is done because the counts match. The copy takes every non-system schema of -the source; `data copy --source-schema NAME` (repeatable) takes only the -named ones. +the source; `data copy --source-schema NAME` and `migrate --data-schema +DATABASE=NAME` (both repeatable) take only the named ones. A name must +match `^[a-z_][a-z0-9_]*$` and be named once; anything else stops the +command with `PRECONDITION` before any playbook runs. ## Step 7: cut over DNS diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index da63d41..2f95dcc 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -129,7 +129,6 @@ ConnectionAddress, Hostname, LinuxUser, - PostgresSchemaName, ReleaseId, ResourceId, ResourceKind, @@ -3113,6 +3112,15 @@ class MigrateArgs(EngineArgs): "Source URL file for a database, as DATABASE=PATH (repeatable)" ), ) + data_schema: tuple[str, ...] = Flag( + default=(), + description=( + "Source schema to copy into a database given --data, as " + "DATABASE=NAME (repeatable); NAME matches ^[a-z_][a-z0-9_]*$, at " + "most 63 characters, each named once per database; without it " + "every non-system schema of the source is copied" + ), + ) yes: bool = Flag( default=False, confirm=True, @@ -3154,9 +3162,15 @@ def __post_init__(self) -> None: ("data", self.data), ): _pairs(entries, option) + _grouped_pairs(self.data_schema, "data-schema") def options(self, fleet: Fleet) -> MigrateOptions: - """Return what this run builds, deploys and migrates.""" + """Return what this run builds, deploys and migrates. + + A malformed or repeated ``--data-schema`` name raises + ``LifecycleError``, which the command answers with ``PRECONDITION`` + (S-007-12). + """ return MigrateOptions( plan_file=fleet.path(self.plan_file), builds=_pairs(self.build, "build"), @@ -3172,6 +3186,12 @@ def options(self, fleet: Fleet) -> MigrateOptions: database: fleet.path(Path(value)) for database, value in _pairs(self.data, "data").items() }, + data_schemas={ + database: DataCopySchemas.from_boundary(names) + for database, names in _grouped_pairs( + self.data_schema, "data-schema" + ).items() + }, execute=self.yes, restart=self.restart, ) @@ -3205,6 +3225,18 @@ def _pairs(entries: tuple[str, ...], option: str) -> dict[str, str]: return pairs +def _grouped_pairs(entries: tuple[str, ...], option: str) -> dict[str, list[str]]: + """Group repeatable NAME=VALUE pairs by name, keeping every value given.""" + grouped: dict[str, list[str]] = {} + for entry in entries: + key, separator, value = entry.partition("=") + if not separator or not key or not value: + message = f"--{option} expects NAME=VALUE, got {entry!r}" + raise ParseError(message, context={"flag": option}) + grouped.setdefault(key, []).append(value) + return grouped + + @dataclass(frozen=True, slots=True) class Migration: """The migration plan's steps and how far the run got.""" @@ -3244,6 +3276,8 @@ def migrate(args: MigrateArgs, ctx: Ctx, fleet: Fleet) -> Migration: """Run each pending step, saving progress after every one.""" try: result = execute_migration(args.config(fleet, ctx), args.options(fleet)) + except LifecycleError as error: + raise _lifecycle_failed(error) from error except MigrateError as error: raise Exit.PRECONDITION( error.detail, context={"code": error.code} @@ -3843,8 +3877,9 @@ class DataMigrateArgs(EngineArgs): source_schema: tuple[str, ...] = Flag( default=(), description=( - "Source schema to copy (repeatable); without it every non-system " - "schema of the source is copied" + "Source schema to copy (repeatable), matching ^[a-z_][a-z0-9_]*$ " + "and at most 63 characters, each named once; without it every " + "non-system schema of the source is copied" ), ) data_migrations: Path = Flag( @@ -3861,18 +3896,18 @@ class DataMigrateArgs(EngineArgs): ) def __post_init__(self) -> None: - """Keep relative paths inside the project; check the schema names.""" + """Keep relative paths inside the project.""" EngineArgs.__post_init__(self) _inside_project(self.data_migrations, "data-migrations") - self.source_schemas() def source_schemas(self) -> DataCopySchemas: - """Return the named source schemas, refusing a malformed name.""" - try: - names = tuple(PostgresSchemaName(name) for name in self.source_schema) - except ValueError as error: - raise ParseError(str(error), context={"flag": "source-schema"}) from error - return DataCopySchemas(names) + """Return the named source schemas. + + A malformed or repeated name raises ``LifecycleError``, which the + command answers with ``PRECONDITION`` before any playbook (S-007-12), + not a usage error. + """ + return DataCopySchemas.from_boundary(self.source_schema) class DeployPayload(Payload): @@ -4066,8 +4101,8 @@ def data_copy( """Restore the source database into the service, or show the plan.""" context = args.context(fleet, ctx) source = fleet.path(args.source_url_file) - schemas = args.source_schemas() try: + schemas = args.source_schemas() if not args.yes: return DataMigratePayload( _plan( diff --git a/sdk/src/cloudfall/domain.py b/sdk/src/cloudfall/domain.py index 42e3b17..5d055c9 100644 --- a/sdk/src/cloudfall/domain.py +++ b/sdk/src/cloudfall/domain.py @@ -27,6 +27,9 @@ _BUILD_TOOL_NAME_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,63}") _BUILD_TOOL_VERSION_PATTERN = re.compile(r"[0-9]{1,9}(?:\.[0-9]{1,9}){0,7}") _CONTROL_CHARACTER_PATTERN = re.compile(r"[\x00-\x1f\x7f]") +# ^[a-z_][a-z0-9_]*$ and at most 63 characters (S-007), read by fullmatch so +# a trailing newline can't pass as it would with $. +_POSTGRES_SCHEMA_NAME_PATTERN = re.compile(r"[a-z_][a-z0-9_]{0,62}") BUILD_STEP_DEFAULT_TIMEOUT_SECONDS = 900 BUILD_STEP_MAXIMUM_TIMEOUT_SECONDS = 3600 ARTIFACT_CLONE_TIMEOUT_SECONDS = 1800 @@ -34,7 +37,6 @@ _FILE_MODE_PATTERN = re.compile(r"^0[0-7]{3}$") _SHA256_PATTERN = re.compile(r"^[a-f0-9]{64}$") _MAX_HOSTNAME_LENGTH = 253 -_POSTGRES_NAME_MAX_BYTES = 63 _MAX_TCP_PORT = 65535 _MIN_HTTP_STATUS = 100 _MAX_HTTP_STATUS = 599 @@ -342,26 +344,22 @@ def from_boundary(cls, value: object) -> LinuxUser: class PostgresSchemaName: """A PostgreSQL schema name a data copy may select. - The name travels to the engine as an Ansible extra var, which Ansible - templates, so braces are refused; PostgreSQL truncates a name past 63 - bytes, and a control character breaks the row-count lines. + Lower-case letters, digits and underscores, not starting with a digit, + at most 63 characters (S-007-12). The name travels to the engine as an + Ansible extra var, which Ansible templates, so no brace may pass; + PostgreSQL truncates a name past 63 bytes, and a quote, a semicolon, or a + control character has no place in a name the dump, the checks, and the + row-count lines all carry. """ value: str def __post_init__(self) -> None: - """Refuse a name PostgreSQL or Ansible would read differently.""" - if not self.value: - message = "schema name must not be empty" - raise ValueError(message) - if len(self.value.encode()) > _POSTGRES_NAME_MAX_BYTES: - message = f"schema name exceeds 63 bytes: {self.value!r}" - raise ValueError(message) - if _CONTROL_CHARACTER_PATTERN.search(self.value) or any( - brace in self.value for brace in "{}" - ): + """Refuse a name outside ``^[a-z_][a-z0-9_]*$`` or past 63 characters.""" + if not _POSTGRES_SCHEMA_NAME_PATTERN.fullmatch(self.value): message = ( - f"schema name holds a control character or a brace: {self.value!r}" + f"schema name {self.value!r} must match ^[a-z_][a-z0-9_]*$ " + "and be at most 63 characters" ) raise ValueError(message) diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index d3a0498..13189b8 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -30,6 +30,7 @@ DnsMode, Hostname, IpAddress, + PostgresSchemaName, Sha256Digest, TlsMode, ) @@ -44,12 +45,11 @@ from cloudfall.validation import SchemaCatalog, validate_config if TYPE_CHECKING: - from collections.abc import Callable, Mapping + from collections.abc import Callable, Mapping, Sequence from pathlib import Path from cloudfall.domain import ( CommandArgv, - PostgresSchemaName, ReleaseCommandTimeout, ReleaseId, ResourceId, @@ -81,6 +81,7 @@ _ERROR_SERVICE_MISSING = "lifecycle_service_missing" _ERROR_DATABASE_MISSING = "lifecycle_database_missing" _ERROR_SOURCE_URL_MISSING = "lifecycle_source_url_missing" +ERROR_DATA_COPY_SCHEMA_INVALID = "data_copy_schema_invalid" class LifecycleError(RuntimeError): @@ -269,9 +270,28 @@ class DataCopySchemas: names: tuple[PostgresSchemaName, ...] = () def __post_init__(self) -> None: - """Keep the names sorted so plans and receipts compare verbatim.""" - unique = sorted(set(self.names), key=lambda name: name.value) - object.__setattr__(self, "names", tuple(unique)) + """Refuse a name given twice; keep the names sorted for plans and receipts.""" + repeated = sorted( + {name.value for name in self.names if self.names.count(name) > 1} + ) + if repeated: + message = f"schema named more than once: {', '.join(repeated)}" + raise ValueError(message) + ordered = sorted(self.names, key=lambda name: name.value) + object.__setattr__(self, "names", tuple(ordered)) + + @classmethod + def from_boundary(cls, values: Sequence[str]) -> DataCopySchemas: + """Check the names a flag or a tool call gave, refusing a bad one. + + Raises ``LifecycleError`` with code ``data_copy_schema_invalid`` for + a name outside ``^[a-z_][a-z0-9_]*$``, past 63 characters, or given + twice (S-007-12), before any playbook runs. + """ + try: + return cls(tuple(PostgresSchemaName(value) for value in values)) + except ValueError as error: + raise LifecycleError(ERROR_DATA_COPY_SCHEMA_INVALID, str(error)) from error @property def every_schema(self) -> bool: diff --git a/sdk/src/cloudfall/migrate.py b/sdk/src/cloudfall/migrate.py index cc42e5e..dda44c6 100644 --- a/sdk/src/cloudfall/migrate.py +++ b/sdk/src/cloudfall/migrate.py @@ -28,6 +28,7 @@ from cloudfall.domain import ReleaseId, ResourceId, TcpPort from cloudfall.inventory import PlatformInventory from cloudfall.lifecycle import ( + DataCopySchemas, DeployOptions, LifecycleError, build_release_artifact, @@ -80,6 +81,7 @@ _ERROR_AUDIT_DRIFT = "migrate_audit_drift" _ERROR_ROUTES_UNHEALTHY = "migrate_routes_unhealthy" _ERROR_DATABASE_UNKNOWN = "migrate_database_unknown" +_ERROR_DATA_SCHEMA_WITHOUT_DATA = "migrate_data_schema_without_data" class MigrateError(RuntimeError): @@ -139,6 +141,8 @@ class MigrateOptions: releases: Mapping[str, str] environment_files: Mapping[str, Path] data_migrations: Mapping[str, Path] = field(default_factory=dict) + data_schemas: Mapping[str, DataCopySchemas] = field(default_factory=dict) + """The schemas each ``data:`` step copies; absent copies all.""" execute: bool = False restart: bool = False @@ -235,8 +239,9 @@ def _computed_steps( steps.extend( PlanStep( f"data:{database}", - f"migrate external data into database {database} with " - "row-count verification", + f"migrate {_data_schemas(options, database).describe()} of the " + f"external data into database {database} with row-count " + "verification", ) for database in sorted(options.data_migrations) ) @@ -344,6 +349,19 @@ def _require_known_databases( f"service: {', '.join(unknown)}" ) raise MigrateError(_ERROR_DATABASE_UNKNOWN, detail) + # A schema for a database nothing is copied into would be silently + # ignored; refuse it before any step runs (S-007-12). + unpaired = sorted(set(options.data_schemas) - set(options.data_migrations)) + if unpaired: + detail = ( + "--data-schema names databases with no --data DATABASE=PATH: " + f"{', '.join(unpaired)}" + ) + raise MigrateError(_ERROR_DATA_SCHEMA_WITHOUT_DATA, detail) + + +def _data_schemas(options: MigrateOptions, database: str) -> DataCopySchemas: + return options.data_schemas.get(database, DataCopySchemas()) def _load_plan( @@ -388,6 +406,11 @@ def _load_plan( "with --restart to discard it" ) raise MigrateError(_ERROR_PLAN_STALE, detail) + # A pending step runs with this run's options (a data step's schemas + # among them), so it is described by them, not by the saved plan's text. + for saved, fresh in zip(persisted, computed, strict=True): + if saved.status is StepStatus.PENDING: + saved.description = fresh.description return persisted @@ -511,7 +534,11 @@ def _run() -> dict[str, object]: runners[f"deploy:{component_id}"] = _deploy_runner(component_id) for database, source_url_file in options.data_migrations.items(): runners[f"data:{database}"] = _data_runner( - config, inventory, database, source_url_file + config, + inventory, + database, + source_url_file, + _data_schemas(options, database), ) return runners @@ -521,6 +548,7 @@ def _data_runner( inventory: PlatformInventory, database: str, source_url_file: Path, + schemas: DataCopySchemas, ) -> Runner: def _run() -> dict[str, object]: service = next( @@ -538,6 +566,7 @@ def _run() -> dict[str, object]: database, source_url_file, config.data_migrations_directory, + schemas, ) return _run diff --git a/sdk/tests/test_agent_server.py b/sdk/tests/test_agent_server.py index a3d5afa..9627e71 100644 --- a/sdk/tests/test_agent_server.py +++ b/sdk/tests/test_agent_server.py @@ -277,6 +277,47 @@ def test_s004_16_a_project_serves_domains_apply_and_observe(tmp_path: Path) -> N assert planned["meta"]["dry_run"] is True +def test_s007_12_the_data_tools_refuse_a_bad_schema_as_the_cli_does( + tmp_path: Path, +) -> None: + # Without yes each call is the plan, so no playbook could run either way. + tools, (repeated, malformed, unpaired) = _talk( + _serve("--project", str(EXAMPLES), cwd=tmp_path), + ( + "data_copy", + { + "service": "postgresql-main", + "database": "crm", + "source_url_file": "tmp/source.url", + "source_schema": ["app", "app"], + }, + ), + ( + "data_copy", + { + "service": "postgresql-main", + "database": "crm", + "source_url_file": "tmp/source.url", + "source_schema": ["{{ lookup('pipe', 'id') }}"], + }, + ), + ("migrate", {"build": ["crm-backend=main"], "data_schema": ["crm=public"]}), + ) + + copy_schema = tools["data_copy"].input_schema["properties"]["source_schema"] + assert copy_schema["type"] == "array" + assert copy_schema["items"]["type"] == "string" + assert "data_schema" in tools["migrate"].input_schema["properties"] + for refused, code in ( + (repeated, "data_copy_schema_invalid"), + (malformed, "data_copy_schema_invalid"), + (unpaired, "migrate_data_schema_without_data"), + ): + assert refused["ok"] is False, refused + assert refused["error"]["code"] == "PRECONDITION" + assert refused["error"]["context"]["code"] == code + + def test_a_project_call_cannot_approve_its_own_proposal(tmp_path: Path) -> None: # A person approves a proposal from a shell; an agent that could call # operator approve would run and verify the change it proposed itself. diff --git a/sdk/tests/test_app.py b/sdk/tests/test_app.py index b176734..77ec5ef 100644 --- a/sdk/tests/test_app.py +++ b/sdk/tests/test_app.py @@ -96,6 +96,22 @@ def test_migrate_resolves_env_and_data_files_against_the_project( assert options.data_migrations == {"crm": tmp_path / "source.url"} +def test_s007_13_migrate_groups_each_data_schema_under_its_database( + tmp_path: Path, +) -> None: + fleet = Fleet(EXAMPLES, validate_config(EXAMPLES, SCHEMAS)) + args = MigrateArgs( + data=(f"crm={tmp_path / 'source.url'}",), + data_schema=("crm=public", "crm=app"), + ) + + schemas = args.options(fleet).data_schemas + + assert {database: entry.values for database, entry in schemas.items()} == { + "crm": ["app", "public"] + } + + def test_dashboard_serve_streams_where_it_listens_until_stopped( tmp_path: Path, ) -> None: diff --git a/sdk/tests/test_data_migration.py b/sdk/tests/test_data_migration.py index a017180..a450f5b 100644 --- a/sdk/tests/test_data_migration.py +++ b/sdk/tests/test_data_migration.py @@ -3,8 +3,10 @@ from __future__ import annotations import json +import subprocess +from dataclasses import replace from pathlib import Path -from typing import cast +from typing import TYPE_CHECKING, cast import pytest from cloudfall.agent_tools import AgentConfig @@ -13,12 +15,17 @@ DataCopySchemas, EngineContext, ExecutionPlan, + ExecutionStep, LifecycleError, migrate_data, plan_data_migration, + subprocess_step, ) from cloudfall.migrate import MigrateError, MigrateOptions, execute_migration +if TYPE_CHECKING: + from cloudfall.lifecycle import StepRunner + ROOT = Path(__file__).parents[2] SCHEMAS = ROOT / "config" / "schemas" / "v1" EXAMPLES = ROOT / "config" / "examples" @@ -115,7 +122,22 @@ def test_s007_2_data_migration_plan_without_schemas_is_unchanged( def test_data_copy_schemas_describe_the_selection() -> None: assert DataCopySchemas().describe() == "every non-system schema" assert _schemas("public").describe() == "schema public" - assert _schemas("public", "app", "public").describe() == "schemas app, public" + assert _schemas("public", "app").describe() == "schemas app, public" + + +def test_s007_12_data_copy_schemas_refuse_a_repeat_and_a_malformed_name() -> None: + with pytest.raises(ValueError, match="more than once: app"): + _schemas("app", "public", "app") + for malformed in ("Bad;Name", "Public", "1app", "app\n", "", "a" * 64): + with pytest.raises(ValueError, match=r"\^\[a-z_\]"): + PostgresSchemaName(malformed) + with pytest.raises(LifecycleError) as error: + DataCopySchemas.from_boundary(["app", "app"]) + assert error.value.code == "data_copy_schema_invalid" + assert DataCopySchemas.from_boundary(["public", "_app"]).values == [ + "_app", + "public", + ] def test_data_migration_rejects_an_undeclared_service(tmp_path: Path) -> None: @@ -157,7 +179,9 @@ def test_data_migration_rejects_an_empty_source_url_file( assert error.value.code == "lifecycle_source_url_missing" -def _agent_config(tmp_path: Path) -> AgentConfig: +def _agent_config( + tmp_path: Path, run: StepRunner = subprocess_step +) -> AgentConfig: return AgentConfig( project_directory=EXAMPLES, schema_directory=SCHEMAS, @@ -169,9 +193,125 @@ def _agent_config(tmp_path: Path) -> AgentConfig: releases_directory=tmp_path / "releases", artifacts_directory=tmp_path / "artifacts", data_migrations_directory=tmp_path / "data-migrations", + run=run, ) +def _succeeding(calls: list[ExecutionStep]) -> StepRunner: + """Record each engine step and answer it as a run that passed.""" + + def run(step: ExecutionStep) -> subprocess.CompletedProcess[str]: + calls.append(step) + return subprocess.CompletedProcess(step.argv, 0, "", "") + + return run + + +def test_s007_13_migrate_copies_only_the_data_schema_as_data_copy_does( + tmp_path: Path, +) -> None: + source = _url_file(tmp_path) + calls: list[ExecutionStep] = [] + config = _agent_config(tmp_path, _succeeding(calls)) + options = MigrateOptions( + plan_file=tmp_path / "plan.json", + builds={"crm-backend": "main"}, + releases={}, + environment_files={}, + data_migrations={"crm": source}, + data_schemas={"crm": _schemas("public")}, + execute=True, + ) + planned = execute_migration(config, replace(options, execute=False)) + steps = cast("list[dict[str, object]]", planned["steps"]) + (described,) = [step for step in steps if step["id"] == "data:crm"] + assert "schema public" in str(described["description"]) + # Every other step already ran, so the resumed run is the data step alone. + for step in steps: + if step["id"] != "data:crm": + step["status"] = "completed" + options.plan_file.write_text( + json.dumps({"version": 1, "steps": steps}), encoding="utf-8" + ) + + result = execute_migration(config, options) + + assert result["status"] == "ok", result + (playbook,) = [step for step in calls if "--extra-vars" in step.argv] + copied = plan_data_migration( + _context(tmp_path), + ResourceId("postgresql-main"), + "crm", + source, + config.data_migrations_directory, + _schemas("public"), + ) + assert playbook.argv == copied.steps[-1].argv + assert _extra_vars(copied)["cloudfall_data_schemas"] == ["public"] + + +def test_s007_13_a_resumed_data_step_names_the_schemas_it_copies( + tmp_path: Path, +) -> None: + source = _url_file(tmp_path) + calls: list[ExecutionStep] = [] + config = _agent_config(tmp_path, _succeeding(calls)) + options = MigrateOptions( + plan_file=tmp_path / "plan.json", + builds={"crm-backend": "main"}, + releases={}, + environment_files={}, + data_migrations={"crm": source}, + data_schemas={"crm": _schemas("public")}, + execute=True, + ) + steps = cast( + "list[dict[str, object]]", + execute_migration(config, replace(options, execute=False))["steps"], + ) + for step in steps: + if step["id"] != "data:crm": + step["status"] = "completed" + options.plan_file.write_text( + json.dumps({"version": 1, "steps": steps}), encoding="utf-8" + ) + + # Resumed without --data-schema, the step copies every schema, and the + # plan it reports must say so, not the "schema public" it was saved with. + result = execute_migration(config, replace(options, data_schemas={})) + + (playbook,) = [step for step in calls if "--extra-vars" in step.argv] + extra_vars = json.loads(playbook.argv[playbook.argv.index("--extra-vars") + 1]) + assert "cloudfall_data_schemas" not in extra_vars + (described,) = [ + step + for step in cast("list[dict[str, object]]", result["steps"]) + if step["id"] == "data:crm" + ] + assert "every non-system schema" in str(described["description"]) + + +def test_s007_12_migrate_refuses_a_data_schema_without_its_data( + tmp_path: Path, +) -> None: + calls: list[ExecutionStep] = [] + options = MigrateOptions( + plan_file=tmp_path / "plan.json", + builds={"crm-backend": "main"}, + releases={}, + environment_files={}, + data_schemas={"crm": _schemas("public")}, + execute=True, + ) + + with pytest.raises(MigrateError) as error: + execute_migration(_agent_config(tmp_path, _succeeding(calls)), options) + + assert error.value.code == "migrate_data_schema_without_data" + assert calls == [] + assert not options.plan_file.exists() + + def test_migrate_plan_orders_data_steps_after_services( tmp_path: Path, ) -> None: diff --git a/sdk/tests/test_lifecycle.py b/sdk/tests/test_lifecycle.py index bc1b3c3..90bb209 100644 --- a/sdk/tests/test_lifecycle.py +++ b/sdk/tests/test_lifecycle.py @@ -369,15 +369,63 @@ def test_s007_14_data_copy_plan_names_the_selected_schemas( @pytest.mark.parametrize( - "schema", - ["", "{{ lookup('pipe', 'id') }}", "{% raw %}", "a\nb", "x" * 64], + "schemas", + [ + ["Bad;Name"], + ["app", "app"], + [""], + ["{{ lookup('pipe', 'id') }}"], + ["{% raw %}"], + ["app\t"], + ["Public"], + ["1app"], + ['app"'], + ["x" * 64], + ], ) -def test_data_copy_refuses_a_schema_name_ansible_or_postgres_would_misread( - schema: str, tmp_path: Path, capsys: pytest.CaptureFixture[str] +@pytest.mark.parametrize("confirmed", [False, True]) +def test_s007_12_data_copy_refuses_a_malformed_or_repeated_schema( + schemas: list[str], + confirmed: bool, # noqa: FBT001 - a pytest parameter. + tmp_path: Path, + capsys: pytest.CaptureFixture[str], ) -> None: """A name reaches the engine as an extra var, which Ansible templates.""" source_url_file = tmp_path / "source.url" source_url_file.write_text("postgresql://u:p@db.example.test/crm\n") + flags = [argument for name in schemas for argument in ("--source-schema", name)] + + exit_code, payload = _cli_lifecycle( + [ + "data", + "copy", + "postgresql-main", + "--database", + "crm", + "--source-url-file", + str(source_url_file), + *flags, + *(["--yes"] if confirmed else []), + ], + tmp_path, + capsys, + ) + + assert exit_code == 4 + assert payload["error"]["code"] == "PRECONDITION" + assert payload["error"]["context"]["code"] == "data_copy_schema_invalid" + assert "u:p@" not in json.dumps(payload) + # The inventory render is every playbook run's first step. + assert not (tmp_path / "inventory.json").exists() + + +@pytest.mark.parametrize("schema", ["a\nb", "app\n"]) +def test_data_copy_argv_refuses_a_schema_with_a_newline_before_the_command( + schema: str, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """The CLI framework refuses a newline in any argv value as a usage error.""" + source_url_file = tmp_path / "source.url" + source_url_file.write_text("postgresql://u:p@db.example.test/crm\n") exit_code, payload = _cli_lifecycle( [ @@ -390,13 +438,44 @@ def test_data_copy_refuses_a_schema_name_ansible_or_postgres_would_misread( str(source_url_file), "--source-schema", schema, + "--yes", ], tmp_path, capsys, ) assert exit_code == 2 - assert payload["ok"] is False + assert payload["error"]["code"] == "ARG_ERROR" + assert not (tmp_path / "inventory.json").exists() + + +def test_s007_12_data_copy_takes_the_longest_schema_name( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + source_url_file = tmp_path / "source.url" + source_url_file.write_text("postgresql://u:p@db.example.test/crm\n") + longest = "_" + "a1" * 31 + + exit_code, payload = _cli_lifecycle( + [ + "data", + "copy", + "postgresql-main", + "--database", + "crm", + "--source-url-file", + str(source_url_file), + "--source-schema", + longest, + ], + tmp_path, + capsys, + ) + + assert exit_code == 0, payload + assert f"dump schema {longest} of the database" in str( + payload["data"]["would_run"] + ) def _git(*arguments: str) -> None: diff --git a/sdk/tests/test_migrate.py b/sdk/tests/test_migrate.py index 79bb811..354d08b 100644 --- a/sdk/tests/test_migrate.py +++ b/sdk/tests/test_migrate.py @@ -266,6 +266,65 @@ def test_cli_migrate_rejects_malformed_component_pairs( ) +@pytest.mark.parametrize( + ("flags", "code"), + [ + (["--data-schema", "crm=public"], "migrate_data_schema_without_data"), + ( + ["--data", "crm=SOURCE", "--data-schema", "crm=Bad;Name"], + "data_copy_schema_invalid", + ), + ( + [ + "--data", + "crm=SOURCE", + "--data-schema", + "crm=app", + "--data-schema", + "crm=app", + ], + "data_copy_schema_invalid", + ), + ], +) +def test_s007_12_cli_migrate_refuses_a_bad_data_schema_before_any_playbook( + flags: list[str], + code: str, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + source = tmp_path / "source.url" + source.write_text("postgresql://u:p@db.example.test/crm\n", encoding="utf-8") + + exit_code = main( + [ + "migrate", + "--project", + str(EXAMPLES), + "--schemas", + str(SCHEMAS), + "--engine", + str(ENGINE), + "--plan-file", + str(tmp_path / "plan.json"), + "--inventory-file", + str(tmp_path / "inventory.json"), + "--build", + "crm-backend=main", + *(flag.replace("SOURCE", str(source)) for flag in flags), + "--yes", + ] + ) + + payload = json.loads(capsys.readouterr().out) + assert exit_code == 4 + assert payload["error"]["code"] == "PRECONDITION" + assert payload["error"]["context"]["code"] == code + assert "u:p@" not in json.dumps(payload) + assert not (tmp_path / "plan.json").exists() + assert not (tmp_path / "inventory.json").exists() + + # dns-verify's proxy addresses, pinned as they were before domains apply. From 835d7cec34d00a27948ceacee28b3738ac237dbb Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:24:08 +0200 Subject: [PATCH 3/4] Validate Domain routes (S-003-1, S-003-8..12) (#168) * Validate Domain routes (S-003-1, S-003-8..12) A Domain can now declare spec.routes: proxy routes (path + upstream) and static routes (path + static component, directory, spaFallback). The schema holds each route's shape and the release-relative directory and fallback; config validate checks the route source, path syntax, reserved and duplicate paths, and that a static component exists and is deployed on the proxy server. The inventory parses routes into value objects and serializes them only when present, so a Domain with only proxy.upstream is unchanged. Rendering routes comes with #162. Fixes #159 Part of #90 Co-Authored-By: Claude Opus 5.5 * Refuse a trailing newline in a static route's directory or spaFallback The releasePath schema pattern ends with $, which Python's re.search lets match before one trailing newline, so config validate accepted directory: "web\n" that ReleaseRelativePath then refused with a ValueError when the inventory loaded. Refuse control characters in the schema, as component.schema.json's controlCharacterFree does. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: shipmill-romamo[bot] <338557544+shipmill-romamo[bot]@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 --- changelog.d/159-domain-routes-validation.md | 3 + config/schemas/v1/domain.schema.json | 116 ++++++-- sdk/src/cloudfall/domain.py | 70 +++++ sdk/src/cloudfall/inventory.py | 99 ++++++- sdk/src/cloudfall/validation.py | 140 +++++++++ sdk/tests/test_domain_routes.py | 312 ++++++++++++++++++++ 6 files changed, 715 insertions(+), 25 deletions(-) create mode 100644 changelog.d/159-domain-routes-validation.md create mode 100644 sdk/tests/test_domain_routes.py diff --git a/changelog.d/159-domain-routes-validation.md b/changelog.d/159-domain-routes-validation.md new file mode 100644 index 0000000..7641cbb --- /dev/null +++ b/changelog.d/159-domain-routes-validation.md @@ -0,0 +1,3 @@ +### Added + +- A Domain can declare `spec.routes`, path routes that each either proxy to an `upstream` (`address`, `port`) or serve a component's release files (`static`: `component`, `directory`, optional `spaFallback`). `cloudfall config validate` refuses a Domain with both `proxy.upstream` and `routes` (`domain_route_source_invalid`), a path that doesn't start with `/` or uses characters outside letters, digits and `._~/-`, or a static path without a trailing `/` (`domain_route_path_invalid`), a path declared twice (`domain_route_path_duplicate`), a path under `/.well-known/acme-challenge` (`domain_route_path_reserved`), a missing static component (`resource_reference_missing`) or one not deployed on the Domain's `proxy.server` (`domain_route_component_not_on_proxy`), and a `directory` or `spaFallback` that is absolute, climbs out with `..`, or, for `directory`, names the release root. Validation only: the domains playbook doesn't render routes yet and still refuses a Domain without `proxy.upstream` before any change. Domains with only `proxy.upstream` validate and render as before (#159) diff --git a/config/schemas/v1/domain.schema.json b/config/schemas/v1/domain.schema.json index 138e594..c3f6028 100644 --- a/config/schemas/v1/domain.schema.json +++ b/config/schemas/v1/domain.schema.json @@ -3,6 +3,90 @@ "$id": "https://cloudfall.internal/schemas/v1/domain.schema.json", "title": "Cloudfall public domain route", "x-cloudfall-kind": "Domain", + "$defs": { + "upstream": { + "type": "object", + "properties": { + "address": { + "type": "string", + "anyOf": [ + { + "format": "ipv4" + }, + { + "format": "ipv6" + } + ] + }, + "port": { + "type": "integer", + "minimum": 1, + "maximum": 65535 + } + }, + "required": [ + "address", + "port" + ], + "additionalProperties": false + }, + "releasePath": { + "type": "string", + "pattern": "^(?!/)(?!(?:.*/)?\\.\\.(?:/|$))[A-Za-z0-9._/-]+$", + "not": { + "pattern": "[\\u0000-\\u001f\\u007f]" + } + }, + "route": { + "type": "object", + "properties": { + "path": { + "type": "string" + }, + "upstream": { + "$ref": "#/$defs/upstream" + }, + "static": { + "type": "object", + "properties": { + "component": { + "$ref": "common.schema.json#/$defs/resourceId" + }, + "directory": { + "$ref": "#/$defs/releasePath", + "not": { + "pattern": "^\\.(?:/+\\.)*/*$" + } + }, + "spaFallback": { + "$ref": "#/$defs/releasePath" + } + }, + "required": [ + "component", + "directory" + ], + "additionalProperties": false + } + }, + "required": [ + "path" + ], + "oneOf": [ + { + "required": [ + "upstream" + ] + }, + { + "required": [ + "static" + ] + } + ], + "additionalProperties": false + } + }, "type": "object", "properties": { "apiVersion": { @@ -42,30 +126,7 @@ "$ref": "common.schema.json#/$defs/serviceName" }, "upstream": { - "type": "object", - "properties": { - "address": { - "type": "string", - "anyOf": [ - { - "format": "ipv4" - }, - { - "format": "ipv6" - } - ] - }, - "port": { - "type": "integer", - "minimum": 1, - "maximum": 65535 - } - }, - "required": [ - "address", - "port" - ], - "additionalProperties": false + "$ref": "#/$defs/upstream" } }, "required": [ @@ -75,6 +136,13 @@ ], "additionalProperties": false }, + "routes": { + "type": "array", + "items": { + "$ref": "#/$defs/route" + }, + "minItems": 1 + }, "origin": { "type": "object", "properties": { diff --git a/sdk/src/cloudfall/domain.py b/sdk/src/cloudfall/domain.py index 5d055c9..d02f8e2 100644 --- a/sdk/src/cloudfall/domain.py +++ b/sdk/src/cloudfall/domain.py @@ -850,6 +850,76 @@ def from_boundary(cls, value: object) -> RelativeDirectory: return cls(_required_string(value, "relative directory")) +_ROUTE_PATH_PATTERN = re.compile(r"/[A-Za-z0-9._~/-]*") +_ACME_CHALLENGE_PATH = "/.well-known/acme-challenge" + + +@dataclass(frozen=True, slots=True) +class DomainRoutePath: + """URL path prefix of a Domain route, safe to render as an nginx location.""" + + value: str + + def __post_init__(self) -> None: + """Accept only ``/`` then letters, digits and ``._~/-``.""" + if not _ROUTE_PATH_PATTERN.fullmatch(self.value): + message = ( + f"invalid route path: {self.value!r} must start with '/' and use " + "only letters, digits and '._~/-'" + ) + raise ValueError(message) + + @property + def is_directory(self) -> bool: + """Return whether the path ends with ``/``, as a static route needs.""" + return self.value.endswith("/") + + @property + def is_reserved(self) -> bool: + """Return whether certificate issuance owns the path.""" + return self.value == _ACME_CHALLENGE_PATH or self.value.startswith( + f"{_ACME_CHALLENGE_PATH}/" + ) + + @classmethod + def from_boundary(cls, value: object) -> DomainRoutePath: + """Coerce a boundary value into a route path.""" + return cls(_required_string(value, "route path")) + + +_RELEASE_PATH_PATTERN = re.compile(r"[A-Za-z0-9._/-]+") + + +@dataclass(frozen=True, slots=True) +class ReleaseRelativePath: + """Path inside a component release that nginx may serve files from.""" + + value: str + + def __post_init__(self) -> None: + """Reject absolute paths, ``..`` segments and characters outside ``._-/``.""" + if ( + not _RELEASE_PATH_PATTERN.fullmatch(self.value) + or self.value.startswith("/") + or ".." in self.value.split("/") + ): + message = ( + f"invalid release path: {self.value!r} must be relative, use only " + "letters, digits and '._-/', and have no '..' segment" + ) + raise ValueError(message) + + @property + def is_release_root(self) -> bool: + """Return whether the path names the release root itself, such as ``.``.""" + return PurePosixPath(self.value) == PurePosixPath(".") + + @classmethod + def from_boundary(cls, value: object) -> ReleaseRelativePath: + """Coerce a boundary value into a release path.""" + return cls(_required_string(value, "release path")) + + _RELEASE_TIMEOUT_MINIMUM = 1 _RELEASE_TIMEOUT_MAXIMUM = 1800 _RELEASE_TIMEOUT_DEFAULT = 300 diff --git a/sdk/src/cloudfall/inventory.py b/sdk/src/cloudfall/inventory.py index 4ecf56b..34595c7 100644 --- a/sdk/src/cloudfall/inventory.py +++ b/sdk/src/cloudfall/inventory.py @@ -23,6 +23,7 @@ DayTime, DeploymentApproval, DnsMode, + DomainRoutePath, EmailAddress, FileMode, FilesystemName, @@ -56,6 +57,7 @@ RelativeDirectory, RelativeRepositoryPath, ReleaseCommandTimeout, + ReleaseRelativePath, RepositoryUrl, RequiredServiceState, RequiredServiceStatus, @@ -698,6 +700,57 @@ class DomainHealthCheck: timeout_seconds: PositiveCount +@dataclass(frozen=True, slots=True) +class DomainProxyRoute: + """A path a Domain proxies to an upstream, keeping the full request path.""" + + path: DomainRoutePath + upstream_address: IpAddress + upstream_port: TcpPort + + def as_dict(self) -> dict[str, object]: + """Serialize the route in its config shape.""" + return { + "path": self.path.value, + "upstream": { + "address": self.upstream_address.value, + "port": self.upstream_port.value, + }, + } + + +@dataclass(frozen=True, slots=True) +class DomainStaticRoute: + """A path a Domain serves as files from a component's current release.""" + + path: DomainRoutePath + component_id: ResourceId + directory: ReleaseRelativePath + spa_fallback: ReleaseRelativePath | None = None + + def __post_init__(self) -> None: + """Keep the alias-safe trailing slash and the release root out.""" + if not self.path.is_directory: + message = f"static route path must end with '/': {self.path.value!r}" + raise ValueError(message) + if self.directory.is_release_root: + message = "static route directory must not be the release root" + raise ValueError(message) + + def as_dict(self) -> dict[str, object]: + """Serialize the route in its config shape.""" + static: dict[str, object] = { + "component": self.component_id.value, + "directory": self.directory.value, + } + if self.spa_fallback is not None: + static["spaFallback"] = self.spa_fallback.value + return {"path": self.path.value, "static": static} + + +type DomainRoute = DomainProxyRoute | DomainStaticRoute + + @dataclass(frozen=True, slots=True) class DomainInventory: """Typed desired state for one public domain route.""" @@ -710,6 +763,13 @@ class DomainInventory: edge: DomainEdge tls_mode: TlsMode health_check: DomainHealthCheck + routes: tuple[DomainRoute, ...] = () + + def __post_init__(self) -> None: + """Hold at most one route source: ``proxy.upstream`` or ``routes``.""" + if self.routes and self.proxy.upstream_address is not None: + message = "a Domain declares proxy.upstream or routes, not both" + raise ValueError(message) def as_dict(self) -> dict[str, object]: """Serialize public routing intent without secret material.""" @@ -726,7 +786,7 @@ def as_dict(self) -> dict[str, object]: "address": self.proxy.upstream_address.value, "port": self.proxy.upstream_port.value, } - return { + result: dict[str, object] = { "id": self.resource_id.value, "primaryName": self.primary_name.value, "aliases": [alias.value for alias in self.aliases], @@ -753,6 +813,9 @@ def as_dict(self) -> dict[str, object]: "timeoutSeconds": self.health_check.timeout_seconds.value, }, } + if self.routes: + result["routes"] = [route.as_dict() for route in self.routes] + return result @dataclass(frozen=True, slots=True) @@ -1855,6 +1918,40 @@ def _domain_inventory(document: ResourceDocument) -> DomainInventory: ), timeout_seconds=PositiveCount.from_boundary(health.get("timeoutSeconds")), ), + routes=( + () + if spec.get("routes") is None + else tuple( + _domain_route(item) + for item in _mapping_list(spec.get("routes"), "domain routes") + ) + ), + ) + + +def _domain_route(route: Mapping[str, object]) -> DomainRoute: + path = DomainRoutePath.from_boundary(route.get("path")) + if route.get("static") is None: + upstream = _mapping(route, "upstream") + return DomainProxyRoute( + path=path, + upstream_address=IpAddress.from_boundary(upstream.get("address")), + upstream_port=TcpPort.from_boundary(upstream.get("port")), + ) + if route.get("upstream") is not None: + message = f"route {path.value!r} declares both upstream and static" + raise ValueError(message) + static = _mapping(route, "static") + raw_spa_fallback = static.get("spaFallback") + return DomainStaticRoute( + path=path, + component_id=ResourceId.from_boundary(static.get("component")), + directory=ReleaseRelativePath.from_boundary(static.get("directory")), + spa_fallback=( + None + if raw_spa_fallback is None + else ReleaseRelativePath.from_boundary(raw_spa_fallback) + ), ) diff --git a/sdk/src/cloudfall/validation.py b/sdk/src/cloudfall/validation.py index 8dfd5ed..7adf351 100644 --- a/sdk/src/cloudfall/validation.py +++ b/sdk/src/cloudfall/validation.py @@ -19,6 +19,7 @@ from referencing import Registry, Resource from cloudfall.domain import ( + DomainRoutePath, IpAddress, IpNetworkCidr, OpenSshPublicKey, @@ -571,6 +572,7 @@ def _validate_domain_references( document.source, ("spec", field_name, "server"), ) + _validate_domain_routes(document, spec, proxy, index) primary_name = spec.get("primaryName") aliases = spec.get("aliases") if isinstance(aliases, list) and primary_name in aliases: @@ -593,6 +595,144 @@ def _validate_domain_references( raise ConfigValidationError(issue) +def _validate_domain_routes( + document: ResourceDocument, + spec: Mapping[str, object], + proxy: Mapping[str, object], + index: Mapping[ResourceKey, ResourceDocument], +) -> None: + """Check ``spec.routes``: one route source, safe unique paths, served components. + + The schema holds each route's shape, ``directory`` and ``spaFallback``; the + path rules live here so each refusal names its own code. + """ + if "routes" not in spec: + return + if "upstream" in proxy: + issue = ValidationIssue( + code="domain_route_source_invalid", + message="a Domain declares proxy.upstream or routes, not both", + source=document.source, + field_path=("spec", "routes"), + ) + raise ConfigValidationError(issue) + proxy_server_id = _resource_id_value(proxy, "server", document.source) + raw_routes = spec.get("routes") + if not isinstance(raw_routes, list): + issue = ValidationIssue( + code="internal_state_shape_invalid", + message="validated field 'routes' is not a list", + source=document.source, + field_path=("spec", "routes"), + ) + raise ConfigValidationError(issue) + seen: set[DomainRoutePath] = set() + for position, raw_route in enumerate(raw_routes): + route_field: tuple[str | int, ...] = ("spec", "routes", position) + if not isinstance(raw_route, dict): + issue = ValidationIssue( + code="internal_state_shape_invalid", + message="validated route is not an object", + source=document.source, + field_path=route_field, + ) + raise ConfigValidationError(issue) + route = cast("Mapping[str, object]", raw_route) + path = _domain_route_path(route, document.source, (*route_field, "path")) + if path in seen: + issue = ValidationIssue( + code="domain_route_path_duplicate", + message=f"two routes declare the path {path.value!r}", + source=document.source, + field_path=(*route_field, "path"), + ) + raise ConfigValidationError(issue) + seen.add(path) + if "static" in route: + _validate_static_route( + document, + path, + _required_mapping(route, "static", document.source), + proxy_server_id, + index, + route_field, + ) + + +def _domain_route_path( + route: Mapping[str, object], + source: SourceLocation, + field_path: tuple[str | int, ...], +) -> DomainRoutePath: + try: + path = DomainRoutePath.from_boundary(route.get("path")) + except (TypeError, ValueError) as error: + issue = ValidationIssue( + code="domain_route_path_invalid", + message=str(error), + source=source, + field_path=field_path, + ) + raise ConfigValidationError(issue) from error + if path.is_reserved: + issue = ValidationIssue( + code="domain_route_path_reserved", + message=( + f"route path {path.value!r} is under /.well-known/acme-challenge, " + "which certificate issuance owns" + ), + source=source, + field_path=field_path, + ) + raise ConfigValidationError(issue) + return path + + +def _validate_static_route( # noqa: PLR0913 + document: ResourceDocument, + path: DomainRoutePath, + static: Mapping[str, object], + proxy_server_id: ResourceId, + index: Mapping[ResourceKey, ResourceDocument], + route_field: tuple[str | int, ...], +) -> None: + if not path.is_directory: + issue = ValidationIssue( + code="domain_route_path_invalid", + message=( + f"static route path {path.value!r} must end with '/', so nginx " + "can't serve files from outside its directory" + ), + source=document.source, + field_path=(*route_field, "path"), + ) + raise ConfigValidationError(issue) + component_field = (*route_field, "static", "component") + component_id = _resource_id_value(static, "component", document.source) + component = _require_resource( + index, + ResourceKind.COMPONENT, + component_id, + document.source, + component_field, + ) + component_spec = _required_mapping(component.content, "spec", component.source) + deployment = _required_mapping(component_spec, "deployment", component.source) + if proxy_server_id not in _resource_id_list( + deployment, "servers", component.source + ): + issue = ValidationIssue( + code="domain_route_component_not_on_proxy", + message=( + f"Component/{component_id} is not deployed on the proxy server " + f"{proxy_server_id}, where nginx reads its files" + ), + source=document.source, + field_path=component_field, + ) + raise ConfigValidationError(issue) + + def _validate_service_references( document: ResourceDocument, spec: Mapping[str, object], diff --git a/sdk/tests/test_domain_routes.py b/sdk/tests/test_domain_routes.py new file mode 100644 index 0000000..dbae945 --- /dev/null +++ b/sdk/tests/test_domain_routes.py @@ -0,0 +1,312 @@ +"""Domain routes: the `routes` schema, its value objects and its validation (S-003).""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path +from typing import Any + +import pytest +from cloudfall.cli import main +from cloudfall.domain import DomainRoutePath, ReleaseRelativePath, ResourceId +from cloudfall.inventory import ( + DomainProxyRoute, + DomainStaticRoute, + PlatformInventory, +) +from cloudfall.validation import validate_config + +ROOT = Path(__file__).parents[2] +SCHEMAS = ROOT / "config" / "schemas" / "v1" +EXAMPLES = ROOT / "config" / "examples" + +_UPSTREAM = " upstream:\n address: 127.0.0.1\n port: 8100\n" +_SPA_ROUTES = """\ + routes: + - path: /api/ + upstream: { address: 127.0.0.1, port: 8310 } + - path: / + static: + component: crm-backend + directory: web/dist + spaFallback: index.html +""" + + +def _project(tmp_path: Path, routes: str, *, keep_upstream: bool = False) -> Path: + """Copy the examples and give their Domain ``routes`` in place of its upstream.""" + project = tmp_path / "config" + shutil.copytree(EXAMPLES, project) + path = project / "domains" / "crm-site.yaml" + text = path.read_text(encoding="utf-8") + assert _UPSTREAM in text + if not keep_upstream: + text = text.replace(_UPSTREAM, "") + text = text.replace(" origin:\n", routes + " origin:\n") + path.write_text(text, encoding="utf-8") + return project + + +def _validate( + project: Path, capsys: pytest.CaptureFixture[str] +) -> tuple[int, dict[str, Any]]: + exit_code = main( + [ + "config", + "validate", + "--project", + str(project), + "--schemas", + str(SCHEMAS), + ] + ) + return exit_code, json.loads(capsys.readouterr().out) + + +def _refused( + project: Path, capsys: pytest.CaptureFixture[str] +) -> dict[str, object]: + exit_code, payload = _validate(project, capsys) + assert exit_code == 80 + assert payload["error"]["code"] == "CONFIG_INVALID" + context: dict[str, object] = payload["error"]["context"] + return context + + +def _static_route(path: str, directory: str = "web/dist", extra: str = "") -> str: + return ( + " routes:\n" + f" - path: {path}\n" + " static:\n" + " component: crm-backend\n" + f" directory: {directory}\n" + f"{extra}" + ) + + +def _proxy_route(path: str) -> str: + return ( + " routes:\n" + f" - path: {path}\n" + " upstream: { address: 127.0.0.1, port: 8310 }\n" + ) + + +def test_s003_1_proxy_and_spa_static_routes_pass_config_validate( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + project = _project(tmp_path, _SPA_ROUTES) + + exit_code, payload = _validate(project, capsys) + + assert exit_code == 0 + assert payload["data"]["status"] == "ok" + domain =PlatformInventory.from_state(validate_config(project, SCHEMAS)).domains[0] + assert domain.proxy.upstream_address is None + api, web = domain.routes + assert isinstance(api, DomainProxyRoute) + assert api.path == DomainRoutePath("/api/") + assert (api.upstream_address.value, api.upstream_port.value) == ("127.0.0.1", 8310) + assert isinstance(web, DomainStaticRoute) + assert web.component_id == ResourceId("crm-backend") + assert web.directory == ReleaseRelativePath("web/dist") + assert web.spa_fallback == ReleaseRelativePath("index.html") + serialized: dict[str, Any] = domain.as_dict() + assert "upstream" not in serialized["proxy"] + assert serialized["routes"] == [ + {"path": "/api/", "upstream": {"address": "127.0.0.1", "port": 8310}}, + { + "path": "/", + "static": { + "component": "crm-backend", + "directory": "web/dist", + "spaFallback": "index.html", + }, + }, + ] + + +def test_s003_1_a_domain_with_only_an_upstream_serializes_as_before() -> None: + domain = PlatformInventory.from_state(validate_config(EXAMPLES, SCHEMAS)).domains[0] + + assert domain.routes == () + assert "routes" not in domain.as_dict() + + +def test_s003_8_two_routes_for_one_path_are_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + routes = ( + " routes:\n" + " - path: /api/\n" + " upstream: { address: 127.0.0.1, port: 8310 }\n" + " - path: /api/\n" + " upstream: { address: 127.0.0.1, port: 8311 }\n" + ) + + context = _refused(_project(tmp_path, routes), capsys) + + assert context["code"] == "domain_route_path_duplicate" + assert context["field"] == "spec.routes.1.path" + + +def test_s003_9_a_domain_with_upstream_and_routes_is_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + context = _refused(_project(tmp_path, _SPA_ROUTES, keep_upstream=True), capsys) + + assert context["code"] == "domain_route_source_invalid" + assert context["field"] == "spec.routes" + + +def test_s003_9_a_domain_with_neither_upstream_nor_routes_still_passes( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + exit_code, payload = _validate(_project(tmp_path, ""), capsys) + + assert exit_code == 0 + assert payload["data"]["status"] == "ok" + + +@pytest.mark.parametrize( + "path", + [ + "/.well-known/acme-challenge", + "/.well-known/acme-challenge/", + "/.well-known/acme-challenge/token/", + ], +) +def test_s003_10_a_route_on_the_acme_challenge_is_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str], path: str +) -> None: + context = _refused(_project(tmp_path, _proxy_route(path)), capsys) + + assert context["code"] == "domain_route_path_reserved" + assert context["field"] == "spec.routes.0.path" + + +@pytest.mark.parametrize( + "routes", + [ + _static_route("/web"), + _static_route("'/a b/'"), + _static_route("'/a;b/'"), + _static_route("'/a{b/'"), + _proxy_route("api/"), + _proxy_route("'/a b'"), + _proxy_route("''"), + ], + ids=[ + "static-no-slash", + "space", + "semicolon", + "brace", + "relative", + "proxy-space", + "empty", + ], +) +def test_s003_10_an_invalid_route_path_is_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str], routes: str +) -> None: + context = _refused(_project(tmp_path, routes), capsys) + + assert context["code"] == "domain_route_path_invalid" + assert context["field"] == "spec.routes.0.path" + + +def test_s003_10_a_proxy_route_path_needs_no_trailing_slash( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + routes = _proxy_route("/api.v1_~x-y") + + exit_code, _ = _validate(_project(tmp_path, routes), capsys) + + assert exit_code == 0 + + +def test_s003_11_a_static_route_naming_a_missing_component_is_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + routes = _static_route("/").replace("crm-backend", "crm-missing") + + context = _refused(_project(tmp_path, routes), capsys) + + assert context["code"] == "resource_reference_missing" + assert context["field"] == "spec.routes.0.static.component" + assert "Component/crm-missing" in str(context["message"]) + + +def test_s003_11_a_static_component_off_the_proxy_server_is_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + project = _project(tmp_path, _static_route("/")) + component = project / "components" / "crm-backend.yaml" + text = component.read_text(encoding="utf-8") + component.write_text( + text.replace(" - h1\n - h2\n", " - h1\n"), encoding="utf-8" + ) + + context = _refused(project, capsys) + + assert context["code"] == "domain_route_component_not_on_proxy" + assert context["field"] == "spec.routes.0.static.component" + assert "h2" in str(context["message"]) + + +@pytest.mark.parametrize( + "routes", + [ + _static_route("/", directory="../web"), + _static_route("/", directory="web/../../etc"), + _static_route("/", directory="/srv/web"), + _static_route("/", directory="."), + _static_route("/", directory="./"), + _static_route("/", directory="'web dist'"), + _static_route("/", directory='"web\\n"'), + _static_route("/", extra=" spaFallback: ../index.html\n"), + _static_route("/", extra=" spaFallback: /index.html\n"), + _static_route("/", extra=" spaFallback: a/../../index.html\n"), + _static_route("/", extra=' spaFallback: "index.html\\n"\n'), + ], + ids=[ + "directory-parent", + "directory-inner-parent", + "directory-absolute", + "directory-root", + "directory-root-slash", + "directory-space", + "directory-trailing-newline", + "fallback-parent", + "fallback-absolute", + "fallback-inner-parent", + "fallback-trailing-newline", + ], +) +def test_s003_12_a_static_directory_or_fallback_outside_the_release_is_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str], routes: str +) -> None: + context = _refused(_project(tmp_path, routes), capsys) + + assert context["code"] == "schema_validation_failed" + assert str(context["field"]).startswith("spec.routes.0") + + +def test_s003_12_route_value_objects_match_the_schema() -> None: + for path in ("/", "/api/", "/a.b_c~d-e/f"): + assert DomainRoutePath.from_boundary(path).value == path + for path in ("", "api/", "/a b", "/a;b", "/a{b}", "/a\nb"): + with pytest.raises(ValueError, match="invalid route path"): + DomainRoutePath(path) + assert DomainRoutePath("/.well-known/acme-challenge/x").is_reserved + assert not DomainRoutePath("/.well-known/").is_reserved + assert not DomainRoutePath("/.well-known/acme-challenger").is_reserved + for value in ("web/dist", "index.html", "...", "a/./b"): + assert ReleaseRelativePath.from_boundary(value).value == value + for value in ("", "/web", "..", "a/../b", "a b", "a;b"): + with pytest.raises(ValueError, match="invalid release path"): + ReleaseRelativePath(value) + for root in (".", "./", "././", ".//."): + assert ReleaseRelativePath(root).is_release_root + assert not ReleaseRelativePath("web").is_release_root From 05f2a16c9190fd5433928bd38a29cf9b3c355794 Mon Sep 17 00:00:00 2001 From: "shipmill-romamo[bot]" <338557544+shipmill-romamo[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:07:04 +0200 Subject: [PATCH 4/4] Refuse a data copy the target can't hold before dumping it A preflight on the target host, after the empty-target check and before pg_dump, reads the source and target catalogs and names every schema the source lacks, extension and policy role the target lacks, and auth. reference, taking every name only as a psql variable. Any finding refuses the copy with nothing dumped: data copy and migrate's data step exit PRECONDITION with data_copy_preflight_refused and the four lists, read whole from a findings file under --data-migrations. pg_dump takes the needed extensions with named schemas, and pg_restore runs in one transaction, so a failed restore leaves no table and no receipt. Spec 007 is built (S-007-6 to S-007-11). Fixes #125 Co-Authored-By: Claude Opus 5.5 --- changelog.d/125-data-copy-preflight.md | 3 + docs/neon-migration-guide.md | 13 +- docs/render-migration-guide.md | 7 +- docs/specs/007-data-copy-schemas.md | 19 +- .../defaults/main.yml | 167 ++++++++++ .../meta/argument_specs.yml | 21 +- .../cloudfall_data_migration/tasks/main.yml | 125 ++++++- engine/tests/test_data_migration_role.py | 306 +++++++++++++++++- sdk/src/cloudfall/app.py | 37 ++- sdk/src/cloudfall/lifecycle.py | 131 +++++++- sdk/tests/test_data_copy_preflight.py | 278 ++++++++++++++++ 11 files changed, 1074 insertions(+), 33 deletions(-) create mode 100644 changelog.d/125-data-copy-preflight.md create mode 100644 sdk/tests/test_data_copy_preflight.py diff --git a/changelog.d/125-data-copy-preflight.md b/changelog.d/125-data-copy-preflight.md new file mode 100644 index 0000000..2010786 --- /dev/null +++ b/changelog.d/125-data-copy-preflight.md @@ -0,0 +1,3 @@ +### Added + +- `cloudfall data copy`, and the `data:` step of `cloudfall migrate`, run a preflight on the target host after the empty-target check and before `pg_dump`: it reads both catalogs and refuses a copy the target can't hold, with nothing dumped, the target untouched and no receipt, exiting `PRECONDITION` with code `data_copy_preflight_refused` and four lists in the error context, each item named: `missingSchemas` (a named schema the source lacks), `extensions` (one the selected schemas need, through `pg_depend` or by living in them, or every source extension but `plpgsql` without `--source-schema`, that the target's `pg_available_extensions` lacks), `roles` (a role a row-level security policy on a selected table names that the target's `pg_roles` lacks) and `authReferences` (a function or policy of the selected schemas that references `auth.` when `auth` isn't copied). The role writes the findings to `-.preflight.json` under `--data-migrations`, so the lists arrive whole. With `--source-schema`, `pg_dump` also gets one `--extension` per extension the schemas need (pg_dump 14 or later), and `pg_restore` now runs with `--single-transaction`, so a failed restore rolls back whole, writes no receipt, exits `ENGINE_STEP_FAILED`, and the next run passes the empty-target check (#125) diff --git a/docs/neon-migration-guide.md b/docs/neon-migration-guide.md index e617441..04dd0d9 100644 --- a/docs/neon-migration-guide.md +++ b/docs/neon-migration-guide.md @@ -175,7 +175,18 @@ the peer-authenticated socket as the declared owner, verifies per-table row counts match the source exactly, removes the staged URL file and the dump, and writes a `DataMigrationReceipt` with per-table counts. It refuses a target database that already has tables in the copied schemas, -so a re-run requires deliberately dropping what was restored. +so a re-run after a finished copy requires deliberately dropping what was +restored. A restore that fails runs in one transaction and rolls back +whole, so the next run needs no cleanup. + +Before it dumps anything, a preflight reads both catalogs and refuses a +copy the target can't hold, with `PRECONDITION` and the code +`data_copy_preflight_refused`, naming every finding at once: +`missingSchemas` (a named schema the source lacks), `extensions` (one the +copied schemas need that the target's `pg_available_extensions` lacks), +`roles` (a role a row-level security policy names that the target lacks), +and `authReferences` (a function or policy that calls `auth.` when `auth` +isn't copied). Provide what it names, or rewrite it, and run again. By default the copy takes every non-system schema of the source, and the row-count check and the receipt cover all of them. To take only some, diff --git a/docs/render-migration-guide.md b/docs/render-migration-guide.md index 4b6a9e1..23bceaf 100644 --- a/docs/render-migration-guide.md +++ b/docs/render-migration-guide.md @@ -147,8 +147,11 @@ uv run cloudfall migrate --build acme-api=main --data acme=~/.render/acme-db.url ``` Without `--yes` the command shows the plan. The engine dumps on the target -host, restores over the peer-authenticated socket, refuses a target -database that already has tables in the copied schemas, verifies per-table +host, restores over the peer-authenticated socket in one transaction, +refuses a target database that already has tables in the copied schemas, +refuses before the dump a copy whose schemas, extensions, policy roles, +or `auth.` references the target can't hold (`data_copy_preflight_refused`, +each one named), verifies per-table row counts, and writes a `DataMigrationReceipt`. The row-count check is the verify step: the migration is not done because the restore exited zero, it is done because the counts match. The copy takes every non-system schema of diff --git a/docs/specs/007-data-copy-schemas.md b/docs/specs/007-data-copy-schemas.md index e709e3b..ad80f34 100644 --- a/docs/specs/007-data-copy-schemas.md +++ b/docs/specs/007-data-copy-schemas.md @@ -1,6 +1,6 @@ # S-007: Data copy schemas -status: approved +status: built serves: G-3 ## Problem @@ -134,3 +134,20 @@ copy. - cloudfall-dev/cloudfall#126: S-007-12, S-007-13 ## Verification + +No criterion has run against a live PostgreSQL yet: the role's tasks and queries were rendered with Ansible's own templating and psql's variable quoting replayed, the queries were parsed with PostgreSQL's parser (libpg_query, through pglast) but not executed, and the SDK side ran with a stand-in for the playbook that writes what the role writes. What a real `pg_dump --schema --extension`, the catalog joins of the preflight, and a rolled-back `pg_restore --single-transaction` do on a server is unexercised. + +- S-007-1: `test_s007_1_named_schemas_reach_pg_dump_and_the_receipt`, `test_s007_1_a_schema_name_stays_one_quoted_pg_dump_argument` and `test_s007_1_data_migration_plan_passes_the_named_schemas_sorted`, the dump got one quoted `--schema` per name and the rendered receipt `spec.schemas` was `["app", "public"]` +- S-007-2: `test_s007_2_without_schema_the_dump_is_the_whole_database_as_in_0_7_0`, `test_s007_2_source_schemas_query_skips_only_the_system_schemas` and `test_s007_2_data_migration_plan_without_schemas_is_unchanged`, the dump command and extra vars were 0.7.0's and `spec.schemas` was `["app", "audit", "public"]` +- S-007-3: `test_s007_3_receipt_tables_carry_schema_table_and_rows`, every rendered entry had `schema`, `table` and `rows`, a quoted name with a comma included +- S-007-4: `test_s007_4_row_count_query_reads_only_the_selected_schemas` and `test_s007_4_both_sides_pass_the_schemas_to_the_row_count_query`, the schemas reached the query only as the psql variable, ordered by schema and table +- S-007-5: `test_s007_5_empty_target_check_counts_only_the_selected_schemas`, the count read only the selected schemas and a count of 3 failed naming "already contains 3 tables" +- S-007-6: `test_s007_6_a_missing_schema_is_refused_before_the_dump`, `test_s007_6_the_source_preflight_takes_names_only_as_psql_variables` and `test_s007_6_a_missing_schema_is_refused_with_its_name`, the refusal comes before `pg_dump` and answered `data_copy_preflight_refused` with `missingSchemas` `["missing"]` +- S-007-7: `test_s007_7_an_extension_the_target_lacks_is_refused`, `test_s007_7_named_schemas_dump_their_needed_extensions` and `test_s007_7_an_extension_the_target_lacks_is_named`, the target query read `pg_available_extensions` from a psql variable, `extensions` named `citext`, no receipt was written +- S-007-8: `test_s007_8_policy_roles_the_target_lacks_are_refused` and `test_s007_8_policy_roles_the_target_lacks_are_named`, `roles` was `["anon", "authenticated"]` +- S-007-9: `test_s007_9_auth_references_name_the_function_and_the_policy` and `test_s007_9_auth_references_are_named`, both `function public.current_tenant()` and `policy notes_owner on public.notes` were named and the copy refused +- S-007-10: `test_s007_10_every_finding_is_named_in_one_answer`, `test_s007_10_every_finding_comes_whole_in_one_answer` (400 role names, past the 2000-character output tail) and `test_s007_10_the_migrate_data_step_surfaces_the_same_refusal` (`migrate` answers `PRECONDITION` with the same lists) +- S-007-11: `test_s007_11_the_restore_is_one_transaction_and_the_receipt_follows_it`, `test_s007_11_a_failed_restore_is_an_engine_failure_without_a_receipt`, `test_s007_11_a_stale_findings_file_never_answers_for_a_new_run` and `test_s007_11_a_failed_migrate_data_step_stays_an_engine_failure`, `pg_restore` runs `--single-transaction`, the receipt task follows the block, a failure answered `ENGINE_STEP_FAILED` and the next run went ahead +- S-007-12: `test_s007_12_data_copy_refuses_a_malformed_or_repeated_schema`, `test_s007_12_cli_migrate_refuses_a_bad_data_schema_before_any_playbook`, `test_s007_12_migrate_refuses_a_data_schema_without_its_data` and `test_s007_12_the_data_tools_refuse_a_bad_schema_as_the_cli_does` (over MCP), each exited `PRECONDITION` with no inventory render and no plan file; a value holding a newline is refused earlier by the CLI framework as `ARG_ERROR` +- S-007-13: `test_s007_13_migrate_copies_only_the_data_schema_as_data_copy_does` and `test_s007_13_migrate_groups_each_data_schema_under_its_database`, the `data:crm` step's playbook argv equalled `data copy --source-schema public`'s +- S-007-14: `test_s007_14_data_copy_plan_names_the_selected_schemas`, the plan said "dump schemas app, public" and, without a name, "dump every non-system schema" diff --git a/engine/ansible/roles/cloudfall_data_migration/defaults/main.yml b/engine/ansible/roles/cloudfall_data_migration/defaults/main.yml index 8ba838b..1b84604 100644 --- a/engine/ansible/roles/cloudfall_data_migration/defaults/main.yml +++ b/engine/ansible/roles/cloudfall_data_migration/defaults/main.yml @@ -19,6 +19,173 @@ cloudfall_data_migration_target_tables_query: >- SELECT count(*) FROM information_schema.tables WHERE table_schema IN (SELECT jsonb_array_elements_text(:'cloudfall_data_schemas'::jsonb)) +# What the source must give the target before anything is dumped, as one +# JSON object line. The selected schemas arrive as the psql variable +# cloudfall_data_schemas and whether none was named as +# cloudfall_data_every_schema, so no name is ever spliced into the SQL text. +# missingSchemas: a selected schema the source doesn't have +# extensions: an extension the selected schemas need: one installed in a +# selected schema, one whose member object an object of a selected schema +# depends on in pg_depend (a column type, a default, an index operator +# class, a view, a trigger, a policy), or, with no schema named, every +# extension but plpgsql, since the whole dump creates them all +# policyRoles: a role a row-level security policy on a table of a selected +# schema names (pg_policy.polroles; 0 is PUBLIC) +# authReferences: when auth isn't selected, a function of a selected schema +# whose source mentions auth. or that depends on an object in auth, and a +# policy on a table of a selected schema whose USING or WITH CHECK does +cloudfall_data_migration_source_preflight_query: >- + WITH selected AS ( + SELECT jsonb_array_elements_text(:'cloudfall_data_schemas'::jsonb) + AS nspname + ), + selected_ns AS ( + SELECT n.oid, n.nspname + FROM pg_catalog.pg_namespace n JOIN selected s USING (nspname) + ), + auth_selected AS ( + SELECT EXISTS (SELECT 1 FROM selected WHERE nspname = 'auth') AS chosen + ), + selected_objects AS ( + SELECT 'pg_catalog.pg_class'::regclass AS classid, c.oid AS objid + FROM pg_catalog.pg_class c + WHERE c.relnamespace IN (SELECT oid FROM selected_ns) + UNION ALL + SELECT 'pg_catalog.pg_type'::regclass, t.oid FROM pg_catalog.pg_type t + WHERE t.typnamespace IN (SELECT oid FROM selected_ns) + UNION ALL + SELECT 'pg_catalog.pg_proc'::regclass, p.oid FROM pg_catalog.pg_proc p + WHERE p.pronamespace IN (SELECT oid FROM selected_ns) + UNION ALL + SELECT 'pg_catalog.pg_constraint'::regclass, k.oid + FROM pg_catalog.pg_constraint k + WHERE k.connamespace IN (SELECT oid FROM selected_ns) + UNION ALL + SELECT 'pg_catalog.pg_attrdef'::regclass, a.oid + FROM pg_catalog.pg_attrdef a JOIN pg_catalog.pg_class c ON c.oid = a.adrelid + WHERE c.relnamespace IN (SELECT oid FROM selected_ns) + UNION ALL + SELECT 'pg_catalog.pg_trigger'::regclass, g.oid + FROM pg_catalog.pg_trigger g JOIN pg_catalog.pg_class c ON c.oid = g.tgrelid + WHERE c.relnamespace IN (SELECT oid FROM selected_ns) + UNION ALL + SELECT 'pg_catalog.pg_rewrite'::regclass, r.oid + FROM pg_catalog.pg_rewrite r JOIN pg_catalog.pg_class c ON c.oid = r.ev_class + WHERE c.relnamespace IN (SELECT oid FROM selected_ns) + UNION ALL + SELECT 'pg_catalog.pg_policy'::regclass, y.oid + FROM pg_catalog.pg_policy y JOIN pg_catalog.pg_class c ON c.oid = y.polrelid + WHERE c.relnamespace IN (SELECT oid FROM selected_ns) + ), + members AS ( + SELECT d.classid, d.objid, d.refobjid AS extoid + FROM pg_catalog.pg_depend d + WHERE d.refclassid = 'pg_catalog.pg_extension'::regclass + AND d.deptype = 'e' + ), + needed AS ( + SELECT e.extname FROM pg_catalog.pg_extension e + WHERE (:'cloudfall_data_every_schema'::boolean AND e.extname <> 'plpgsql') + OR e.extnamespace IN (SELECT oid FROM selected_ns) + OR e.oid IN ( + SELECT m.extoid + FROM pg_catalog.pg_depend d + JOIN selected_objects o ON o.classid = d.classid AND o.objid = d.objid + JOIN members m ON m.classid = d.refclassid AND m.objid = d.refobjid + UNION + SELECT d.refobjid + FROM pg_catalog.pg_depend d + JOIN selected_objects o ON o.classid = d.classid AND o.objid = d.objid + WHERE d.refclassid = 'pg_catalog.pg_extension'::regclass + ) + ), + missing AS ( + SELECT s.nspname FROM selected s + WHERE NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_namespace n WHERE n.nspname = s.nspname + ) + ), + policy_roles AS ( + SELECT DISTINCT r.rolname + FROM pg_catalog.pg_policy y + JOIN pg_catalog.pg_class c ON c.oid = y.polrelid + CROSS JOIN LATERAL unnest(y.polroles) AS named(role_oid) + JOIN pg_catalog.pg_roles r ON r.oid = named.role_oid + WHERE c.relnamespace IN (SELECT oid FROM selected_ns) + ), + auth_ns AS ( + SELECT oid FROM pg_catalog.pg_namespace WHERE nspname = 'auth' + ), + auth_references AS ( + SELECT format('function %I.%I(%s)', n.nspname, p.proname, + pg_catalog.pg_get_function_identity_arguments(p.oid)) + AS reference + FROM pg_catalog.pg_proc p JOIN selected_ns n ON n.oid = p.pronamespace + WHERE p.prosrc ~ '\mauth\.' + OR EXISTS ( + SELECT 1 FROM pg_catalog.pg_depend d + LEFT JOIN pg_catalog.pg_proc dp + ON d.refclassid = 'pg_catalog.pg_proc'::regclass + AND dp.oid = d.refobjid + LEFT JOIN pg_catalog.pg_class dc + ON d.refclassid = 'pg_catalog.pg_class'::regclass + AND dc.oid = d.refobjid + LEFT JOIN pg_catalog.pg_type dt + ON d.refclassid = 'pg_catalog.pg_type'::regclass + AND dt.oid = d.refobjid + WHERE d.classid = 'pg_catalog.pg_proc'::regclass AND d.objid = p.oid + AND coalesce(dp.pronamespace, dc.relnamespace, dt.typnamespace) + IN (SELECT oid FROM auth_ns) + ) + UNION + SELECT format('policy %I on %I.%I', y.polname, n.nspname, c.relname) + FROM pg_catalog.pg_policy y + JOIN pg_catalog.pg_class c ON c.oid = y.polrelid + JOIN selected_ns n ON n.oid = c.relnamespace + WHERE coalesce(pg_catalog.pg_get_expr(y.polqual, y.polrelid), '') + ~ '\mauth\.' + OR coalesce(pg_catalog.pg_get_expr(y.polwithcheck, y.polrelid), '') + ~ '\mauth\.' + ) + SELECT json_build_object( + 'missingSchemas', + coalesce((SELECT json_agg(nspname ORDER BY nspname) FROM missing), '[]'), + 'extensions', + coalesce((SELECT json_agg(extname::text ORDER BY extname) FROM needed), + '[]'), + 'policyRoles', + coalesce((SELECT json_agg(rolname::text ORDER BY rolname) + FROM policy_roles), '[]'), + 'authReferences', + CASE WHEN (SELECT chosen FROM auth_selected) THEN '[]'::json + ELSE coalesce((SELECT json_agg(reference ORDER BY reference) + FROM auth_references), '[]') + END + ) +# Which of the needed extensions and policy roles the target lacks, as one +# JSON object line. Both lists arrive as the psql variables +# cloudfall_data_extensions and cloudfall_data_policy_roles. +cloudfall_data_migration_target_preflight_query: >- + SELECT json_build_object( + 'extensions', + coalesce(( + SELECT json_agg(wanted.value ORDER BY wanted.value) + FROM jsonb_array_elements_text(:'cloudfall_data_extensions'::jsonb) + AS wanted(value) + WHERE wanted.value NOT IN ( + SELECT a.name::text FROM pg_catalog.pg_available_extensions a + ) + ), '[]'), + 'roles', + coalesce(( + SELECT json_agg(wanted.value ORDER BY wanted.value) + FROM jsonb_array_elements_text(:'cloudfall_data_policy_roles'::jsonb) + AS wanted(value) + WHERE wanted.value NOT IN ( + SELECT r.rolname::text FROM pg_catalog.pg_roles r + ) + ), '[]') + ) # Exact per-table row counts of the selected schemas in one deterministic # query, one schema,table,rows line per base table, so source and target # evidence can be compared verbatim. The schema names arrive as the psql diff --git a/engine/ansible/roles/cloudfall_data_migration/meta/argument_specs.yml b/engine/ansible/roles/cloudfall_data_migration/meta/argument_specs.yml index a55b2b1..b0ddc34 100644 --- a/engine/ansible/roles/cloudfall_data_migration/meta/argument_specs.yml +++ b/engine/ansible/roles/cloudfall_data_migration/meta/argument_specs.yml @@ -29,7 +29,10 @@ argument_specs: type: str required: false default: "" - description: Controller directory receiving the migration receipt + description: >- + Controller directory receiving the migration receipt, and before + the dump the preflight findings file + -.preflight.json cloudfall_data_migration_url_path: type: path required: false @@ -51,6 +54,22 @@ argument_specs: description: >- Query returning every non-system schema of the source as one JSON array, used when no schema is named + cloudfall_data_migration_source_preflight_query: + type: str + required: false + description: >- + Query returning, as one JSON object, the named schemas the source + lacks, the extensions and policy roles the selected schemas need, + and their auth. references; it reads the psql variables + cloudfall_data_schemas and cloudfall_data_every_schema + cloudfall_data_migration_target_preflight_query: + type: str + required: false + description: >- + Query returning, as one JSON object, which needed extensions the + target's pg_available_extensions and which policy roles its pg_roles + lack; it reads the psql variables cloudfall_data_extensions and + cloudfall_data_policy_roles cloudfall_data_migration_target_tables_query: type: str required: false diff --git a/engine/ansible/roles/cloudfall_data_migration/tasks/main.yml b/engine/ansible/roles/cloudfall_data_migration/tasks/main.yml index 6a019c7..982097b 100644 --- a/engine/ansible/roles/cloudfall_data_migration/tasks/main.yml +++ b/engine/ansible/roles/cloudfall_data_migration/tasks/main.yml @@ -28,6 +28,16 @@ cloudfall_data_migration_owner: >- {{ cloudfall_data_migration_entries[0].owner }} +# The preflight findings land here before the dump, the receipt after it. +- name: Ensure the controller receipt directory exists + ansible.builtin.file: + path: "{{ cloudfall_data_migration_receipt_directory }}" + state: directory + mode: "0755" + delegate_to: localhost + become: false + when: cloudfall_data_migration_receipt_directory | length > 0 + - name: Migrate and verify the declared database block: - name: Stage the source connection URL on the target host @@ -97,6 +107,101 @@ {{ cloudfall_data_migration_target_tables.stdout | trim }} tables; data migration refuses to overwrite existing data + - name: Read what the selected schemas need from the source + ansible.builtin.shell: + cmd: >- + psql + --dbname "$(cat {{ cloudfall_data_migration_url_path | quote }})" + --no-psqlrc --tuples-only --no-align + --set ON_ERROR_STOP=1 + --set {{ cloudfall_data_migration_schemas_variable | quote }} + --set {{ + ('cloudfall_data_every_schema=' + ~ (cloudfall_data_migration_schemas | length == 0) | string + | lower) | quote + }} + stdin: "{{ cloudfall_data_migration_source_preflight_query }};" + register: cloudfall_data_migration_source_preflight + changed_when: false + + - name: Keep what the source needs + ansible.builtin.set_fact: + cloudfall_data_migration_needs: >- + {{ cloudfall_data_migration_source_preflight.stdout | from_json }} + + - name: Read which needed extensions and policy roles the target lacks + ansible.builtin.command: + argv: + - /usr/bin/psql + - --dbname + - "{{ cloudfall_data_migration_database }}" + - --no-psqlrc + - --tuples-only + - --no-align + - --set + - ON_ERROR_STOP=1 + - --set + - >- + cloudfall_data_extensions={{ + cloudfall_data_migration_needs.extensions | to_json + }} + - --set + - >- + cloudfall_data_policy_roles={{ + cloudfall_data_migration_needs.policyRoles | to_json + }} + stdin: "{{ cloudfall_data_migration_target_preflight_query }};" + become: true + become_user: postgres + register: cloudfall_data_migration_target_preflight + changed_when: false + + - name: Collect every preflight finding + ansible.builtin.set_fact: + cloudfall_data_migration_findings: + missingSchemas: "{{ cloudfall_data_migration_needs.missingSchemas }}" + extensions: >- + {{ + (cloudfall_data_migration_target_preflight.stdout + | from_json).extensions + }} + roles: >- + {{ + (cloudfall_data_migration_target_preflight.stdout + | from_json).roles + }} + authReferences: "{{ cloudfall_data_migration_needs.authReferences }}" + + # The SDK reads the whole lists from this file, not from the playbook's + # output, which it keeps only the tail of. + - name: Write the preflight findings on the controller + ansible.builtin.copy: + content: "{{ cloudfall_data_migration_findings | to_nice_json }}" + dest: >- + {{ cloudfall_data_migration_receipt_directory }}/{{ + cloudfall_data_migration_service.id + }}-{{ cloudfall_data_migration_database }}.preflight.json + mode: "0644" + delegate_to: localhost + become: false + when: cloudfall_data_migration_receipt_directory | length > 0 + + - name: Refuse a copy the target can't hold + ansible.builtin.assert: + that: + - cloudfall_data_migration_findings.missingSchemas | length == 0 + - cloudfall_data_migration_findings.extensions | length == 0 + - cloudfall_data_migration_findings.roles | length == 0 + - cloudfall_data_migration_findings.authReferences | length == 0 + fail_msg: >- + The data copy preflight refused the copy; nothing was dumped and + the target is untouched. Missing schemas + {{ cloudfall_data_migration_findings.missingSchemas }}, extensions + the target lacks {{ cloudfall_data_migration_findings.extensions }}, + policy roles the target lacks + {{ cloudfall_data_migration_findings.roles }}, auth. references + {{ cloudfall_data_migration_findings.authReferences }} + - name: Dump the source database on the target host ansible.builtin.shell: cmd: >- @@ -107,7 +212,13 @@ for schema in cloudfall_data_migration_schemas | unique | sort %} --schema {{ ('"' ~ (schema | replace('"', '""')) ~ '"') | quote - }}{% endfor %} + }}{% endfor %}{% + if cloudfall_data_migration_schemas | length > 0 + %}{% + for extension in cloudfall_data_migration_needs.extensions + %} --extension {{ + ('"' ~ (extension | replace('"', '""')) ~ '"') | quote + }}{% endfor %}{% endif %} register: cloudfall_data_migration_dump retries: 3 delay: 10 @@ -142,10 +253,13 @@ {{ cloudfall_data_migration_selected_schemas | join(', ') }}; there is nothing to migrate + # One transaction, which also stops at the first error: a failed restore + # rolls back whole, so a re-run passes the empty-target check. - name: Restore the dump into the declared database ansible.builtin.command: argv: - /usr/bin/pg_restore + - --single-transaction - --no-owner - --no-privileges - --role @@ -223,15 +337,6 @@ path: "{{ cloudfall_data_migration_dump_path }}" state: absent -- name: Ensure the controller receipt directory exists - ansible.builtin.file: - path: "{{ cloudfall_data_migration_receipt_directory }}" - state: directory - mode: "0755" - delegate_to: localhost - become: false - when: cloudfall_data_migration_receipt_directory | length > 0 - - name: Write the data-migration receipt on the controller ansible.builtin.copy: content: >- diff --git a/engine/tests/test_data_migration_role.py b/engine/tests/test_data_migration_role.py index 33ded94..40613f7 100644 --- a/engine/tests/test_data_migration_role.py +++ b/engine/tests/test_data_migration_role.py @@ -61,6 +61,8 @@ def _trusted(value: object) -> object: return trust_as_template(value) if isinstance(value, list): return [_trusted(item) for item in value] + if isinstance(value, dict): + return {key: _trusted(item) for key, item in value.items()} return value @@ -110,7 +112,16 @@ def _dump_command(named: list[str]) -> str: shell = _module( "Dump the source database on the target host", "ansible.builtin.shell" ) - return str(_render(shell["cmd"], {"cloudfall_data_migration_schemas": named})) + return str( + _render( + shell["cmd"], + { + "cloudfall_data_migration_schemas": named, + # The preflight found the schemas need no extension. + "cloudfall_data_migration_needs": {"extensions": []}, + }, + ) + ) def _psql_substitute(sql: str, variable: str) -> str: @@ -288,6 +299,299 @@ def test_s007_5_empty_target_check_counts_only_the_selected_schemas() -> None: assert "already contains 3 tables" in str(message) +NO_FINDINGS: dict[str, list[str]] = { + "missingSchemas": [], + "extensions": [], + "roles": [], + "authReferences": [], +} + + +def _task_names() -> list[str]: + return [str(task.get("name")) for task in _tasks()] + + +def _before(first: str, second: str) -> bool: + names = _task_names() + return names.index(first) < names.index(second) + + +def _findings( + needs: dict[str, list[str]], target: dict[str, list[str]] +) -> dict[str, object]: + """Replay how the role merges the source and target catalog answers.""" + fact = _module("Collect every preflight finding", "ansible.builtin.set_fact") + rendered = _render( + fact["cloudfall_data_migration_findings"], + { + "cloudfall_data_migration_needs": needs, + "cloudfall_data_migration_target_preflight": { + "stdout": json.dumps(target) + }, + }, + ) + assert isinstance(rendered, dict) + return rendered + + +def _refuses(findings: dict[str, object]) -> str | None: + """Return the refusal message, or None when the preflight passes.""" + refuse = _module("Refuse a copy the target can't hold", "ansible.builtin.assert") + conditions = refuse["that"] + assert isinstance(conditions, list) + variables: dict[str, object] = {"cloudfall_data_migration_findings": findings} + if all(_holds(condition, variables) for condition in conditions): + return None + return str(_render(refuse["fail_msg"], variables)) + + +def _findings_file(findings: dict[str, object]) -> tuple[str, object]: + write = _module( + "Write the preflight findings on the controller", "ansible.builtin.copy" + ) + variables: dict[str, object] = { + "cloudfall_data_migration_findings": findings, + "cloudfall_data_migration_receipt_directory": "/srv/fleet/tmp/data-migrations", + "cloudfall_data_migration_service": {"id": "postgresql-main"}, + "cloudfall_data_migration_database": "crm", + } + destination = str(_render(write["dest"], variables)) + return destination, json.loads(str(_render(write["content"], variables))) + + +def _source_preflight_sql(schemas: list[str], *, every_schema: bool) -> str: + query = str(_defaults()["cloudfall_data_migration_source_preflight_query"]) + query = _psql_substitute(query, _schemas_variable(schemas)) + return _psql_substitute( + query, f"cloudfall_data_every_schema={str(every_schema).lower()}" + ) + + +def _source_preflight_command(named: list[str]) -> str: + shell = _module( + "Read what the selected schemas need from the source", + "ansible.builtin.shell", + ) + return str( + _render( + shell["cmd"], + { + "cloudfall_data_migration_schemas": named, + "cloudfall_data_migration_schemas_variable": _schemas_variable( + named or ["app"] + ), + }, + ) + ) + + +def _target_preflight_argv(needs: dict[str, list[str]]) -> list[str]: + command = _module( + "Read which needed extensions and policy roles the target lacks", + "ansible.builtin.command", + ) + argv = _render( + command["argv"], + { + "cloudfall_data_migration_needs": needs, + "cloudfall_data_migration_database": "crm", + }, + ) + assert isinstance(argv, list) + return [str(argument) for argument in argv] + + +def test_s007_6_a_missing_schema_is_refused_before_the_dump() -> None: + needs = {**NO_FINDINGS, "missingSchemas": ["missing"], "policyRoles": []} + findings = _findings(needs, {"extensions": [], "roles": []}) + destination, written = _findings_file(findings) + sql = _source_preflight_sql(["missing"], every_schema=False) + + assert findings["missingSchemas"] == ["missing"] + assert written == {**NO_FINDINGS, "missingSchemas": ["missing"]} + assert destination == ( + "/srv/fleet/tmp/data-migrations/postgresql-main-crm.preflight.json" + ) + assert "Missing schemas ['missing']" in str(_refuses(findings)) + # The preflight runs after the empty-target check, and refuses before + # pg_dump writes the dump file. + assert _before( + "Refuse to restore into a non-empty database", + "Read what the selected schemas need from the source", + ) + assert _before( + "Refuse a copy the target can't hold", + "Dump the source database on the target host", + ) + assert _before( + "Write the preflight findings on the controller", + "Refuse a copy the target can't hold", + ) + assert "jsonb_array_elements_text('[\"missing\"]'::jsonb)" in sql + assert "'missing'" not in sql + + +def test_s007_6_the_source_preflight_takes_names_only_as_psql_variables() -> None: + query = str(_defaults()["cloudfall_data_migration_source_preflight_query"]) + named = _source_preflight_command(['Bad;Name"$(touch x)']) + every = _source_preflight_command([]) + + assert ":'cloudfall_data_schemas'::jsonb" in query + assert ":'cloudfall_data_every_schema'::boolean" in query + assert "{{" not in query + assert "--set ON_ERROR_STOP=1" in named + assert "--set cloudfall_data_every_schema=false" in named + assert "--set cloudfall_data_every_schema=true" in every + assert "touch" not in named.replace( + "'cloudfall_data_schemas=[\"Bad;Name\\\"$(touch x)\"]'", "" + ) + assert '"$(cat /root/.cloudfall-data-source-url)"' in named + + +def test_s007_7_an_extension_the_target_lacks_is_refused() -> None: + needs = {**NO_FINDINGS, "extensions": ["citext", "pg_trgm"], "policyRoles": []} + argv = _target_preflight_argv(needs) + target = str(_defaults()["cloudfall_data_migration_target_preflight_query"]) + findings = _findings(needs, {"extensions": ["citext"], "roles": []}) + sql = _source_preflight_sql(["public"], every_schema=False) + + assert argv[-4:] == [ + "--set", + 'cloudfall_data_extensions=["citext", "pg_trgm"]', + "--set", + "cloudfall_data_policy_roles=[]", + ] + assert ":'cloudfall_data_extensions'::jsonb" in target + assert "pg_catalog.pg_available_extensions" in target + assert findings["extensions"] == ["citext"] + assert "extensions the target lacks ['citext']" in str(_refuses(findings)) + # A column's type is a member of the extension: two pg_depend hops. + assert "d.refclassid = 'pg_catalog.pg_extension'::regclass" in sql + assert "d.deptype = 'e'" in sql + assert "e.extnamespace IN (SELECT oid FROM selected_ns)" in sql + assert "e.extname <> 'plpgsql'" in sql + + +def test_s007_7_named_schemas_dump_their_needed_extensions() -> None: + shell = _module( + "Dump the source database on the target host", "ansible.builtin.shell" + ) + needs = {**NO_FINDINGS, "extensions": ["citext", "uuid-ossp"]} + + named = _render( + shell["cmd"], + { + "cloudfall_data_migration_schemas": ["public"], + "cloudfall_data_migration_needs": needs, + }, + ) + every = _render( + shell["cmd"], + { + "cloudfall_data_migration_schemas": [], + "cloudfall_data_migration_needs": needs, + }, + ) + + assert named == ( + f"{WHOLE_DATABASE_DUMP} --schema '\"public\"' " + "--extension '\"citext\"' --extension '\"uuid-ossp\"'" + ) + assert every == WHOLE_DATABASE_DUMP + # An extension name read from the source stays one quoted shell word. + hostile = _render( + shell["cmd"], + { + "cloudfall_data_migration_schemas": ["public"], + "cloudfall_data_migration_needs": {"extensions": ['x";$(touch y)']}, + }, + ) + assert str(hostile).endswith("--extension '\"x\"\";$(touch y)\"'") + + +def test_s007_8_policy_roles_the_target_lacks_are_refused() -> None: + needs = {**NO_FINDINGS, "policyRoles": ["anon", "authenticated"]} + argv = _target_preflight_argv(needs) + target = str(_defaults()["cloudfall_data_migration_target_preflight_query"]) + sql = _source_preflight_sql(["public"], every_schema=False) + findings = _findings(needs, {"extensions": [], "roles": ["anon", "authenticated"]}) + + assert argv[-1] == 'cloudfall_data_policy_roles=["anon", "authenticated"]' + assert ":'cloudfall_data_policy_roles'::jsonb" in target + assert "pg_catalog.pg_roles" in target + assert "unnest(y.polroles)" in sql + assert findings["roles"] == ["anon", "authenticated"] + assert "policy roles the target lacks ['anon', 'authenticated']" in str( + _refuses(findings) + ) + + +def test_s007_9_auth_references_name_the_function_and_the_policy() -> None: + references = [ + "function public.current_tenant()", + "policy notes_owner on public.notes", + ] + needs = {**NO_FINDINGS, "authReferences": references, "policyRoles": []} + findings = _findings(needs, {"extensions": [], "roles": []}) + sql = _source_preflight_sql(["public"], every_schema=False) + + assert findings["authReferences"] == references + assert "auth. references" in str(_refuses(findings)) + assert "format('function %I.%I(%s)', n.nspname, p.proname," in sql + assert "format('policy %I on %I.%I', y.polname, n.nspname, c.relname)" in sql + assert "pg_get_expr(y.polqual, y.polrelid)" in sql + assert "pg_get_expr(y.polwithcheck, y.polrelid)" in sql + assert "p.prosrc ~ '\\mauth\\.'" in sql + # A selected auth schema copies auth too, so nothing is reported. + assert "SELECT EXISTS (SELECT 1 FROM selected WHERE nspname = 'auth')" in sql + + +def test_s007_10_every_finding_is_named_in_one_answer() -> None: + needs = { + "missingSchemas": [], + "extensions": ["postgis"], + "policyRoles": ["anon"], + "authReferences": ["function public.current_tenant()"], + } + findings = _findings(needs, {"extensions": ["postgis"], "roles": ["anon"]}) + _, written = _findings_file(findings) + message = str(_refuses(findings)) + + assert written == { + "missingSchemas": [], + "extensions": ["postgis"], + "roles": ["anon"], + "authReferences": ["function public.current_tenant()"], + } + for named in ("postgis", "anon", "public.current_tenant()"): + assert named in message + clean = _findings( + {**NO_FINDINGS, "policyRoles": []}, {"extensions": [], "roles": []} + ) + assert _refuses(clean) is None + + +def test_s007_11_the_restore_is_one_transaction_and_the_receipt_follows_it() -> None: + restore = _module( + "Restore the dump into the declared database", "ansible.builtin.command" + ) + argv = restore["argv"] + tasks = DataLoader().load_from_file(str(ROLE / "tasks" / "main.yml")) + assert isinstance(tasks, list) + top = [task.get("name") for task in tasks] + (block,) = [task for task in tasks if "block" in task] + + assert isinstance(argv, list) + assert argv[:2] == ["/usr/bin/pg_restore", "--single-transaction"] + # A failing block skips every task after it: no receipt is written. + assert top.index(block["name"]) < top.index( + "Write the data-migration receipt on the controller" + ) + assert "Restore the dump into the declared database" in [ + task["name"] for task in block["block"] + ] + + def test_the_data_playbook_passes_the_named_schemas_to_the_role() -> None: plays = DataLoader().load_from_file(str(PLAYBOOK)) assert isinstance(plays, list) diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index 2f95dcc..635f29f 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -164,6 +164,7 @@ investigation_events, ) from cloudfall.lifecycle import ( + ERROR_DATA_COPY_PREFLIGHT_REFUSED, ERROR_EXECUTION_FAILED, DataCopySchemas, DeployOptions, @@ -1825,7 +1826,7 @@ def run(argv: Sequence[str], environment: Mapping[str, str], diff: Path) -> int: def _lifecycle_failed(error: LifecycleError) -> Exception: - context = {"code": error.code} + context = {**error.context, "code": error.code} if error.code == ERROR_EXECUTION_FAILED: return Exit.ENGINE_STEP_FAILED(error.detail, context=context) return Exit.PRECONDITION(error.detail, context=context) @@ -3292,18 +3293,34 @@ def migrate(args: MigrateArgs, ctx: Ctx, fleet: Fleet) -> Migration: next=cast("str | None", result.get("next")), step=cast("str | None", result.get("step")), ) + failure = _migration_failure( + migration, cast("Mapping[str, object]", result.get("error", {})) + ) + if failure is not None: + raise failure + return migration + + +def _migration_failure( + migration: Migration, error: Mapping[str, object] +) -> CliExit | None: + """Return the exit a paused or failed run answers with, or None.""" # The step's own error goes in error.context, beside the plan in data. - cause = { - "step": migration.step, - **cast("Mapping[str, object]", result.get("error", {})), - } - if status == "paused": + cause = {"step": migration.step, **error} + if migration.status == "paused": message = f"the migration paused at {migration.step}; resume it with --yes" - raise Exit.PARTIAL_FAILURE(message, context=cause, data=migration) - if status == "error": + return Exit.PARTIAL_FAILURE(message, context=cause, data=migration) + if ( + migration.status == "error" + and cause.get("code") == ERROR_DATA_COPY_PREFLIGHT_REFUSED + ): + # The data step refuses as data copy does: nothing was dumped (S-007). + message = f"the migration stopped at {migration.step}: {cause['message']}" + return Exit.PRECONDITION(message, context=cause, data=migration) + if migration.status == "error": message = f"the migration failed at {migration.step}" - raise Exit.ENGINE_STEP_FAILED(message, context=cause, data=migration) - return migration + return Exit.ENGINE_STEP_FAILED(message, context=cause, data=migration) + return None # why diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index 13189b8..fd09e61 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -82,22 +82,30 @@ _ERROR_DATABASE_MISSING = "lifecycle_database_missing" _ERROR_SOURCE_URL_MISSING = "lifecycle_source_url_missing" ERROR_DATA_COPY_SCHEMA_INVALID = "data_copy_schema_invalid" +ERROR_DATA_COPY_PREFLIGHT_REFUSED = "data_copy_preflight_refused" +_ERROR_PREFLIGHT_FINDINGS_INVALID = "data_copy_preflight_findings_invalid" class LifecycleError(RuntimeError): """Fail-fast lifecycle error with a stable machine-readable code.""" - def __init__(self, code: str, message: str) -> None: - """Record the failure code and human-readable detail.""" + def __init__( + self, + code: str, + message: str, + context: Mapping[str, object] | None = None, + ) -> None: + """Record the failure code, its detail, and what the answer names.""" self.code = code self.detail = message + self.context: Mapping[str, object] = dict(context or {}) super().__init__(f"{code}: {message}") def as_dict(self) -> dict[str, object]: """Serialize the error envelope for system boundaries.""" return { "status": "error", - "error": {"code": self.code, "message": self.detail}, + "error": {**self.context, "code": self.code, "message": self.detail}, } @@ -311,6 +319,85 @@ def describe(self) -> str: return f"{noun} {', '.join(self.values)}" +_PREFLIGHT_KEYS = ("missingSchemas", "extensions", "roles", "authReferences") +_PREFLIGHT_LABELS = { + "missingSchemas": "schemas the source lacks", + "extensions": "extensions the target lacks", + "roles": "policy roles the target lacks", + "authReferences": "auth. references to rewrite", +} + + +@dataclass(frozen=True, slots=True) +class DataCopyPreflightFindings: + """What the data copy preflight found the target can't hold (S-007). + + The role writes these to ``-.preflight.json`` under + the receipt directory before it dumps anything, so the whole lists + reach the answer rather than the tail of the playbook's output. + """ + + missing_schemas: tuple[str, ...] + extensions: tuple[str, ...] + roles: tuple[str, ...] + auth_references: tuple[str, ...] + + @classmethod + def from_file(cls, path: Path) -> DataCopyPreflightFindings: + """Read the findings file strictly: four lists of names, nothing else.""" + try: + raw = cast("object", json.loads(path.read_text(encoding="utf-8"))) + except json.JSONDecodeError as error: + detail = f"preflight findings are not JSON: {path}" + raise LifecycleError(_ERROR_PREFLIGHT_FINDINGS_INVALID, detail) from error + if not isinstance(raw, dict) or set(raw) != set(_PREFLIGHT_KEYS): + keys = ", ".join(_PREFLIGHT_KEYS) + detail = f"preflight findings must hold exactly {keys}: {path}" + raise LifecycleError(_ERROR_PREFLIGHT_FINDINGS_INVALID, detail) + lists: dict[str, tuple[str, ...]] = {} + for key in _PREFLIGHT_KEYS: + value = cast("object", raw[key]) + if not isinstance(value, list) or not all( + isinstance(item, str) for item in cast("list[object]", value) + ): + detail = f"preflight findings {key} must be a list of names: {path}" + raise LifecycleError(_ERROR_PREFLIGHT_FINDINGS_INVALID, detail) + lists[key] = tuple(cast("list[str]", value)) + return cls( + missing_schemas=lists["missingSchemas"], + extensions=lists["extensions"], + roles=lists["roles"], + auth_references=lists["authReferences"], + ) + + @property + def refused(self) -> bool: + """Whether any finding stops the copy.""" + return any(self.as_dict().values()) + + def as_dict(self) -> dict[str, list[str]]: + """Return the four lists, keyed as the error context names them.""" + return { + "missingSchemas": list(self.missing_schemas), + "extensions": list(self.extensions), + "roles": list(self.roles), + "authReferences": list(self.auth_references), + } + + def refusal(self) -> LifecycleError: + """Return the ``data_copy_preflight_refused`` error naming every finding.""" + named = "; ".join( + f"{_PREFLIGHT_LABELS[key]}: {', '.join(values)}" + for key, values in self.as_dict().items() + if values + ) + detail = ( + "the data copy preflight refused the copy, so nothing was dumped " + f"and the target is untouched; {named}" + ) + return LifecycleError(ERROR_DATA_COPY_PREFLIGHT_REFUSED, detail, self.as_dict()) + + class PreviewSubject(StrEnum): """Kind of declared resource a previewed action targets.""" @@ -1346,12 +1433,40 @@ def migrate_data( # noqa: PLR0913 - as plan_data_migration. receipt_directory: Path | None = None, schemas: DataCopySchemas | None = None, ) -> dict[str, object]: - """Dump one external database and restore it into a declared service.""" + """Dump one external database and restore it into a declared service. + + A preflight finding refuses the copy before ``pg_dump`` with + ``data_copy_preflight_refused`` and the four lists in its context, read + from the findings file the role writes under ``receipt_directory``. + """ service = _checked_data_migration(context, service_id, database, source_url_file) plan = plan_data_migration( context, service_id, database, source_url_file, receipt_directory, schemas ) - execute_plan(plan, context.run) + findings = ( + receipt_directory / f"{service_id.value}-{database}.preflight.json" + if receipt_directory is not None + else None + ) + # A file an earlier run left must not answer for this one. + if findings is not None: + findings.unlink(missing_ok=True) + try: + execute_plan(plan, context.run) + except LifecycleError as error: + if ( + error.code != ERROR_EXECUTION_FAILED + or findings is None + or not findings.is_file() + ): + raise + found = DataCopyPreflightFindings.from_file(findings) + if found.refused: + raise found.refusal() from error + raise + finally: + if findings is not None: + findings.unlink(missing_ok=True) result: dict[str, object] = { "status": "ok", "action": "data-migration", @@ -1387,8 +1502,10 @@ def preview_data_migration( f"dump {selection.describe()} of the database whose URL is in " f"{source_url_file} on {service.server_id} and restore it into " f"declared database {database} of service {service_id}, refusing " - "a target with tables in those schemas and verifying per-table " - "row counts" + "before the dump a target with tables in those schemas, a schema " + "the source lacks, an extension or policy role the target lacks, " + "or an auth. reference; it restores in one transaction and " + "verifies per-table row counts" ), ) diff --git a/sdk/tests/test_data_copy_preflight.py b/sdk/tests/test_data_copy_preflight.py new file mode 100644 index 0000000..74ad104 --- /dev/null +++ b/sdk/tests/test_data_copy_preflight.py @@ -0,0 +1,278 @@ +"""The data copy preflight's refusal and the failed restore, as the SDK reads them. + +The role writes the preflight findings to a file under the receipt +directory and fails; these tests stand in for the playbook with a runner +that does the same, since no PostgreSQL runs here. +""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from typing import cast + +import pytest +from cloudfall.agent_tools import AgentConfig +from cloudfall.app import Migration, _migration_failure +from cloudfall.domain import ResourceId +from cloudfall.lifecycle import ( + EngineContext, + ExecutionStep, + LifecycleError, + StepRunner, + migrate_data, +) +from cloudfall.migrate import MigrateOptions, execute_migration + +ROOT = Path(__file__).parents[2] +SCHEMAS = ROOT / "config" / "schemas" / "v1" +EXAMPLES = ROOT / "config" / "examples" +ENGINE = ROOT / "engine" +SOURCE_URL = "postgresql://owner:hunter2@source.example.test/crm" +FINDINGS = "postgresql-main-crm.preflight.json" +NO_FINDINGS: dict[str, list[str]] = { + "missingSchemas": [], + "extensions": [], + "roles": [], + "authReferences": [], +} +# More than the 2000 characters of output a failed step keeps. +NOISE = "TASK [cloudfall_data_migration : noise] ok\n" * 100 + + +def _source(tmp_path: Path) -> Path: + url_file = tmp_path / "source.url" + url_file.write_text(f"{SOURCE_URL}\n", encoding="utf-8") + return url_file + + +def _playbook( + findings: dict[str, list[str]] | None, *, returncode: int +) -> StepRunner: + """Stand in for data.yml: write the findings file, then end as told.""" + + def run(step: ExecutionStep) -> subprocess.CompletedProcess[str]: + if "--extra-vars" not in step.argv: + return subprocess.CompletedProcess(step.argv, 0, "", "") + extra_vars = json.loads(step.argv[step.argv.index("--extra-vars") + 1]) + directory = Path(extra_vars["cloudfall_data_receipt_directory"]) + if findings is not None: + directory.mkdir(parents=True, exist_ok=True) + (directory / FINDINGS).write_text(json.dumps(findings), encoding="utf-8") + return subprocess.CompletedProcess(step.argv, returncode, NOISE, "") + + return run + + +def _copy( + tmp_path: Path, run: StepRunner, receipts: Path +) -> dict[str, object]: + context = EngineContext( + project_directory=EXAMPLES, + schema_directory=SCHEMAS, + engine_directory=ENGINE, + inventory_file=tmp_path / "inventory.json", + run=run, + ) + return migrate_data( + context, ResourceId("postgresql-main"), "crm", _source(tmp_path), receipts + ) + + +def _refusal(tmp_path: Path, findings: dict[str, list[str]]) -> LifecycleError: + receipts = tmp_path / "data-migrations" + with pytest.raises(LifecycleError) as raised: + _copy(tmp_path, _playbook(findings, returncode=2), receipts) + error = raised.value + assert error.code == "data_copy_preflight_refused" + assert not (receipts / FINDINGS).exists() + assert not (receipts / "postgresql-main-crm.json").exists() + assert "hunter2" not in f"{error.detail}{error.context}" + return error + + +def test_s007_6_a_missing_schema_is_refused_with_its_name(tmp_path: Path) -> None: + error = _refusal(tmp_path, {**NO_FINDINGS, "missingSchemas": ["missing"]}) + + assert error.context == {**NO_FINDINGS, "missingSchemas": ["missing"]} + assert "schemas the source lacks: missing" in error.detail + + +def test_s007_7_an_extension_the_target_lacks_is_named(tmp_path: Path) -> None: + error = _refusal(tmp_path, {**NO_FINDINGS, "extensions": ["citext"]}) + + assert error.context["extensions"] == ["citext"] + + +def test_s007_8_policy_roles_the_target_lacks_are_named(tmp_path: Path) -> None: + error = _refusal( + tmp_path, {**NO_FINDINGS, "roles": ["anon", "authenticated"]} + ) + + assert error.context["roles"] == ["anon", "authenticated"] + + +def test_s007_9_auth_references_are_named(tmp_path: Path) -> None: + references = [ + "function public.current_tenant()", + "policy notes_owner on public.notes", + ] + error = _refusal(tmp_path, {**NO_FINDINGS, "authReferences": references}) + + assert error.context["authReferences"] == references + + +def test_s007_10_every_finding_comes_whole_in_one_answer(tmp_path: Path) -> None: + # Far more names than the output tail could carry. + roles = [f"role_{index:04d}" for index in range(400)] + findings = { + "missingSchemas": [], + "extensions": ["postgis"], + "roles": roles, + "authReferences": ["function public.current_tenant()"], + } + + error = _refusal(tmp_path, findings) + + assert error.context == findings + for named in ("postgis", "role_0399", "public.current_tenant()"): + assert named in error.detail + + +def test_s007_11_a_failed_restore_is_an_engine_failure_without_a_receipt( + tmp_path: Path, +) -> None: + receipts = tmp_path / "data-migrations" + + with pytest.raises(LifecycleError) as raised: + _copy(tmp_path, _playbook(NO_FINDINGS, returncode=2), receipts) + + assert raised.value.code == "lifecycle_execution_failed" + assert not (receipts / FINDINGS).exists() + assert not (receipts / "postgresql-main-crm.json").exists() + # Nothing an earlier run left behind stops the next one. + done = _copy(tmp_path, _playbook(NO_FINDINGS, returncode=0), receipts) + assert done["status"] == "ok" + assert not (receipts / FINDINGS).exists() + + +def test_s007_11_a_stale_findings_file_never_answers_for_a_new_run( + tmp_path: Path, +) -> None: + receipts = tmp_path / "data-migrations" + receipts.mkdir() + (receipts / FINDINGS).write_text( + json.dumps({**NO_FINDINGS, "roles": ["anon"]}), encoding="utf-8" + ) + + with pytest.raises(LifecycleError) as raised: + _copy(tmp_path, _playbook(None, returncode=2), receipts) + + assert raised.value.code == "lifecycle_execution_failed" + + +def test_a_malformed_findings_file_is_refused(tmp_path: Path) -> None: + receipts = tmp_path / "data-migrations" + + with pytest.raises(LifecycleError) as raised: + _copy( + tmp_path, + _playbook( + cast("dict[str, list[str]]", {"roles": "anon"}), returncode=2 + ), + receipts, + ) + + assert raised.value.code == "data_copy_preflight_findings_invalid" + + +def _migrate_data_step(tmp_path: Path, run: StepRunner) -> dict[str, object]: + config = AgentConfig( + project_directory=EXAMPLES, + schema_directory=SCHEMAS, + engine_directory=ENGINE, + inventory_file=tmp_path / "inventory.json", + observed_directory=tmp_path / "observed", + service_observed_directory=tmp_path / "observed-services", + deployments_directory=tmp_path / "deployments", + releases_directory=tmp_path / "releases", + artifacts_directory=tmp_path / "artifacts", + data_migrations_directory=tmp_path / "data-migrations", + run=run, + ) + options = MigrateOptions( + plan_file=tmp_path / "plan.json", + builds={"crm-backend": "main"}, + releases={}, + environment_files={}, + data_migrations={"crm": _source(tmp_path)}, + ) + planned = execute_migration(config, options) + steps = cast("list[dict[str, object]]", planned["steps"]) + # Every other step already ran, so the resumed run is the data step alone. + for step in steps: + if step["id"] != "data:crm": + step["status"] = "completed" + options.plan_file.write_text( + json.dumps({"version": 1, "steps": steps}), encoding="utf-8" + ) + return execute_migration( + config, + MigrateOptions( + plan_file=options.plan_file, + builds=options.builds, + releases={}, + environment_files={}, + data_migrations=options.data_migrations, + execute=True, + ), + ) + + +def _migration(result: dict[str, object]) -> Migration: + return Migration( + effect="updated", + status=str(result["status"]), + steps=cast("list[dict[str, object]]", result["steps"]), + completed=int(cast("int", result["completed"])), + next=cast("str | None", result.get("next")), + step=cast("str | None", result.get("step")), + ) + + +def test_s007_10_the_migrate_data_step_surfaces_the_same_refusal( + tmp_path: Path, +) -> None: + findings = { + "missingSchemas": ["missing"], + "extensions": ["postgis"], + "roles": ["anon"], + "authReferences": ["function public.current_tenant()"], + } + + result = _migrate_data_step(tmp_path, _playbook(findings, returncode=2)) + error = cast("dict[str, object]", result["error"]) + failure = _migration_failure(_migration(result), error) + + assert result["status"] == "error" + assert result["step"] == "data:crm" + assert error["code"] == "data_copy_preflight_refused" + assert failure is not None + assert failure.code == "PRECONDITION" + context = failure.context + assert {key: context[key] for key in findings} == findings + assert context["step"] == "data:crm" + assert "hunter2" not in json.dumps(result) + + +def test_s007_11_a_failed_migrate_data_step_stays_an_engine_failure( + tmp_path: Path, +) -> None: + result = _migrate_data_step(tmp_path, _playbook(NO_FINDINGS, returncode=2)) + failure = _migration_failure( + _migration(result), cast("dict[str, object]", result["error"]) + ) + + assert failure is not None + assert failure.code == "ENGINE_STEP_FAILED"