From de5979acc312ced94d382caa9487ca5952b62b65 Mon Sep 17 00:00:00 2001 From: wchwawa Date: Tue, 22 Sep 2026 17:30:48 +1000 Subject: [PATCH] feat(workbench): bring up the LoopX Stage 2A stack from one command LoopX qualifies its NoKV authority candidate with two environment-gated ladder rows that need an etcd control path, an S3-compatible object store, one serving owner, one existing workbench, a client configuration in the exact key shape of LoopX's JSON-lines helper, and thirteen environment variables. Until now that recipe lived only in a contributor's scratch directory, so the LoopX maintainer could not run the gate themselves. `scripts/workbench/loopx_stage2a_stack.py up` produces the whole stack from one command, reusing the parts CI already qualifies: an isolated etcd member, the digest-pinned RustFS container from `start_rustfs.sh` (or a `moto` S3 server without Docker), and the `provision` / `serve` argument shape of `live_workbench.py`. It refuses a wheel whose version differs from the owner binary or that lacks `WorkspaceIncarnationMismatch`, creates one workbench through the SDK and asserts exactly one incarnation, writes `nokv-client.json` and `live.env` as 0600 files, keeps every credential out of stdout, and refuses to write files whose leaves would trip LoopX's ladder privacy scan. `plan`, `status` and `down --purge` cover dry-run, liveness and teardown. The unit test freezes the command shapes, the LoopX key contracts, secret redaction and the fail-closed paths without starting a process; the workbench-contract CI job compiles and runs it. Signed-off-by: wchwawa --- .github/workflows/rust.yml | 4 +- docs/development/loopx-stage2a-stack.md | 110 +++ docs/index.md | 5 +- scripts/workbench/README.md | 9 + scripts/workbench/loopx_stage2a_stack.py | 871 ++++++++++++++++++ scripts/workbench/loopx_stage2a_stack_test.py | 245 +++++ 6 files changed, 1241 insertions(+), 3 deletions(-) create mode 100644 docs/development/loopx-stage2a-stack.md create mode 100644 scripts/workbench/loopx_stage2a_stack.py create mode 100644 scripts/workbench/loopx_stage2a_stack_test.py diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 9f4bed28f..f1cca4c2c 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -54,7 +54,8 @@ jobs: scripts/workbench/local_wal_recovery_gate.py \ scripts/workbench/object_namespace_recovery_gate.py \ scripts/workbench/restore_composition_gate.py \ - scripts/workbench/fork_restore_recovery_gate.py + scripts/workbench/fork_restore_recovery_gate.py \ + scripts/workbench/loopx_stage2a_stack.py python3 scripts/workbench/pre423_contract_ledger.py python3 scripts/workbench/pre423_contract_ledger_test.py python3 scripts/workbench/workbench_contract_test.py @@ -63,6 +64,7 @@ jobs: python3 scripts/workbench/object_namespace_recovery_gate_test.py python3 scripts/workbench/restore_composition_gate_test.py python3 scripts/workbench/fork_restore_recovery_gate_test.py + python3 scripts/workbench/loopx_stage2a_stack_test.py bash -n scripts/workbench/start_rustfs.sh - name: Check Phase 1 qualification framework diff --git a/docs/development/loopx-stage2a-stack.md b/docs/development/loopx-stage2a-stack.md new file mode 100644 index 000000000..515da59fe --- /dev/null +++ b/docs/development/loopx-stage2a-stack.md @@ -0,0 +1,110 @@ + + +# LoopX Stage 2A Stack + +Status: test-only bring-up for LoopX's NoKV authority qualification. One owner, +one node, loopback only. It is not a deployment profile. + +LoopX qualifies its NoKV `AuthorityStore` candidate with two environment-gated +rows of its shared-goal-authority ladder, `s0.nokv_live_matrix` and +`s2a.nokv_live_qualification` (`loopx/control_plane/testing/authority_e2e_ladder.py`). +Both rows need a serving NoKV owner, an etcd control path, an S3-compatible +object store, an existing workbench, an ignored client configuration in the +exact key shape LoopX's `nokv_jsonl_helper.py` admits, and a fixed set of +environment variables. `scripts/workbench/loopx_stage2a_stack.py` produces all +of that from one command so a LoopX maintainer can run the gate without +reconstructing the recipe by hand. + +## Prerequisites + +- A `nokv` binary of the release under test: `target/release/nokv` of this + checkout (`--build` compiles it) or `--nokv-binary PATH`. +- A Python virtual environment with the matching `nokv` wheel installed, for + example: + + ```bash + python3 -m venv stage2a-venv + stage2a-venv/bin/pip install "nokv==0.11.1" \ + --find-links https://github.com/NoKV-Lab/NoKV/releases/expanded_assets/v0.11.1 + ``` + + The script refuses a wheel whose `__version__` differs from `nokv --version`, + and a wheel without `WorkspaceIncarnationMismatch`, because LoopX's helper + admits only a fenced SDK from the same release as the owner. +- `etcd` and `etcdctl` on `PATH` (`brew install etcd`, or the pinned archive + NoKV CI installs) or `--etcd-bin` / `--etcdctl-bin`. +- An object store. The default is the digest-pinned RustFS container from + `scripts/workbench/start_rustfs.sh`, which needs `docker` and the `aws` CLI. + Without Docker, `--object-store moto` runs a `moto` S3 server from the same + venv (`pip install "moto[server]" boto3`). A single-disk MinIO does not pass + NoKV's object admission probe and is not supported here. + +## Commands + +```bash +# Start everything and create one workbench. The directory must be empty. +python3 scripts/workbench/loopx_stage2a_stack.py up \ + --stack-dir /path/to/stage2a \ + --python /path/to/stage2a-venv/bin/python \ + --loopx-python /path/to/loopx-venv/bin/python + +# Print the redacted plan (identities, ports, commands) without starting anything. +python3 scripts/workbench/loopx_stage2a_stack.py plan --stack-dir /path/to/stage2a --python /path/to/stage2a-venv/bin/python + +# Liveness of the recorded processes; exit 1 when any is gone. +python3 scripts/workbench/loopx_stage2a_stack.py status --stack-dir /path/to/stage2a + +# Stop the owner, etcd, moto or the RustFS container; --purge also deletes the directory. +python3 scripts/workbench/loopx_stage2a_stack.py down --stack-dir /path/to/stage2a --purge +``` + +`up` writes into the stack directory: + +| File | Content | +| --- | --- | +| `nokv-client.json` (0600) | `root_id`, `routing` (`etcd` kind: `endpoints`, `key_prefix`, `lease_ttl_seconds`), `object_store` (`s3` kind: `bucket`, `region`, `root`, `endpoint`, `access_key_id`, `secret_access_key`) and `workbench_root`, exactly the keys the LoopX helper accepts | +| `live.env` (0600) | the `env:nokv_legacy` variables (`NOKV_COORDINATION_LIVE`, `NOKV_ETCD`, `NOKV_ETCD_PREFIX`, `NOKV_ROOT_ID`, `NOKV_BUCKET`, `NOKV_OBJECT_ENDPOINT`, `NOKV_OBJECT_ROOT`, `NOKV_OBJECT_KEY`, `NOKV_OBJECT_SECRET`) and the `env:nokv_authority` variables (`LOOPX_NOKV_AUTHORITY_LIVE`, `LOOPX_NOKV_AUTHORITY_CONFIG_JSON`, `LOOPX_NOKV_AUTHORITY_PYTHON`, `LOOPX_NOKV_AUTHORITY_WORKBENCH`) | +| `stack.json` (0600) | process ids, ports, the binary's SHA-256, the SDK version and digests of the workbench name and configuration; no credentials | +| `next-steps.txt` | the three LoopX commands below | +| `etcd/`, `metadata/`, `rustfs/`, `logs/` | member data, the owner's local WAL, object data, process logs | + +stdout carries only digests: the workbench name, the credentials and the etcd +prefix never leave the two 0600 files. Generated names are random with fixed +prefixes (`lx-`, `rt-`, `ak`, `sk`, `wb`, `nd-`), because LoopX's ladder turns +every configuration leaf and environment value into a forbidden token and fails +a report that contains one; `up` refuses to write files whose leaves collide +with ladder vocabulary. + +## Running the LoopX gate + +From a LoopX checkout whose Python can import `loopx`: + +```bash +set -a && source /path/to/stage2a/live.env && set +a +python -m loopx.control_plane.testing.authority_e2e_ladder \ + --row s0.nokv_live_matrix --row s2a.nokv_live_qualification \ + --report-json /path/to/stage2a/ladder-stage2a.json +# The complete ladder; rows whose environment is absent report unverified. +python -m loopx.control_plane.testing.authority_e2e_ladder --allow-unverified \ + --report-json /path/to/stage2a/ladder-full.json +``` + +A green `s2a.nokv_live_qualification` row proves single-node store conformance +for the SDK release the helper pins, including the stale-incarnation +publication fence (`stale_incarnation_fence_rejected`, +`stale_incarnation_fence_left_generation_unchanged`). The same stack with a +wheel from an older release is the negative pairing: the helper refuses it at +admission and the row fails typed, which is the expected outcome, not a stack +defect. + +## What this does not prove + +Availability, failover, restart or restore recovery, capacity, multi-owner +operation, production object stores or authenticated transport. Those remain +LoopX qualification holds on the NoKV profile and NoKV's own acceptance gates +(see [Workspace Acceptance](./workspace-acceptance.md)); this script only +removes the manual reconstruction of the single-node stack the two Stage 2A +rows require. diff --git a/docs/index.md b/docs/index.md index 7fec00792..a7eaf5a7d 100644 --- a/docs/index.md +++ b/docs/index.md @@ -100,8 +100,9 @@ SPDX-License-Identifier: Apache-2.0 - Development: [Code Contract](./development/code_contract.md), [`nokv-agent` Handbook](./development/nokv-agent.md), [PR Review Checklist](./development/pr_review_checklist.md), - [Change Governance](./development/change-governance.md), and - [Path-Native Metadata Comparison](./development/path-native-metadata-comparison.md). + [Change Governance](./development/change-governance.md), + [Path-Native Metadata Comparison](./development/path-native-metadata-comparison.md), and + [LoopX Stage 2A Stack](./development/loopx-stage2a-stack.md). - Storage architecture: [Metadata Store Interface](./development/metadata-store-interface.md). - Collaboration record: [NoKV x LingTai](./announcements/nokv-lingtai-design-partner.md) and [Chinese version](./announcements/nokv-lingtai-design-partner.zh-CN.md). diff --git a/scripts/workbench/README.md b/scripts/workbench/README.md index 1dd45f522..b4194ee2e 100644 --- a/scripts/workbench/README.md +++ b/scripts/workbench/README.md @@ -62,6 +62,14 @@ The checked-in integration assets are deliberately small: immutable objects to prove zero-copy reuse. - `fork_restore_recovery_gate_test.py` freezes the fault classifier, 1 GiB profile, terminal evidence, and independent GitHub Actions job contract. +- `loopx_stage2a_stack.py` brings up the single-node stack LoopX's Stage 2A + ladder rows expect (isolated etcd, the digest-pinned RustFS container or a + `moto` fallback, one provisioned owner, one workbench) and writes the ignored + client configuration and environment file those rows read; see + [LoopX Stage 2A Stack](../../docs/development/loopx-stage2a-stack.md). +- `loopx_stage2a_stack_test.py` freezes the command shapes, the exact LoopX + helper/ladder key contracts, secret redaction, the privacy-collision check, + and the fail-closed `plan`/`down`/`status` behaviour without starting a process. - `start_rustfs.sh` starts the optional local S3-compatible artifact backend with a digest-pinned image and bounded AWS CLI readiness attempts. It uses a Docker-managed volume by default so RustFS's non-root UID owns `/data` on @@ -91,6 +99,7 @@ python3 scripts/workbench/live_workbench_test.py python3 scripts/workbench/local_wal_recovery_gate_test.py python3 scripts/workbench/object_namespace_recovery_gate_test.py python3 scripts/workbench/restore_composition_gate_test.py +python3 scripts/workbench/loopx_stage2a_stack_test.py PYTHONPATH=scripts/workbench python3 -m unittest \ scripts/workbench/typed_live_qualification_test.py \ scripts/workbench/live_gap_qualification_test.py \ diff --git a/scripts/workbench/loopx_stage2a_stack.py b/scripts/workbench/loopx_stage2a_stack.py new file mode 100644 index 000000000..e6dbe5e4c --- /dev/null +++ b/scripts/workbench/loopx_stage2a_stack.py @@ -0,0 +1,871 @@ +#!/usr/bin/env python3 +"""Bring up the single-node NoKV stack that LoopX's Stage 2A rows expect. + +LoopX qualifies its NoKV authority candidate against a live owner with two +environment-gated ladder rows (`s0.nokv_live_matrix`, `s2a.nokv_live_qualification` +in `loopx/control_plane/testing/authority_e2e_ladder.py`). Those rows need an +etcd control path, an S3-compatible object store, one serving `nokv` owner, one +existing workbench, an ignored client configuration file in the exact key shape +of LoopX's `nokv_jsonl_helper.py`, and a handful of environment variables. This +script produces all of that from one command so the LoopX maintainer can run +the gate without reconstructing the recipe by hand. + +It reuses the parts NoKV CI already qualifies: an isolated etcd member, the +digest-pinned RustFS container from `start_rustfs.sh` (or a `moto` S3 server +when Docker is unavailable), and the `provision` / `serve` argument shape of +`live_workbench.py`. It is a test stack: one owner, one node, no HA, no +production hardening. Credentials it generates are random, live only in the +0600 files it writes, and never appear on stdout. +""" + +from __future__ import annotations + +import argparse +import dataclasses +import hashlib +import json +import os +import secrets +import shlex +import shutil +import signal +import socket +import subprocess +import sys +import time +from pathlib import Path +from typing import Any, Callable, Sequence + +REPO = Path(__file__).resolve().parents[2] +START_RUSTFS = REPO / "scripts" / "workbench" / "start_rustfs.sh" +STATE_FILE = "stack.json" +CLIENT_CONFIG_FILE = "nokv-client.json" +LIVE_ENV_FILE = "live.env" +NEXT_STEPS_FILE = "next-steps.txt" +OBJECT_REGION = "us-east-1" +WORKBENCH_ROOT = "/agents/stage2a/wb" +ETCD_LEASE_TTL_SECONDS = 10 +# Every string leaf of the client configuration and every value of the live +# environment becomes a forbidden token in LoopX's ladder privacy scan, which +# fails a report when any token appears in the evidence. Generated names are +# therefore prefixed and random so none can collide with ladder vocabulary. +LADDER_VOCABULARY = ( + "s0.nokv_live_matrix", + "s2a.nokv_live_qualification", + "stage_2a_single_node_store_conformance", + "loopx_nokv_authority_live_qualification_v0", + "loopx_shared_goal_authority_e2e_report_v0", + "nokv_sdk_version", + "nokv_api_version", + "privacy_violations", + "unverified", + "pending", + "pass", + "fail", +) +SECRET_FLAGS = {"--object-secret-access-key"} + + +class StackError(RuntimeError): + """The stack could not be prepared; nothing partially prepared is trusted.""" + + +@dataclasses.dataclass(frozen=True) +class StackIdentity: + root_id: str + shard_id: str + agent_id: str + etcd_prefix: str + bucket: str + object_root: str + access_key: str + secret_key: str + workbench: str + node: str + + +@dataclasses.dataclass(frozen=True) +class Endpoints: + etcd_client_port: int + etcd_peer_port: int + object_port: int + object_console_port: int + owner_port: int + + @property + def etcd(self) -> str: + return f"http://127.0.0.1:{self.etcd_client_port}" + + @property + def etcd_peer(self) -> str: + return f"http://127.0.0.1:{self.etcd_peer_port}" + + @property + def object_store(self) -> str: + return f"http://127.0.0.1:{self.object_port}" + + @property + def owner(self) -> str: + return f"127.0.0.1:{self.owner_port}" + + +def fresh_identity(token_hex: Callable[[int], str] = secrets.token_hex) -> StackIdentity: + """Random ids and credentials whose prefixes cannot collide with ladder vocabulary.""" + + return StackIdentity( + root_id=token_hex(16), + shard_id=token_hex(16), + agent_id=token_hex(16), + etcd_prefix=f"/lx-{token_hex(6)}", + bucket=f"lx-{token_hex(8)}", + object_root=f"rt-{token_hex(6)}", + access_key=f"ak{token_hex(12)}", + secret_key=f"sk{token_hex(24)}", + workbench=f"wb{token_hex(5)}", + node=f"nd-{token_hex(4)}", + ) + + +def digest(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def free_port() -> int: + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + return int(probe.getsockname()[1]) + + +def fresh_endpoints(port: Callable[[], int] = free_port) -> Endpoints: + return Endpoints( + etcd_client_port=port(), + etcd_peer_port=port(), + object_port=port(), + object_console_port=port(), + owner_port=port(), + ) + + +# --- command construction (pure; frozen by loopx_stage2a_stack_test.py) ------ + + +def etcd_command( + etcd: Path, data_dir: Path, name: str, endpoints: Endpoints +) -> list[str]: + return [ + str(etcd), + "--name", + name, + "--data-dir", + str(data_dir), + "--listen-client-urls", + endpoints.etcd, + "--advertise-client-urls", + endpoints.etcd, + "--listen-peer-urls", + endpoints.etcd_peer, + "--initial-advertise-peer-urls", + endpoints.etcd_peer, + "--initial-cluster", + f"{name}={endpoints.etcd_peer}", + "--initial-cluster-state", + "new", + ] + + +def moto_command(python: Path, endpoints: Endpoints) -> list[str]: + return [ + str(python), + "-m", + "moto.server", + "-H", + "127.0.0.1", + "-p", + str(endpoints.object_port), + ] + + +def control_args(binary: Path, identity: StackIdentity, endpoints: Endpoints) -> list[str]: + return [ + str(binary), + "--root-id", + identity.root_id, + "--etcd-endpoint", + endpoints.etcd, + "--etcd-key-prefix", + identity.etcd_prefix, + "--etcd-lease-ttl-seconds", + str(ETCD_LEASE_TTL_SECONDS), + ] + + +def object_args(identity: StackIdentity, endpoints: Endpoints) -> list[str]: + return [ + "--object-bucket", + identity.bucket, + "--object-endpoint", + endpoints.object_store, + "--object-root", + identity.object_root, + "--object-region", + OBJECT_REGION, + "--object-access-key-id", + identity.access_key, + "--object-secret-access-key", + identity.secret_key, + ] + + +def provision_command(binary: Path, identity: StackIdentity, endpoints: Endpoints) -> list[str]: + return [ + *control_args(binary, identity, endpoints), + "--agent-id", + identity.agent_id, + *object_args(identity, endpoints), + "provision", + identity.shard_id, + ] + + +def server_command( + binary: Path, identity: StackIdentity, endpoints: Endpoints, metadata: Path +) -> list[str]: + return [ + *control_args(binary, identity, endpoints), + *object_args(identity, endpoints), + "--bind", + endpoints.owner, + "--advertise-endpoint", + endpoints.owner, + "--node-id", + identity.node, + "--metadata-create", + str(metadata), + "--lifecycle-interval-millis", + "100", + "serve", + ] + + +def client_config(identity: StackIdentity, endpoints: Endpoints) -> dict[str, Any]: + """The exact key shape `loopx/control_plane/coordination/nokv_jsonl_helper.py` admits.""" + + return { + "root_id": identity.root_id, + "routing": { + "kind": "etcd", + "endpoints": [endpoints.etcd], + "key_prefix": identity.etcd_prefix, + "lease_ttl_seconds": ETCD_LEASE_TTL_SECONDS, + }, + "object_store": { + "kind": "s3", + "bucket": identity.bucket, + "region": OBJECT_REGION, + "root": identity.object_root, + "endpoint": endpoints.object_store, + "access_key_id": identity.access_key, + "secret_access_key": identity.secret_key, + }, + "workbench_root": WORKBENCH_ROOT, + } + + +def live_env( + identity: StackIdentity, + endpoints: Endpoints, + *, + config_path: Path, + python: Path, +) -> dict[str, str]: + """Variables for the ladder gates `env:nokv_legacy` and `env:nokv_authority`.""" + + return { + "NOKV_COORDINATION_LIVE": "1", + "NOKV_ETCD": endpoints.etcd, + "NOKV_ETCD_PREFIX": identity.etcd_prefix, + "NOKV_ROOT_ID": identity.root_id, + "NOKV_BUCKET": identity.bucket, + "NOKV_OBJECT_ENDPOINT": endpoints.object_store, + "NOKV_OBJECT_ROOT": identity.object_root, + "NOKV_OBJECT_KEY": identity.access_key, + "NOKV_OBJECT_SECRET": identity.secret_key, + "LOOPX_NOKV_AUTHORITY_LIVE": "1", + "LOOPX_NOKV_AUTHORITY_CONFIG_JSON": str(config_path), + "LOOPX_NOKV_AUTHORITY_PYTHON": str(python), + "LOOPX_NOKV_AUTHORITY_WORKBENCH": identity.workbench, + } + + +def render_env(values: dict[str, str]) -> str: + return "".join(f"export {key}={shlex.quote(value)}\n" for key, value in values.items()) + + +def string_leaves(value: Any) -> list[str]: + if isinstance(value, dict): + return [leaf for item in value.values() for leaf in string_leaves(item)] + if isinstance(value, list): + return [leaf for item in value for leaf in string_leaves(item)] + if isinstance(value, str): + return [value] + return [] + + +def privacy_collisions(config: dict[str, Any], env: dict[str, str]) -> list[str]: + """Generated leaves that LoopX's ladder privacy scan would find inside its own report.""" + + collisions: list[str] = [] + for leaf in {*string_leaves(config), *env.values()}: + if len(leaf) < 4: + continue + for word in LADDER_VOCABULARY: + if leaf in word: + collisions.append(f"{leaf!r} is a substring of ladder vocabulary {word!r}") + return sorted(collisions) + + +def ladder_commands(stack_dir: Path, loopx_python: str) -> list[str]: + env = stack_dir / LIVE_ENV_FILE + ladder = "-m loopx.control_plane.testing.authority_e2e_ladder" + return [ + f"set -a && source {shlex.quote(str(env))} && set +a", + f"{loopx_python} {ladder} --row s0.nokv_live_matrix --row s2a.nokv_live_qualification" + f" --report-json {shlex.quote(str(stack_dir / 'ladder-stage2a.json'))}", + f"{loopx_python} {ladder} --allow-unverified --report-json" + f" {shlex.quote(str(stack_dir / 'ladder-full.json'))}", + ] + + +def redact_argv(argv: Sequence[str]) -> list[str]: + output: list[str] = [] + redact_next = False + for argument in argv: + if redact_next: + output.append(f"") + redact_next = False + else: + output.append(argument) + redact_next = argument in SECRET_FLAGS + return output + + +def plan( + *, + binary: Path, + python: Path, + stack_dir: Path, + identity: StackIdentity, + endpoints: Endpoints, + object_store: str, +) -> dict[str, Any]: + """Everything `up` will do, with secrets redacted; `plan` prints this without starting anything.""" + + config = client_config(identity, endpoints) + env = live_env(identity, endpoints, config_path=stack_dir / CLIENT_CONFIG_FILE, python=python) + return { + "stack_dir": str(stack_dir), + "object_store": object_store, + "endpoints": dataclasses.asdict(endpoints), + "commands": { + "etcd": redact_argv(etcd_command(Path("etcd"), stack_dir / "etcd", identity.node, endpoints)), + "provision": redact_argv(provision_command(binary, identity, endpoints)), + "serve": redact_argv(server_command(binary, identity, endpoints, stack_dir / "metadata")), + }, + "client_config_sha256": digest(canonical_json(config)), + "workbench_sha256_prefix": digest(identity.workbench)[:12], + "privacy_collisions": privacy_collisions(config, env), + "files": [CLIENT_CONFIG_FILE, LIVE_ENV_FILE, STATE_FILE, NEXT_STEPS_FILE], + } + + +# --- process helpers --------------------------------------------------------- + + +def start_process(argv: Sequence[str], log: Path, env: dict[str, str] | None = None) -> subprocess.Popen[str]: + handle = log.open("a", encoding="utf-8") + return subprocess.Popen( + list(argv), + stdin=subprocess.DEVNULL, + stdout=handle, + stderr=subprocess.STDOUT, + text=True, + start_new_session=True, + env=env, + ) + + +def stop_pid(pid: int) -> None: + """Terminate a process started with its own session; escalate to SIGKILL after 10 s.""" + + for sig in (signal.SIGTERM, signal.SIGKILL): + try: + os.kill(pid, sig) + except ProcessLookupError: + return + except PermissionError: + pass + try: + os.killpg(pid, sig) + except (ProcessLookupError, PermissionError): + pass + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + os.kill(pid, 0) + except ProcessLookupError: + return + time.sleep(0.1) + + +def wait_tcp(process: subprocess.Popen[str], port: int, timeout: float, what: str) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if process.poll() is not None: + raise StackError(f"{what} exited before listening (code {process.returncode})") + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.2): + return + except OSError: + time.sleep(0.05) + raise StackError(f"{what} did not listen on 127.0.0.1:{port} within {timeout:.0f}s") + + +def wait_etcd(etcdctl: Path, endpoint: str, process: subprocess.Popen[str], timeout: float) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if process.poll() is not None: + raise StackError(f"etcd exited before readiness (code {process.returncode})") + result = subprocess.run( + [str(etcdctl), f"--endpoints={endpoint}", "endpoint", "health"], + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + if result.returncode == 0: + return + time.sleep(0.1) + raise StackError(f"etcd did not become healthy at {endpoint}") + + +def run_checked(argv: Sequence[str], *, timeout: float, what: str, log: Path | None = None) -> str: + result = subprocess.run(list(argv), capture_output=True, text=True, timeout=timeout, check=False) + if log is not None: + with log.open("a", encoding="utf-8") as handle: + handle.write(f"$ {' '.join(redact_argv(argv))}\n{result.stdout}{result.stderr}\n") + if result.returncode != 0: + raise StackError(f"{what} failed (code {result.returncode}); see {log or 'stderr'}") + return result.stdout + + +def python_json(python: Path, snippet: str, payload: dict[str, Any], *, timeout: float, what: str) -> dict[str, Any]: + """Run a snippet under the SDK interpreter; the snippet reads JSON on stdin and prints JSON.""" + + result = subprocess.run( + [str(python), "-I", "-c", snippet], + input=json.dumps(payload), + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + if result.returncode != 0: + raise StackError(f"{what} failed under {python}: {result.stderr.strip().splitlines()[-1:] or 'no output'}") + try: + value = json.loads(result.stdout) + except json.JSONDecodeError as error: + raise StackError(f"{what} printed invalid JSON") from error + if not isinstance(value, dict): + raise StackError(f"{what} printed a non-object") + return value + + +SDK_VERSION_SNIPPET = """ +import json, sys +import nokv +print(json.dumps({"version": nokv.__version__, "api_version": nokv.API_VERSION, + "fence": hasattr(nokv, "WorkspaceIncarnationMismatch")})) +""" + +CREATE_WORKBENCH_SNIPPET = """ +import json, sys +import nokv +p = json.load(sys.stdin) +client = nokv.Client( + root_id=p["root_id"], + routing=nokv.RoutingConfig.etcd(p["etcd_endpoints"], p["etcd_prefix"], p["lease_ttl"]), + object_store=nokv.ObjectStoreConfig.s3(p["bucket"], region=p["region"], root=p["object_root"], + endpoint=p["object_endpoint"], access_key_id=p["access_key"], secret_access_key=p["secret_key"]), + workbench_root=p["workbench_root"], +) +client.create_workspace(p["workbench"]) +found = [] +cursor = None +while True: + page = client.find_workspaces(cursor=cursor, limit=100) + for entry in page["workspaces"]: + workspace = entry["workspace"] + if workspace["workbench"] == p["workbench"]: + found.append(workspace["workspace_incarnation_id"]) + cursor = page.get("next_cursor") + if not cursor: + break +print(json.dumps({"incarnations": found})) +""" + +MOTO_BUCKET_SNIPPET = """ +import json, sys +import boto3 +p = json.load(sys.stdin) +s3 = boto3.client("s3", endpoint_url=p["endpoint"], region_name=p["region"], + aws_access_key_id=p["access_key"], aws_secret_access_key=p["secret_key"]) +s3.create_bucket(Bucket=p["bucket"]) +print(json.dumps({"buckets": [b["Name"] for b in s3.list_buckets()["Buckets"]]})) +""" + + +# --- commands ----------------------------------------------------------------- + + +def require_executable(path: str, hint: str) -> Path: + resolved = path if os.path.sep in path else shutil.which(path) + if resolved is None or not os.access(resolved, os.X_OK): + raise StackError(f"{path} is not executable; {hint}") + # Keep the caller's path (absolute, symlinks intact): a venv's bin/python is + # a symlink to the base interpreter and must stay the venv's entry point. + return Path(os.path.abspath(resolved)) + + +def write_private(path: Path, content: str) -> None: + path.write_text(content, encoding="utf-8") + os.chmod(path, 0o600) + + +def command_up(args: argparse.Namespace) -> int: + stack_dir = Path(args.stack_dir).resolve() + if stack_dir.exists() and any(stack_dir.iterdir()): + raise StackError(f"stack directory is not empty: {stack_dir}") + python = require_executable(args.python, "pass --python from a venv that has the nokv wheel installed") + if not python.is_absolute(): + raise StackError("--python must be an absolute path") + binary = Path(args.nokv_binary).resolve() if args.nokv_binary else REPO / "target" / "release" / "nokv" + if args.build: + cargo = require_executable("cargo", "install the Rust toolchain or pass --nokv-binary") + run_checked( + [str(cargo), "build", "--release", "--locked", "-p", "nokv", "--bin", "nokv"], + timeout=3600, + what="cargo build", + ) + if not binary.is_file() or not os.access(binary, os.X_OK): + raise StackError(f"nokv binary is missing: {binary} (pass --build or --nokv-binary)") + etcd = require_executable(args.etcd_bin, "install etcd (for example `brew install etcd`) or pass --etcd-bin") + etcdctl = require_executable(args.etcdctl_bin, "install etcd or pass --etcdctl-bin") + + sdk = python_json(python, SDK_VERSION_SNIPPET, {}, timeout=60, what="NoKV SDK import") + binary_version = run_checked([str(binary), "--version"], timeout=30, what="nokv --version").split()[-1] + if sdk["version"] != binary_version: + raise StackError( + f"the nokv binary reports {binary_version} but the SDK under --python reports {sdk['version']};" + " LoopX's helper requires wheel and owner from the same release" + ) + if not sdk["fence"]: + raise StackError("the SDK under --python lacks WorkspaceIncarnationMismatch; LoopX pins a fenced wheel") + + object_store = args.object_store + if object_store == "auto": + object_store = "rustfs" if shutil.which("docker") and shutil.which("aws") else "moto" + if object_store == "moto": + probe = subprocess.run([str(python), "-c", "import moto, boto3"], capture_output=True, text=True, check=False) + if probe.returncode != 0: + raise StackError("--object-store moto needs `pip install 'moto[server]' boto3` in the --python venv") + elif not START_RUSTFS.is_file(): + raise StackError(f"missing {START_RUSTFS}") + + identity = fresh_identity() + endpoints = fresh_endpoints() + logs = stack_dir / "logs" + logs.mkdir(parents=True) + (stack_dir / "etcd").mkdir() + state: dict[str, Any] = { + "schema": "nokv.loopx_stage2a_stack.v1", + "stack_dir": str(stack_dir), + "object_store": object_store, + "endpoints": dataclasses.asdict(endpoints), + "pids": {}, + "binary": str(binary), + "binary_sha256": hashlib.sha256(binary.read_bytes()).hexdigest(), + "sdk": sdk, + "workbench_sha256_prefix": digest(identity.workbench)[:12], + } + started: list[subprocess.Popen[str]] = [] + + def persist() -> None: + write_private(stack_dir / STATE_FILE, json.dumps(state, indent=2, sort_keys=True) + "\n") + + try: + etcd_process = start_process( + etcd_command(etcd, stack_dir / "etcd", identity.node, endpoints), logs / "etcd.log" + ) + started.append(etcd_process) + state["pids"]["etcd"] = etcd_process.pid + persist() + wait_etcd(etcdctl, endpoints.etcd, etcd_process, timeout=30) + + if object_store == "moto": + moto_process = start_process(moto_command(python, endpoints), logs / "moto.log") + started.append(moto_process) + state["pids"]["moto"] = moto_process.pid + persist() + wait_tcp(moto_process, endpoints.object_port, 60, "moto") + python_json( + python, + MOTO_BUCKET_SNIPPET, + { + "endpoint": endpoints.object_store, + "region": OBJECT_REGION, + "access_key": identity.access_key, + "secret_key": identity.secret_key, + "bucket": identity.bucket, + }, + timeout=60, + what="moto bucket creation", + ) + else: + container = f"lx-stage2a-rustfs-{identity.node[3:]}" + state["rustfs_container"] = container + persist() + env = os.environ.copy() + env.update( + { + "NOKV_WORKBENCH_RUSTFS_CONTAINER": container, + "NOKV_WORKBENCH_RUSTFS_PORT": str(endpoints.object_port), + "NOKV_WORKBENCH_RUSTFS_CONSOLE_PORT": str(endpoints.object_console_port), + "NOKV_WORKBENCH_RUSTFS_DATA_DIR": str(stack_dir / "rustfs"), + "NOKV_WORKBENCH_S3_ACCESS_KEY_ID": identity.access_key, + "NOKV_WORKBENCH_S3_SECRET_ACCESS_KEY": identity.secret_key, + "NOKV_WORKBENCH_S3_BUCKET": identity.bucket, + } + ) + env.pop("NOKV_WORKBENCH_RUSTFS_VOLUME", None) + result = subprocess.run( + ["bash", str(START_RUSTFS)], env=env, capture_output=True, text=True, timeout=600, check=False + ) + with (logs / "rustfs.log").open("a", encoding="utf-8") as handle: + handle.write(result.stdout + result.stderr) + if result.returncode != 0: + raise StackError(f"start_rustfs.sh failed (code {result.returncode}); see {logs / 'rustfs.log'}") + + run_checked( + provision_command(binary, identity, endpoints), + timeout=300, + what="nokv provision", + log=logs / "provision.log", + ) + owner = start_process( + server_command(binary, identity, endpoints, stack_dir / "metadata"), logs / "owner.log" + ) + started.append(owner) + state["pids"]["owner"] = owner.pid + persist() + wait_tcp(owner, endpoints.owner_port, 120, "nokv owner") + + created = python_json( + python, + CREATE_WORKBENCH_SNIPPET, + { + "root_id": identity.root_id, + "etcd_endpoints": [endpoints.etcd], + "etcd_prefix": identity.etcd_prefix, + "lease_ttl": ETCD_LEASE_TTL_SECONDS, + "bucket": identity.bucket, + "region": OBJECT_REGION, + "object_root": identity.object_root, + "object_endpoint": endpoints.object_store, + "access_key": identity.access_key, + "secret_key": identity.secret_key, + "workbench_root": WORKBENCH_ROOT, + "workbench": identity.workbench, + }, + timeout=180, + what="workbench creation through the SDK", + ) + incarnations = created.get("incarnations") + if not isinstance(incarnations, list) or len(incarnations) != 1: + raise StackError(f"expected exactly one incarnation for the new workbench, found {incarnations!r}") + state["workbench_incarnation_sha256_prefix"] = digest(str(incarnations[0]))[:12] + + config = client_config(identity, endpoints) + env_values = live_env(identity, endpoints, config_path=stack_dir / CLIENT_CONFIG_FILE, python=python) + collisions = privacy_collisions(config, env_values) + if collisions: + raise StackError("generated identity collides with ladder vocabulary: " + "; ".join(collisions)) + write_private(stack_dir / CLIENT_CONFIG_FILE, json.dumps(config, indent=2) + "\n") + write_private(stack_dir / LIVE_ENV_FILE, render_env(env_values)) + state["client_config_sha256"] = digest(canonical_json(config)) + commands = ladder_commands(stack_dir, args.loopx_python) + write_private( + stack_dir / NEXT_STEPS_FILE, + "# Run from a LoopX checkout whose Python has loopx importable.\n" + "\n".join(commands) + "\n", + ) + state["ready"] = True + persist() + except BaseException as error: + exit_codes = {process.pid: process.poll() for process in started} + try: + for process in reversed(started): + stop_pid(process.pid) + if state.get("rustfs_container"): + subprocess.run( + ["docker", "rm", "-f", state["rustfs_container"]], capture_output=True, check=False + ) + finally: + state["ready"] = False + state["failure"] = {"error": str(error), "exit_codes_before_cleanup": exit_codes} + persist() + raise + + summary = { + "stack_dir": str(stack_dir), + "object_store": object_store, + "owner": endpoints.owner, + "pids": state["pids"], + "binary_sha256": state["binary_sha256"], + "sdk": sdk, + "client_config_sha256": state["client_config_sha256"], + "workbench_sha256_prefix": state["workbench_sha256_prefix"], + "files": {name: str(stack_dir / name) for name in (CLIENT_CONFIG_FILE, LIVE_ENV_FILE, NEXT_STEPS_FILE)}, + } + print(json.dumps(summary, indent=2, sort_keys=True)) + print("\nnext steps (also in next-steps.txt):", file=sys.stderr) + for line in commands: + print(f" {line}", file=sys.stderr) + return 0 + + +def load_state(stack_dir: Path) -> dict[str, Any]: + path = stack_dir / STATE_FILE + if not path.is_file(): + raise StackError(f"no {STATE_FILE} in {stack_dir}; nothing to act on") + state = json.loads(path.read_text(encoding="utf-8")) + if state.get("schema") != "nokv.loopx_stage2a_stack.v1": + raise StackError(f"unknown state schema in {path}") + return state + + +def pid_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +def command_down(args: argparse.Namespace) -> int: + stack_dir = Path(args.stack_dir).resolve() + state = load_state(stack_dir) + for name in ("owner", "moto", "etcd"): + pid = state.get("pids", {}).get(name) + if isinstance(pid, int): + stop_pid(pid) + container = state.get("rustfs_container") + if container: + subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False) + if args.purge: + shutil.rmtree(stack_dir) + else: + state["ready"] = False + write_private(stack_dir / STATE_FILE, json.dumps(state, indent=2, sort_keys=True) + "\n") + print(json.dumps({"stack_dir": str(stack_dir), "stopped": True, "purged": bool(args.purge)})) + return 0 + + +def command_status(args: argparse.Namespace) -> int: + stack_dir = Path(args.stack_dir).resolve() + state = load_state(stack_dir) + report = { + "stack_dir": str(stack_dir), + "ready": bool(state.get("ready")), + "object_store": state.get("object_store"), + "owner": f"127.0.0.1:{state['endpoints']['owner_port']}", + "processes": {name: pid_alive(pid) for name, pid in state.get("pids", {}).items()}, + "client_config_sha256": state.get("client_config_sha256"), + "workbench_sha256_prefix": state.get("workbench_sha256_prefix"), + } + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if all(report["processes"].values()) and report["ready"] else 1 + + +def command_plan(args: argparse.Namespace) -> int: + stack_dir = Path(args.stack_dir).resolve() + binary = Path(args.nokv_binary).resolve() if args.nokv_binary else REPO / "target" / "release" / "nokv" + object_store = args.object_store + if object_store == "auto": + object_store = "rustfs" if shutil.which("docker") and shutil.which("aws") else "moto" + print( + json.dumps( + plan( + binary=binary, + python=Path(args.python), + stack_dir=stack_dir, + identity=fresh_identity(), + endpoints=fresh_endpoints(), + object_store=object_store, + ), + indent=2, + sort_keys=True, + ) + ) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + commands = parser.add_subparsers(dest="command", required=True) + + def common(sub: argparse.ArgumentParser, *, needs_python: bool) -> None: + sub.add_argument("--stack-dir", required=True, help="empty directory that receives data, logs and the two private files") + if needs_python: + sub.add_argument("--python", required=True, help="absolute path to a Python whose venv has the matching nokv wheel") + sub.add_argument("--nokv-binary", help="serving binary (default: target/release/nokv of this checkout)") + sub.add_argument("--object-store", choices=("auto", "rustfs", "moto"), default="auto") + sub.add_argument("--loopx-python", default="python", help="interpreter name to print in the LoopX ladder commands") + + up = commands.add_parser("up", help="start etcd, the object store and one owner; create one workbench") + common(up, needs_python=True) + up.add_argument("--build", action="store_true", help="cargo build --release -p nokv --bin nokv first") + up.add_argument("--etcd-bin", default="etcd") + up.add_argument("--etcdctl-bin", default="etcdctl") + up.set_defaults(run=command_up) + + planned = commands.add_parser("plan", help="print the redacted plan without starting anything") + common(planned, needs_python=True) + planned.set_defaults(run=command_plan) + + down = commands.add_parser("down", help="stop every process and container of a stack") + common(down, needs_python=False) + down.add_argument("--purge", action="store_true", help="also delete the stack directory") + down.set_defaults(run=command_down) + + status = commands.add_parser("status", help="report whether the stack's processes are alive") + common(status, needs_python=False) + status.set_defaults(run=command_status) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + return int(args.run(args)) + except StackError as error: + print(f"loopx_stage2a_stack: {error}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/workbench/loopx_stage2a_stack_test.py b/scripts/workbench/loopx_stage2a_stack_test.py new file mode 100644 index 000000000..791237488 --- /dev/null +++ b/scripts/workbench/loopx_stage2a_stack_test.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +"""Unit tests for the LoopX Stage 2A stack bring-up script (no processes are started).""" + +from __future__ import annotations + +import io +import json +import re +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path + +import loopx_stage2a_stack as stack + +HEX32 = re.compile(r"^[0-9a-f]{32}$") + + +def counted_hex() -> stack.StackIdentity: + counter = iter(range(1, 100)) + + def token_hex(length: int) -> str: + value = next(counter) + return f"{value:0{length * 2}x}" + + return stack.fresh_identity(token_hex) + + +def fixed_endpoints() -> stack.Endpoints: + ports = iter((23791, 23801, 29001, 29011, 27101)) + return stack.fresh_endpoints(lambda: next(ports)) + + +class IdentityTest(unittest.TestCase): + def test_ids_are_hex32_and_names_carry_non_colliding_prefixes(self) -> None: + identity = counted_hex() + for value in (identity.root_id, identity.shard_id, identity.agent_id): + self.assertRegex(value, HEX32) + self.assertTrue(identity.etcd_prefix.startswith("/lx-")) + self.assertTrue(identity.bucket.startswith("lx-")) + self.assertTrue(identity.object_root.startswith("rt-")) + self.assertTrue(identity.access_key.startswith("ak")) + self.assertTrue(identity.secret_key.startswith("sk")) + self.assertTrue(identity.workbench.startswith("wb")) + self.assertTrue(identity.node.startswith("nd-")) + + def test_random_identity_never_collides_with_ladder_vocabulary(self) -> None: + for _ in range(200): + identity = stack.fresh_identity() + endpoints = fixed_endpoints() + config = stack.client_config(identity, endpoints) + env = stack.live_env( + identity, endpoints, config_path=Path("/tmp/x/nokv-client.json"), python=Path("/tmp/x/bin/python") + ) + self.assertEqual(stack.privacy_collisions(config, env), []) + + def test_privacy_check_reports_a_leaf_inside_ladder_vocabulary(self) -> None: + identity = counted_hex() + endpoints = fixed_endpoints() + config = stack.client_config(identity, endpoints) + config["object_store"]["bucket"] = "nokv_live" + env = stack.live_env(identity, endpoints, config_path=Path("/tmp/c.json"), python=Path("/tmp/p")) + collisions = stack.privacy_collisions(config, env) + self.assertEqual( + collisions, + [ + "'nokv_live' is a substring of ladder vocabulary 's0.nokv_live_matrix'", + "'nokv_live' is a substring of ladder vocabulary 's2a.nokv_live_qualification'", + ], + ) + + +class CommandShapeTest(unittest.TestCase): + def setUp(self) -> None: + self.identity = counted_hex() + self.endpoints = fixed_endpoints() + self.binary = Path("/opt/nokv/bin/nokv") + + def test_provision_carries_agent_id_and_positional_shard(self) -> None: + argv = stack.provision_command(self.binary, self.identity, self.endpoints) + self.assertEqual(argv[0], str(self.binary)) + self.assertEqual(argv[-2:], ["provision", self.identity.shard_id]) + self.assertIn("--agent-id", argv) + self.assertEqual(argv[argv.index("--agent-id") + 1], self.identity.agent_id) + self.assertEqual(argv[argv.index("--etcd-endpoint") + 1], "http://127.0.0.1:23791") + self.assertEqual(argv[argv.index("--etcd-key-prefix") + 1], self.identity.etcd_prefix) + + def test_serve_creates_fresh_metadata_and_never_carries_an_agent_id(self) -> None: + argv = stack.server_command(self.binary, self.identity, self.endpoints, Path("/stack/metadata")) + self.assertEqual(argv[-1], "serve") + self.assertNotIn("--agent-id", argv) + self.assertEqual(argv[argv.index("--metadata-create") + 1], "/stack/metadata") + self.assertEqual(argv[argv.index("--bind") + 1], "127.0.0.1:27101") + self.assertEqual(argv[argv.index("--advertise-endpoint") + 1], "127.0.0.1:27101") + self.assertEqual(argv[argv.index("--node-id") + 1], self.identity.node) + self.assertEqual(argv[argv.index("--lifecycle-interval-millis") + 1], "100") + + def test_etcd_is_a_single_new_member_on_loopback(self) -> None: + argv = stack.etcd_command(Path("/usr/local/bin/etcd"), Path("/stack/etcd"), "nd-01", self.endpoints) + self.assertEqual(argv[argv.index("--listen-client-urls") + 1], "http://127.0.0.1:23791") + self.assertEqual(argv[argv.index("--initial-cluster") + 1], "nd-01=http://127.0.0.1:23801") + self.assertEqual(argv[argv.index("--initial-cluster-state") + 1], "new") + + def test_moto_listens_on_the_object_port_only(self) -> None: + argv = stack.moto_command(Path("/venv/bin/python"), self.endpoints) + self.assertEqual(argv, ["/venv/bin/python", "-m", "moto.server", "-H", "127.0.0.1", "-p", "29001"]) + + def test_secret_flag_values_are_redacted(self) -> None: + argv = stack.redact_argv(stack.provision_command(self.binary, self.identity, self.endpoints)) + self.assertNotIn(self.identity.secret_key, argv) + self.assertTrue(any(item.startswith(" None: + self.identity = counted_hex() + self.endpoints = fixed_endpoints() + + def test_client_config_has_exactly_the_helper_keys(self) -> None: + config = stack.client_config(self.identity, self.endpoints) + self.assertEqual(set(config), {"root_id", "routing", "object_store", "workbench_root"}) + self.assertEqual(set(config["routing"]), {"kind", "endpoints", "key_prefix", "lease_ttl_seconds"}) + self.assertEqual(config["routing"]["kind"], "etcd") + self.assertEqual(config["routing"]["endpoints"], ["http://127.0.0.1:23791"]) + self.assertEqual( + set(config["object_store"]), + {"kind", "bucket", "region", "root", "endpoint", "access_key_id", "secret_access_key"}, + ) + self.assertEqual(config["object_store"]["kind"], "s3") + self.assertEqual(config["object_store"]["endpoint"], "http://127.0.0.1:29001") + self.assertEqual(config["workbench_root"], stack.WORKBENCH_ROOT) + + def test_live_env_covers_both_ladder_gates(self) -> None: + env = stack.live_env( + self.identity, self.endpoints, config_path=Path("/stack/nokv-client.json"), python=Path("/venv/bin/python") + ) + self.assertEqual( + list(env), + [ + "NOKV_COORDINATION_LIVE", + "NOKV_ETCD", + "NOKV_ETCD_PREFIX", + "NOKV_ROOT_ID", + "NOKV_BUCKET", + "NOKV_OBJECT_ENDPOINT", + "NOKV_OBJECT_ROOT", + "NOKV_OBJECT_KEY", + "NOKV_OBJECT_SECRET", + "LOOPX_NOKV_AUTHORITY_LIVE", + "LOOPX_NOKV_AUTHORITY_CONFIG_JSON", + "LOOPX_NOKV_AUTHORITY_PYTHON", + "LOOPX_NOKV_AUTHORITY_WORKBENCH", + ], + ) + self.assertEqual(env["NOKV_COORDINATION_LIVE"], "1") + self.assertEqual(env["LOOPX_NOKV_AUTHORITY_LIVE"], "1") + self.assertEqual(env["LOOPX_NOKV_AUTHORITY_CONFIG_JSON"], "/stack/nokv-client.json") + self.assertEqual(env["LOOPX_NOKV_AUTHORITY_WORKBENCH"], self.identity.workbench) + + def test_env_rendering_is_shell_safe(self) -> None: + rendered = stack.render_env({"A": "plain", "B": "with space", "C": "quo'te"}) + self.assertEqual(rendered, "export A=plain\nexport B='with space'\nexport C='quo'\"'\"'te'\n") + + def test_ladder_commands_source_the_env_and_name_both_live_rows(self) -> None: + commands = stack.ladder_commands(Path("/stack"), "/venv/bin/python") + self.assertEqual(len(commands), 3) + self.assertIn("source /stack/live.env", commands[0]) + self.assertIn("--row s0.nokv_live_matrix --row s2a.nokv_live_qualification", commands[1]) + self.assertIn("--allow-unverified", commands[2]) + + +class PlanTest(unittest.TestCase): + def test_plan_is_json_and_carries_no_secret(self) -> None: + identity = counted_hex() + planned = stack.plan( + binary=Path("/opt/nokv/bin/nokv"), + python=Path("/venv/bin/python"), + stack_dir=Path("/stack"), + identity=identity, + endpoints=fixed_endpoints(), + object_store="moto", + ) + text = json.dumps(planned) + self.assertNotIn(identity.secret_key, text) + self.assertNotIn(identity.workbench, text) + self.assertEqual(planned["privacy_collisions"], []) + self.assertEqual(planned["files"], ["nokv-client.json", "live.env", "stack.json", "next-steps.txt"]) + self.assertEqual(planned["commands"]["serve"][-1], "serve") + self.assertEqual(len(planned["client_config_sha256"]), 64) + + def test_plan_subcommand_prints_json_without_touching_the_stack_dir(self) -> None: + with tempfile.TemporaryDirectory() as root: + stack_dir = Path(root) / "stack" + out = io.StringIO() + with redirect_stdout(out): + code = stack.main( + [ + "plan", + "--stack-dir", + str(stack_dir), + "--python", + "/venv/bin/python", + "--object-store", + "moto", + ] + ) + self.assertEqual(code, 0) + self.assertFalse(stack_dir.exists()) + self.assertEqual(json.loads(out.getvalue())["object_store"], "moto") + + +class FailClosedTest(unittest.TestCase): + def test_down_and_status_refuse_a_directory_without_state(self) -> None: + with tempfile.TemporaryDirectory() as root: + for command in ("down", "status"): + err = io.StringIO() + with redirect_stderr(err): + code = stack.main([command, "--stack-dir", root]) + self.assertEqual(code, 2) + self.assertIn("nothing to act on", err.getvalue()) + + def test_up_refuses_a_non_empty_stack_dir(self) -> None: + with tempfile.TemporaryDirectory() as root: + (Path(root) / "leftover").write_text("x", encoding="utf-8") + err = io.StringIO() + with redirect_stderr(err): + code = stack.main(["up", "--stack-dir", root, "--python", "/venv/bin/python"]) + self.assertEqual(code, 2) + self.assertIn("not empty", err.getvalue()) + + def test_status_rejects_an_unknown_state_schema(self) -> None: + with tempfile.TemporaryDirectory() as root: + (Path(root) / stack.STATE_FILE).write_text(json.dumps({"schema": "other"}), encoding="utf-8") + err = io.StringIO() + with redirect_stderr(err): + code = stack.main(["status", "--stack-dir", root]) + self.assertEqual(code, 2) + self.assertIn("unknown state schema", err.getvalue()) + + +if __name__ == "__main__": + unittest.main()