diff --git a/changelog.d/116-converge-one-domain.md b/changelog.d/116-converge-one-domain.md new file mode 100644 index 0000000..9f55ab1 --- /dev/null +++ b/changelog.d/116-converge-one-domain.md @@ -0,0 +1,3 @@ +### Added + +- The `converge_domains` MCP tool takes an optional `domain`: given, `domains.yml` runs only on that Domain's proxy server (`--limit`) and renders only that Domain, and an unknown id exits `PRECONDITION` (`lifecycle_domain_missing`) before the inventory render; omitted, it runs on every server as before. Each domain receipt now records `previousConfigurationSha256`, the sha256 of the vhost file the run replaced, or `null` when none was there; receipts without it still load (#116) diff --git a/config/schemas/v1/deployment-receipt.schema.json b/config/schemas/v1/deployment-receipt.schema.json index 9d287f1..a8209a2 100644 --- a/config/schemas/v1/deployment-receipt.schema.json +++ b/config/schemas/v1/deployment-receipt.schema.json @@ -31,6 +31,17 @@ }, "configurationSha256": { "$ref": "common.schema.json#/$defs/sha256" + }, + "previousConfigurationSha256": { + "description": "sha256 of the file at configurationPath before this run rendered it, or null when no file was there", + "anyOf": [ + { + "$ref": "common.schema.json#/$defs/sha256" + }, + { + "type": "null" + } + ] } }, "required": [ diff --git a/engine/ansible/playbooks/domains.yml b/engine/ansible/playbooks/domains.yml index 60e8779..1d72a2b 100644 --- a/engine/ansible/playbooks/domains.yml +++ b/engine/ansible/playbooks/domains.yml @@ -15,7 +15,15 @@ {{ cloudfall_domains_issue_certificates | default(false) | bool }} cloudfall_nginx_site_receipt_directory: >- {{ cloudfall_domains_receipt_directory | default("") }} - loop: "{{ cloudfall_domains | default([]) }}" + # cloudfall_domains_domain_id limits the run to one Domain. + loop: >- + {{ + cloudfall_domains | default([]) + | selectattr('id', 'equalto', cloudfall_domains_domain_id) + | list + if cloudfall_domains_domain_id is defined + else cloudfall_domains | default([]) + }} loop_control: loop_var: cloudfall_domain_item label: "{{ cloudfall_domain_item.id }}" diff --git a/engine/ansible/roles/cloudfall_nginx_site/tasks/main.yml b/engine/ansible/roles/cloudfall_nginx_site/tasks/main.yml index 01136cf..9fb01e6 100644 --- a/engine/ansible/roles/cloudfall_nginx_site/tasks/main.yml +++ b/engine/ansible/roles/cloudfall_nginx_site/tasks/main.yml @@ -105,6 +105,18 @@ else '' }} +# The receipt records what this run replaced, so a hand-written vhost the +# render overwrites is not lost without a trace. It is hashed once, before +# the first render: the second render below would only see this run's own. +# A link is followed: stat gives no checksum for a link itself, and the file +# it points to is what Nginx read. +- name: Hash the virtual host this run replaces + ansible.builtin.stat: + path: "{{ cloudfall_nginx_site_domain.proxy.configurationPath }}" + checksum_algorithm: sha256 + follow: true + register: cloudfall_nginx_site_replaced + - name: Apply the virtual host before issuance ansible.builtin.include_tasks: apply_virtual_host.yml @@ -212,6 +224,10 @@ cloudfall_nginx_site_domain.proxy.configurationPath, 'configurationSha256': cloudfall_nginx_site_applied.stat.checksum, + 'previousConfigurationSha256': + cloudfall_nginx_site_replaced.stat.checksum + if cloudfall_nginx_site_replaced.stat.exists + else none, }, } | to_nice_json }} diff --git a/engine/tests/test_domains_one_domain.py b/engine/tests/test_domains_one_domain.py new file mode 100644 index 0000000..182dd93 --- /dev/null +++ b/engine/tests/test_domains_one_domain.py @@ -0,0 +1,177 @@ +"""``domains.yml`` for one Domain, and the hash of the vhost a run replaces. + +The role needs a Debian host with Nginx, so the loop is evaluated with +Ansible's own templating, and the receipt tasks run for real on localhost +against a vhost file under ``tmp_path``. +""" + +from __future__ import annotations + +import hashlib +import json +import shutil +import subprocess +from pathlib import Path + +import yaml +from ansible.parsing.dataloader import DataLoader +from ansible.template import Templar, trust_as_template +from cloudfall.validation import SchemaCatalog + +ROOT = Path(__file__).parents[2] +ANSIBLE = ROOT / "engine" / "ansible" +TASKS = ANSIBLE / "roles" / "cloudfall_nginx_site" / "tasks" +SCHEMAS = ROOT / "config" / "schemas" / "v1" +RECEIPT_TASKS = ( + "Hash the virtual host this run replaces", + "Hash the applied configuration for the receipt", + "Ensure the controller receipt directory exists", + "Write the deployment receipt on the controller", +) +RENDERED = "server { listen 80; }\n" + + +def _loop(variables: dict[str, object]) -> object: + play = DataLoader().load_from_file(str(ANSIBLE / "playbooks" / "domains.yml")) + assert isinstance(play, list) + (task,) = play[0]["tasks"] + templar = Templar(loader=DataLoader(), variables=variables) + return templar.template(trust_as_template(task["loop"])) + + +def _domain(domain_id: str) -> dict[str, object]: + return { + "id": domain_id, + "proxy": {"configurationPath": f"/etc/nginx/sites-available/{domain_id}.conf"}, + } + + +def test_s004_2_one_domain_id_renders_only_that_domain() -> None: + domains = [_domain("d1"), _domain("d2")] + + only_d1 = _loop({"cloudfall_domains": domains, "cloudfall_domains_domain_id": "d1"}) + + assert only_d1 == [_domain("d1")] + + +def test_s004_17_without_a_domain_id_every_domain_on_the_host_renders() -> None: + domains = [_domain("d1"), _domain("d2")] + + assert _loop({"cloudfall_domains": domains}) == domains + assert _loop({}) == [] + + +def test_s004_2_a_host_without_the_domain_renders_nothing() -> None: + variables: dict[str, object] = { + "cloudfall_domains": [_domain("d2")], + "cloudfall_domains_domain_id": "d1", + } + + assert _loop(variables) == [] + + +def _receipt_play(vhost: Path, receipts: Path) -> list[dict[str, object]]: + """Replay the role's receipt tasks, with a stand-in for the render.""" + # Plain YAML, not Ansible's loader: the tasks are written out again. + tasks = yaml.safe_load((TASKS / "main.yml").read_text(encoding="utf-8")) + assert isinstance(tasks, list) + named = {task["name"]: task for task in tasks} + render = { + "name": "Stand in for the render", + "ansible.builtin.copy": { + "content": RENDERED, + "dest": str(vhost), + "mode": "0644", + }, + } + hash_replaced, *receipt = (named[name] for name in RECEIPT_TASKS) + return [ + { + "name": "Replay the receipt tasks of cloudfall_nginx_site", + "hosts": "localhost", + "connection": "local", + "gather_facts": True, + "gather_subset": ["min"], + "vars": { + "cloudfall_server_id": "h2", + "cloudfall_nginx_site_receipt_directory": str(receipts), + "cloudfall_nginx_site_domain": { + "id": "crm-site", + "proxy": {"configurationPath": str(vhost)}, + }, + }, + "tasks": [hash_replaced, render, *receipt], + } + ] + + +def _run(tmp_path: Path, vhost: Path) -> dict[str, object]: + binary = shutil.which("ansible-playbook") + assert binary is not None + receipts = tmp_path / "deployments" + playbook = tmp_path / "receipt.yml" + play = _receipt_play(vhost, receipts) + playbook.write_text(yaml.safe_dump(play), encoding="utf-8") + done = subprocess.run( # noqa: S603 - fixed binary and arguments. + [binary, "--inventory", "localhost,", str(playbook)], + check=False, + capture_output=True, + text=True, + timeout=300, + ) + assert done.returncode == 0, done.stdout + done.stderr + receipt = json.loads((receipts / "crm-site.json").read_text(encoding="utf-8")) + SchemaCatalog(SCHEMAS).validate_named("deployment-receipt.schema.json", receipt) + spec = receipt["spec"] + assert isinstance(spec, dict) + return spec + + +def _sha256(text: str) -> str: + return hashlib.sha256(text.encode()).hexdigest() + + +def test_s004_11_the_receipt_keeps_the_hash_of_a_hand_written_vhost( + tmp_path: Path, +) -> None: + vhost = tmp_path / "crm.example.test.conf" + hand_written = "server { listen 80; server_name crm.example.test; }\n" + vhost.write_text(hand_written, encoding="utf-8") + + spec = _run(tmp_path, vhost) + + assert spec["previousConfigurationSha256"] == _sha256(hand_written) + assert spec["configurationSha256"] == _sha256(RENDERED) + + +def test_s004_11_a_run_with_no_vhost_there_records_null(tmp_path: Path) -> None: + spec = _run(tmp_path, tmp_path / "crm.example.test.conf") + + assert spec["previousConfigurationSha256"] is None + assert spec["configurationSha256"] == _sha256(RENDERED) + + +def test_s004_11_a_vhost_linked_into_place_is_hashed_through_the_link( + tmp_path: Path, +) -> None: + """Debian keeps vhosts as links into sites-available; hash what it reads.""" + hand_written = "server { listen 80; server_name crm.example.test; }\n" + target = tmp_path / "sites-available.conf" + target.write_text(hand_written, encoding="utf-8") + vhost = tmp_path / "crm.example.test.conf" + vhost.symlink_to(target) + + spec = _run(tmp_path, vhost) + + assert spec["previousConfigurationSha256"] == _sha256(hand_written) + + +def test_s004_11_the_replaced_vhost_is_hashed_before_the_first_render() -> None: + """The second render would see only this run's own file.""" + tasks = DataLoader().load_from_file(str(TASKS / "main.yml")) + assert isinstance(tasks, list) + names = [task["name"] for task in tasks] + + assert names.index(RECEIPT_TASKS[0]) < names.index( + "Apply the virtual host before issuance" + ) diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index 9bc70cf..d9c1213 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -161,6 +161,7 @@ ERROR_EXECUTION_FAILED, STEP_TIMEOUT_SECONDS, DeployOptions, + DomainsRun, EngineContext, ExecutionStep, LifecycleError, @@ -169,6 +170,7 @@ StepRunner, backup_service, build_release_artifact, + converge_domains, deploy, health, migrate_data, @@ -4192,13 +4194,22 @@ def run(arguments: Mapping[str, object], call: Ctx) -> Converged: } return run - def domain_vars(arguments: Mapping[str, object]) -> dict[str, object]: - return { - "cloudfall_domains_issue_certificates": bool( - arguments.get("issue_certificates", False) - ), - "cloudfall_domains_receipt_directory": str(root.path(args.deployments)), - } + def converge_domain_routes( + arguments: Mapping[str, object], call: Ctx + ) -> Converged: + domain = arguments.get("domain") + run = DomainsRun( + issue_certificates=bool(arguments.get("issue_certificates", False)), + receipt_directory=root.path(args.deployments), + domain_id=None if domain is None else ResourceId.from_boundary(domain), + ) + try: + converge_domains(context(call), run) + except LifecycleError as error: + raise _lifecycle_failed(error) from error + except ConfigValidationError as error: + raise _config_invalid(error) from error + return Converged(status="ok", action="converge-domains", playbook="domains.yml") no_inputs: dict[str, object] = { "type": "object", @@ -4251,16 +4262,20 @@ def domain_vars(arguments: Mapping[str, object]) -> dict[str, object]: name="converge_domains", description=( "Run domains.yml: render every declared domain route, and issue " - "missing certificates when issue_certificates is true" + "missing certificates when issue_certificates is true. Given a " + "domain, run only on that Domain's proxy server and render only it" ), input_schema={ "type": "object", - "properties": {"issue_certificates": {"type": "boolean"}}, + "properties": { + "issue_certificates": {"type": "boolean"}, + "domain": {"type": "string", "pattern": _COMPONENT_PATTERN}, + }, "additionalProperties": False, }, - handler=converge("converge-domains", "domains.yml", domain_vars), + handler=converge_domain_routes, danger_level="destructive", - exit_codes=_ENGINE_EXIT_CODES, + exit_codes=(*_ENGINE_EXIT_CODES, "CONFIG_INVALID"), ), ] diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index 4e9e6ef..76dddbb 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -28,11 +28,16 @@ from pathlib import Path from cloudfall.domain import ReleaseId, ResourceId - from cloudfall.inventory import ComponentInventory, ServiceInventory + from cloudfall.inventory import ( + ComponentInventory, + DomainInventory, + ServiceInventory, + ) _ARTIFACT_SCHEMA = "artifact.schema.json" _OUTPUT_TAIL_CHARACTERS = 2000 _ERROR_COMPONENT_MISSING = "lifecycle_component_missing" +_ERROR_DOMAIN_MISSING = "lifecycle_domain_missing" _ERROR_ARTIFACT_MISSING = "lifecycle_artifact_missing" _ERROR_ARTIFACT_METADATA = "lifecycle_artifact_metadata_invalid" _ERROR_ARTIFACT_IDENTITY = "lifecycle_artifact_identity_mismatch" @@ -399,17 +404,50 @@ def run_engine_playbook( context: EngineContext, playbook_name: str, extra_vars: Mapping[str, object], + limit: ResourceId | None = None, ) -> None: - """Render inventory and run one engine playbook contract.""" + """Render inventory and run one engine playbook contract. + + ``limit`` narrows the run to one declared server, so no other host is + contacted. + """ _execute_steps( ( _render_inventory_step(context), - _playbook_step(context, playbook_name, extra_vars), + _playbook_step(context, playbook_name, extra_vars, limit), ), context.run, ) +@dataclass(frozen=True, slots=True) +class DomainsRun: + """What one ``domains.yml`` run covers.""" + + issue_certificates: bool + receipt_directory: Path + domain_id: ResourceId | None = None + """One Domain to converge on its proxy server alone; ``None`` for all.""" + + +def converge_domains(context: EngineContext, run: DomainsRun) -> None: + """Run ``domains.yml`` on every server, or for one Domain on its proxy. + + One Domain is looked up in the validated config first, so an unknown id + is refused before the inventory render and before any SSH connection. + """ + extra_vars: dict[str, object] = { + "cloudfall_domains_issue_certificates": run.issue_certificates, + "cloudfall_domains_receipt_directory": str(run.receipt_directory), + } + if run.domain_id is None: + run_engine_playbook(context, "domains.yml", extra_vars) + return + proxy_server = _domain(context, run.domain_id).proxy.server_id + extra_vars["cloudfall_domains_domain_id"] = run.domain_id.value + run_engine_playbook(context, "domains.yml", extra_vars, limit=proxy_server) + + def build_release_artifact( context: EngineContext, component_id: ResourceId, @@ -859,6 +897,23 @@ def _component( return component +def _domain(context: EngineContext, domain_id: ResourceId) -> DomainInventory: + state = validate_config(context.project_directory, context.schema_directory) + inventory = PlatformInventory.from_state(state) + domain = next( + ( + candidate + for candidate in inventory.domains + if candidate.resource_id == domain_id + ), + None, + ) + if domain is None: + detail = f"domain does not exist: {domain_id}" + raise LifecycleError(_ERROR_DOMAIN_MISSING, detail) + return domain + + def _server_list(servers: tuple[ResourceId, ...]) -> str: return ", ".join(server.value for server in servers) @@ -887,18 +942,21 @@ def _playbook_step( context: EngineContext, playbook_name: str, extra_vars: Mapping[str, object], + limit: ResourceId | None = None, ) -> ExecutionStep: binary = shutil.which("ansible-playbook") if binary is None: detail = "ansible-playbook is not installed on this controller" raise LifecycleError(_ERROR_ANSIBLE_MISSING, detail) playbook = context.playbook(playbook_name) + limited = () if limit is None else ("--limit", limit.value) return ExecutionStep( description=f"run {playbook_name}", argv=( binary, "--inventory", str(context.inventory_file), + *limited, "--extra-vars", json.dumps(dict(extra_vars), sort_keys=True), str(playbook), diff --git a/sdk/tests/test_domains_converge.py b/sdk/tests/test_domains_converge.py new file mode 100644 index 0000000..79f569d --- /dev/null +++ b/sdk/tests/test_domains_converge.py @@ -0,0 +1,345 @@ +"""``domains.yml`` for one Domain: the SDK side of spec S-004. + +The engine steps run through a recording runner, so no host is contacted: +these tests read the argv each step would run. +""" + +from __future__ import annotations + +import asyncio +import json +import shutil +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import pytest +from cloudfall.cli import main +from cloudfall.domain import ResourceId +from cloudfall.lifecycle import ( + DomainsRun, + EngineContext, + ExecutionStep, + LifecycleError, + converge_domains, +) +from mcp import ClientSession, StdioServerParameters, stdio_client +from mcp.types import TextContent + +if TYPE_CHECKING: + from collections.abc import Sequence + +ROOT = Path(__file__).parents[2] +SCHEMAS = ROOT / "config" / "schemas" / "v1" +EXAMPLES = ROOT / "config" / "examples" +ENGINE = ROOT / "engine" +SERVE = ("-c", "from cloudfall.cli import run; run()", "mcp", "serve") +SHA = "a" * 64 + +SECOND_DOMAIN = """\ +--- +apiVersion: cloudfall/v1 +kind: Domain +metadata: + id: crm-admin + description: A second route on the same proxy server +spec: + primaryName: admin.crm.example.test + aliases: [] + proxy: + server: h2 + configurationPath: /etc/nginx/sites-available/admin.crm.example.test.conf + service: nginx.service + upstream: + address: 127.0.0.1 + port: 8101 + origin: + server: h1 + configurationPath: /etc/nginx/sites-available/crm-admin-origin.conf + service: nginx.service + scheme: http + port: 8101 + serverName: admin.crm.example.test + edge: + provider: cloudflare + mode: dns-only + tls: + mode: required + healthCheck: + scheme: https + path: /health + expectedStatuses: + - 200 + timeoutSeconds: 5 +""" + + +@dataclass(frozen=True, slots=True) +class _Done: + returncode: int + stdout: str = "" + stderr: str = "" + + +@dataclass(slots=True) +class _Recorder: + """Answer each step with an exit code, keeping what it was asked to run.""" + + playbook_exit: int = 0 + steps: list[ExecutionStep] = field(default_factory=list) + + def __call__(self, step: ExecutionStep) -> _Done: + self.steps.append(step) + if step.description.startswith("run "): + return _Done(self.playbook_exit, stderr="nginx: [emerg] rejected") + return _Done(0) + + +def _project(tmp_path: Path) -> Path: + project = tmp_path / "project" + shutil.copytree(EXAMPLES, project) + (project / "domains" / "crm-admin.yaml").write_text( + SECOND_DOMAIN, encoding="utf-8" + ) + return project + + +def _context(project: Path, recorder: _Recorder) -> EngineContext: + return EngineContext( + project_directory=project, + schema_directory=SCHEMAS, + engine_directory=ENGINE, + inventory_file=project / "tmp" / "inventory.json", + run=recorder, + ) + + +def _playbook_argv(recorder: _Recorder) -> Sequence[str]: + render, playbook = recorder.steps + assert render.description == "render Ansible inventory" + assert playbook.argv[-1].endswith("domains.yml") + return playbook.argv + + +def _extra_vars(argv: Sequence[str]) -> dict[str, object]: + extra_vars = json.loads(argv[list(argv).index("--extra-vars") + 1]) + assert isinstance(extra_vars, dict) + return extra_vars + + +def test_s004_2_one_domain_runs_on_its_proxy_and_names_only_it( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + recorder = _Recorder() + + converge_domains( + _context(project, recorder), + DomainsRun( + issue_certificates=False, + receipt_directory=project / "tmp" / "deployments", + domain_id=ResourceId("crm-site"), + ), + ) + + argv = _playbook_argv(recorder) + assert argv[argv.index("--limit") + 1] == "h2" + assert _extra_vars(argv)["cloudfall_domains_domain_id"] == "crm-site" + + +def test_s004_17_without_a_domain_the_run_is_as_before(tmp_path: Path) -> None: + project = _project(tmp_path) + recorder = _Recorder() + receipts = project / "tmp" / "deployments" + + converge_domains( + _context(project, recorder), + DomainsRun(issue_certificates=True, receipt_directory=receipts), + ) + + argv = _playbook_argv(recorder) + assert "--limit" not in argv + assert _extra_vars(argv) == { + "cloudfall_domains_issue_certificates": True, + "cloudfall_domains_receipt_directory": str(receipts), + } + + +def test_s004_17_an_unknown_domain_is_refused_before_any_step( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + recorder = _Recorder() + + with pytest.raises(LifecycleError) as error: + converge_domains( + _context(project, recorder), + DomainsRun( + issue_certificates=False, + receipt_directory=project / "tmp" / "deployments", + domain_id=ResourceId("nope"), + ), + ) + + assert error.value.code == "lifecycle_domain_missing" + assert recorder.steps == [] + + +def test_s004_13_a_rejected_vhost_fails_the_run_as_an_engine_step( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + recorder = _Recorder(playbook_exit=2) + + with pytest.raises(LifecycleError) as error: + converge_domains( + _context(project, recorder), + DomainsRun( + issue_certificates=False, + receipt_directory=project / "tmp" / "deployments", + domain_id=ResourceId("crm-site"), + ), + ) + + # lifecycle_execution_failed is what the CLI and MCP answer as + # ENGINE_STEP_FAILED (``_lifecycle_failed``). + assert error.value.code == "lifecycle_execution_failed" + assert "nginx: [emerg] rejected" in error.value.detail + + +async def _call( + server: StdioServerParameters, name: str, arguments: dict[str, Any] +) -> tuple[Any, dict[str, Any]]: + async with ( + stdio_client(server) as (read, write), + ClientSession(read, write) as session, + ): + await session.initialize() + tools = {tool.name: tool for tool in (await session.list_tools()).tools} + result = await session.call_tool(name, arguments) + content = result.content[0] + assert isinstance(content, TextContent) + return tools[name], json.loads(content.text) + + +def test_s004_17_the_tool_takes_a_domain_and_refuses_an_unknown_one( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + server = StdioServerParameters( + command=sys.executable, + args=[*SERVE, "--project", str(project)], + cwd=str(tmp_path), + ) + + tool, answer = asyncio.run( + _call( + server, + "converge_domains", + {"domain": "nope", "confirm_destructive": True}, + ) + ) + + assert "domain" in tool.input_schema["properties"] + assert answer["ok"] is False, answer + assert answer["error"]["code"] == "PRECONDITION" + assert answer["error"]["context"]["code"] == "lifecycle_domain_missing" + assert not (project / "tmp" / "inventory.json").exists() + + +@pytest.mark.parametrize("domain", ["h1,h2", "all", "*", "h1:h2", "!h1", "CRM"]) +def test_s004_17_a_domain_that_is_no_resource_id_is_refused( + domain: str, tmp_path: Path +) -> None: + """A value Ansible's --limit would read as a pattern never reaches it.""" + project = _project(tmp_path) + server = StdioServerParameters( + command=sys.executable, + args=[*SERVE, "--project", str(project)], + cwd=str(tmp_path), + ) + + _, answer = asyncio.run( + _call( + server, + "converge_domains", + {"domain": domain, "confirm_destructive": True}, + ) + ) + + assert answer["ok"] is False, answer + assert not (project / "tmp" / "inventory.json").exists() + + +def _receipt(directory: Path, spec: dict[str, object]) -> None: + directory.mkdir(parents=True) + receipt = { + "apiVersion": "cloudfall/v1", + "kind": "DeploymentReceipt", + "metadata": {"id": "crm-site", "description": "written by an older run"}, + "spec": { + "domain": "crm-site", + "server": "h2", + "deployedAt": "2026-10-01T00:00:00Z", + "configurationPath": "/etc/nginx/sites-available/crm.example.test.conf", + "configurationSha256": SHA, + **spec, + }, + } + (directory / "crm-site.json").write_text(json.dumps(receipt), encoding="utf-8") + + +def _services_status( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> tuple[int, dict[str, Any]]: + observed = tmp_path / "observed" + observed.mkdir() + exit_code = main( + [ + "services", + "status", + "--project", + str(EXAMPLES), + "--schemas", + str(SCHEMAS), + "--observed", + str(observed), + "--deployments", + str(tmp_path / "deployments"), + ] + ) + payload = json.loads(capsys.readouterr().out) + assert isinstance(payload, dict) + return exit_code, payload + + +@pytest.mark.parametrize( + "spec", + [{}, {"previousConfigurationSha256": None}, {"previousConfigurationSha256": SHA}], + ids=["without-the-field", "null", "a-hash"], +) +def test_s004_12_a_receipt_with_or_without_the_previous_hash_loads( + spec: dict[str, object], + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + _receipt(tmp_path / "deployments", spec) + + exit_code, payload = _services_status(tmp_path, capsys) + + assert exit_code == 0, payload + (service,) = payload["data"]["services"] + assert service["lifecycle"]["deployed"] == "yes" + assert service["lifecycle"]["deployed_at"] == "2026-10-01T00:00:00Z" + + +def test_a_malformed_previous_hash_is_refused( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + _receipt(tmp_path / "deployments", {"previousConfigurationSha256": "not-a-hash"}) + + exit_code, payload = _services_status(tmp_path, capsys) + + assert exit_code != 0 + assert payload["error"]["code"] == "CONFIG_INVALID"