544 lines
20 KiB
Python
544 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from importlib import util
|
|
from pathlib import Path
|
|
from types import ModuleType, SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from ops import gcp_leoclean_nosend_prompt_ingress as payload
|
|
from ops import run_gcp_leoclean_nosend_prompt_ingress as operator
|
|
from scripts import working_leo_open_ended_benchmark as benchmark
|
|
|
|
RUNNER_REVISION = "a" * 40
|
|
RUNTIME_REVISION = "b" * 40
|
|
IMAGE_DIGEST = "sha256:" + "c" * 64
|
|
CONFIG_DIGEST = "sha256:" + "d" * 64
|
|
RELEASE_SHA256 = "e" * 64
|
|
REQUEST_ID = "m5-abcdef1234567890"
|
|
PROVIDER_SECRET_NAME = "livingip-openrouter-staging"
|
|
|
|
|
|
def _plan(tmp_path: Path) -> dict[str, Any]:
|
|
return operator.prepare_plan(
|
|
request_id=REQUEST_ID,
|
|
runner_revision=RUNNER_REVISION,
|
|
runtime_revision=RUNTIME_REVISION,
|
|
image_digest=IMAGE_DIGEST,
|
|
config_digest=CONFIG_DIGEST,
|
|
release_sha256=RELEASE_SHA256,
|
|
provider_secret_name=PROVIDER_SECRET_NAME,
|
|
output=tmp_path / "result.json",
|
|
)
|
|
|
|
|
|
def _remote_result(plan: dict[str, Any]) -> dict[str, Any]:
|
|
service = {
|
|
"container_id": "f" * 64,
|
|
"image_reference": plan["bindings"]["image_reference"],
|
|
"image_config_digest": CONFIG_DIGEST,
|
|
"release_sha256": RELEASE_SHA256,
|
|
"role": "leoclean-gcp-nosend",
|
|
"running": True,
|
|
"health": "healthy",
|
|
"restart_count": 0,
|
|
"started_at": "2026-07-24T00:00:00Z",
|
|
"entrypoint": [
|
|
"/opt/livingip/leoclean-nosend/venv/bin/python",
|
|
"/opt/livingip/leoclean-nosend/entrypoint.py",
|
|
],
|
|
"command": ["service"],
|
|
"auto_remove": True,
|
|
"read_only_rootfs": True,
|
|
"cap_add": ["CHOWN", "SETGID", "SETPCAP", "SETUID"],
|
|
"cap_drop": ["ALL"],
|
|
"pids_limit": 512,
|
|
"security_opt": ["no-new-privileges:true"],
|
|
"network_names": ["bridge"],
|
|
"tmpfs_paths": ["/run/leoclean-profile", "/tmp", "/var/lib/postgresql/data"],
|
|
"published_port_count": 0,
|
|
"provider_or_transport_env_present": False,
|
|
}
|
|
stable = {
|
|
"schema": operator.REMOTE_RESULT_SCHEMA,
|
|
"status": "pass",
|
|
"request_id": REQUEST_ID,
|
|
"target": {
|
|
"project": operator.PROJECT,
|
|
"zone": operator.ZONE,
|
|
"instance": operator.INSTANCE,
|
|
},
|
|
"bindings": {
|
|
"image_reference": plan["bindings"]["image_reference"],
|
|
"image_config_digest": CONFIG_DIGEST,
|
|
"release_sha256": RELEASE_SHA256,
|
|
"runtime_revision": RUNTIME_REVISION,
|
|
"ingress_sha256": plan["ingress"]["payload_sha256"],
|
|
"prompt_catalog_sha256": plan["ingress"]["prompt_catalog_sha256"],
|
|
},
|
|
"service_before": service,
|
|
"service_after": service,
|
|
"service_unchanged": True,
|
|
"turns": [{"prompt_id": prompt_id} for prompt_id in operator.PROMPT_IDS],
|
|
"cleanup": {
|
|
"transient_containers_remaining": [],
|
|
"staged_payload_removed": True,
|
|
},
|
|
"authority": dict(operator.NO_SEND_AUTHORITY),
|
|
"claim_ceiling": (
|
|
"Live execution proves the exact immutable image answered the fixed prompts through "
|
|
"the external hash-bound private ingress. Scoring and behavioral parity are separate."
|
|
),
|
|
}
|
|
return {
|
|
**stable,
|
|
"receipt_sha256": operator._sha256_bytes(operator._canonical_json(stable)),
|
|
}
|
|
|
|
|
|
def test_prompt_catalog_is_the_exact_existing_dc_catalog() -> None:
|
|
expected = [
|
|
{
|
|
"id": row["id"],
|
|
"dimension": row["dimension"],
|
|
"message": row["message"],
|
|
}
|
|
for row in benchmark.M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS
|
|
]
|
|
assert list(payload.PROMPTS) == expected
|
|
assert tuple(row["id"] for row in payload.PROMPTS) == (
|
|
"DC-01",
|
|
"DC-02",
|
|
"DC-03",
|
|
"DC-04",
|
|
"DC-05",
|
|
"DC-06",
|
|
)
|
|
assert len(payload.prompt_catalog_sha256()) == 64
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw",
|
|
[
|
|
b"",
|
|
b"short\n",
|
|
b"x" * (payload.MAX_SECRET_BYTES + 2),
|
|
b"x" * 20 + b"\nextra",
|
|
b"x" * 20 + b"\r",
|
|
b"x" * 20 + b"\0",
|
|
b"\xff" * 20,
|
|
],
|
|
)
|
|
def test_provider_secret_protocol_fails_closed(raw: bytes) -> None:
|
|
with pytest.raises(payload.PromptIngressError):
|
|
payload.read_provider_secret(io.BytesIO(raw))
|
|
|
|
|
|
def test_provider_secret_protocol_accepts_one_bounded_stdin_value() -> None:
|
|
secret = "sk-or-v1-" + "x" * 32
|
|
assert payload.read_provider_secret(io.BytesIO((secret + "\n").encode())) == secret
|
|
|
|
|
|
def test_tool_trace_records_only_hashes_and_terminal_verb() -> None:
|
|
messages = [
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [
|
|
{
|
|
"id": "call-private",
|
|
"function": {
|
|
"name": "terminal",
|
|
"arguments": json.dumps({"command": "teleo-kb status --format json"}),
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": "call-private",
|
|
"content": json.dumps({"output": "private database content", "exit_code": 0}),
|
|
},
|
|
]
|
|
trace = payload._tool_trace(messages)
|
|
assert trace == [
|
|
{
|
|
"call_id_sha256": payload._sha256_bytes(b"call-private"),
|
|
"tool": "terminal",
|
|
"arguments_sha256": payload._sha256_bytes(
|
|
json.dumps({"command": "teleo-kb status --format json"}).encode()
|
|
),
|
|
"terminal_command": "status",
|
|
"output_sha256": payload._sha256_bytes(
|
|
json.dumps({"output": "private database content", "exit_code": 0}).encode()
|
|
),
|
|
"exit_code": 0,
|
|
}
|
|
]
|
|
assert "private database content" not in json.dumps(trace)
|
|
|
|
|
|
def test_plan_requires_full_immutable_bindings(tmp_path: Path) -> None:
|
|
plan = _plan(tmp_path)
|
|
assert plan["mode"] == "dry_run"
|
|
assert plan["bindings"]["image_reference"] == f"{operator.IMAGE_REPOSITORY}@{IMAGE_DIGEST}"
|
|
assert plan["ingress"]["prompt_ids"] == list(operator.PROMPT_IDS)
|
|
assert plan["ingress"]["container_count"] == 6
|
|
assert plan["ingress"]["provider_secret_name"] == PROVIDER_SECRET_NAME
|
|
assert plan["ingress"]["claim_ceiling"] == operator.CLAIM_CEILING
|
|
assert plan["authority"] == {
|
|
"gcp_read": True,
|
|
"transient_container_execution": True,
|
|
"managed_service_mutation": False,
|
|
"database_write": False,
|
|
"telegram": False,
|
|
"vps": False,
|
|
"production_traffic": False,
|
|
"public_endpoint": False,
|
|
"image_rebuild": False,
|
|
}
|
|
with pytest.raises(operator.PromptIngressPlanError, match="image digest"):
|
|
operator.prepare_plan(
|
|
request_id=REQUEST_ID,
|
|
runner_revision=RUNNER_REVISION,
|
|
runtime_revision=RUNTIME_REVISION,
|
|
image_digest="latest",
|
|
config_digest=CONFIG_DIGEST,
|
|
release_sha256=RELEASE_SHA256,
|
|
provider_secret_name=PROVIDER_SECRET_NAME,
|
|
output=tmp_path / "bad.json",
|
|
)
|
|
|
|
|
|
def test_rendered_remote_contract_is_transient_private_and_stdin_only(tmp_path: Path) -> None:
|
|
compile(operator.REMOTE_EXECUTOR_SOURCE, "<remote-executor>", "exec")
|
|
compile(operator.CLEANUP_SOURCE, "<cleanup>", "exec")
|
|
plan = _plan(tmp_path)
|
|
commands = operator.build_gcloud_commands(
|
|
request_id=REQUEST_ID,
|
|
bindings=plan["bindings"],
|
|
payload_sha256=plan["ingress"]["payload_sha256"],
|
|
prompt_catalog_sha256=plan["ingress"]["prompt_catalog_sha256"],
|
|
ssh_key=(tmp_path / "key").resolve(),
|
|
)
|
|
scp, ssh, cleanup = commands
|
|
assert "--tunnel-through-iap" in scp
|
|
assert "--tunnel-through-iap" in ssh
|
|
assert "--tunnel-through-iap" in cleanup
|
|
rendered = operator.REMOTE_EXECUTOR_SOURCE
|
|
assert (
|
|
'docker = ["/usr/bin/docker", "--host=unix:///var/run/docker.sock", '
|
|
'"--config=/var/lib/livingip/leoclean-nosend/docker-config"]'
|
|
) in rendered
|
|
for required in (
|
|
'"--rm"',
|
|
'"--interactive"',
|
|
'"--pull=never"',
|
|
'"--read-only"',
|
|
'"--cap-drop=ALL"',
|
|
'"--security-opt=no-new-privileges:true"',
|
|
'"--network=bridge"',
|
|
'"--label=livingip.container.role=leoclean-m5-prompt-ingress"',
|
|
'"--entrypoint=/opt/livingip/leoclean-nosend/venv/bin/python"',
|
|
):
|
|
assert required in rendered
|
|
for forbidden in (
|
|
"--publish",
|
|
"-p=",
|
|
"docker exec",
|
|
"systemctl",
|
|
"--env OPENROUTER_API_KEY",
|
|
"--env=OPENROUTER_API_KEY",
|
|
):
|
|
assert forbidden not in rendered
|
|
assert "secret + b\"\\n\"" in rendered
|
|
assert "livingip.container.role=leoclean-gcp-nosend\"" not in rendered
|
|
assert "/usr/bin/python3.11" in ssh[-1]
|
|
cleanup_source = operator.CLEANUP_SOURCE
|
|
assert "docker" in cleanup_source
|
|
assert "livingip.m5.request=" in cleanup_source
|
|
assert "leoclean-m5-prompt-ingress" in cleanup_source
|
|
assert "cleanup did not prove container absence" in cleanup_source
|
|
assert REQUEST_ID in cleanup[-1]
|
|
|
|
|
|
def test_remote_managed_inspect_parses_docker_array_directly() -> None:
|
|
rendered = operator.REMOTE_EXECUTOR_SOURCE
|
|
assert 'raw = command([*docker, "inspect", managed_name]).stdout' in rendered
|
|
assert 'value = json.loads(raw.decode("utf-8", "strict"))' in rendered
|
|
assert (
|
|
"require(isinstance(value, list) and len(value) == 1 and "
|
|
'isinstance(value[0], dict), "managed_inspect_invalid")'
|
|
) in rendered
|
|
assert 'strict_json(command([*docker, "inspect", managed_name])' not in rendered
|
|
|
|
|
|
def test_gcloud_environment_resolves_and_verifies_python_311_once(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
home = tmp_path / "home"
|
|
config = home / ".config" / "gcloud"
|
|
bin_dir = tmp_path / "bin"
|
|
home.mkdir(mode=0o700)
|
|
config.mkdir(parents=True, mode=0o700)
|
|
bin_dir.mkdir(mode=0o700)
|
|
uv = bin_dir / "uv"
|
|
python = bin_dir / "python3.11"
|
|
uv.touch(mode=0o755)
|
|
python.touch(mode=0o755)
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
calls.append(list(argv))
|
|
if argv == [str(uv), "python", "find", "3.11"]:
|
|
return subprocess.CompletedProcess(argv, 0, stdout=f"{python}\n".encode(), stderr=b"")
|
|
if argv == [str(python), "--version"]:
|
|
return subprocess.CompletedProcess(argv, 0, stdout=b"Python 3.11.9\n", stderr=b"")
|
|
raise AssertionError(argv)
|
|
|
|
monkeypatch.setenv("HOME", str(home))
|
|
monkeypatch.setenv("CLOUDSDK_CONFIG", str(config))
|
|
monkeypatch.setattr(operator.shutil, "which", lambda *_args, **_kwargs: str(uv))
|
|
environment = operator.pinned_gcloud_environment(runner=fake_runner)
|
|
assert environment["CLOUDSDK_PYTHON"] == str(python)
|
|
assert calls == [
|
|
[str(uv), "python", "find", "3.11"],
|
|
[str(python), "--version"],
|
|
]
|
|
|
|
|
|
def test_payload_narrows_proposal_capability_and_never_uses_a_listener() -> None:
|
|
expected_commands = {
|
|
"context",
|
|
"decision-matrix-status",
|
|
"edges",
|
|
"evidence",
|
|
"list-proposals",
|
|
"search",
|
|
"search-proposals",
|
|
"show",
|
|
"show-proposal",
|
|
"status",
|
|
}
|
|
assert expected_commands == payload.READ_ONLY_TERMINAL_COMMANDS
|
|
assert "propose-core-change" not in payload.READ_ONLY_TERMINAL_COMMANDS
|
|
assert "propose-edge" not in payload.READ_ONLY_TERMINAL_COMMANDS
|
|
source = Path(payload.__file__).read_text(encoding="utf-8")
|
|
assert "AIAgent(" in source
|
|
assert "APIServerAdapter(" not in source
|
|
assert "ResponseStore(" not in source
|
|
assert "persist_session=False" in source
|
|
assert "adapter.start(" not in source
|
|
assert "adapter.connect(" not in source
|
|
assert "Platform.TELEGRAM" not in source
|
|
assert "TELEGRAM_BOT_TOKEN" not in source
|
|
|
|
|
|
def test_m5_tool_registry_remains_exact_after_inference_boundary(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
bootstrap_path = Path("hermes-agent/runtime/leoclean-nosend/bootstrap.py").resolve()
|
|
spec = util.spec_from_file_location("test_m5_runtime_bootstrap", bootstrap_path)
|
|
assert spec is not None and spec.loader is not None
|
|
bootstrap = util.module_from_spec(spec)
|
|
spec.loader.exec_module(bootstrap)
|
|
profile = tmp_path / "profile"
|
|
profile.mkdir()
|
|
|
|
class Terminal:
|
|
def __init__(self) -> None:
|
|
self.handler: Any = None
|
|
self.schema: dict[str, Any] = {}
|
|
self.check_fn: Any = object()
|
|
self.requires_env: list[str] = ["AUTHORITY"]
|
|
|
|
terminal = Terminal()
|
|
registry = SimpleNamespace(
|
|
_tools=bootstrap._SealedToolMap(
|
|
{
|
|
"skills_list": SimpleNamespace(),
|
|
"skill_view": SimpleNamespace(),
|
|
"terminal": terminal,
|
|
}
|
|
)
|
|
)
|
|
tools_module = ModuleType("tools")
|
|
registry_module = ModuleType("tools.registry")
|
|
registry_module.registry = registry # type: ignore[attr-defined]
|
|
monkeypatch.setitem(sys.modules, "tools", tools_module)
|
|
monkeypatch.setitem(sys.modules, "tools.registry", registry_module)
|
|
monkeypatch.setattr(
|
|
bootstrap,
|
|
"_restricted_terminal_handler",
|
|
lambda *_args, **_kwargs: json.dumps({"exit_code": 126}),
|
|
)
|
|
|
|
expected = payload._install_read_only_terminal(bootstrap, profile)
|
|
assert payload._assert_read_only_tool_registry(bootstrap, profile) == expected
|
|
|
|
terminal.handler = lambda *_args, **_kwargs: "{}"
|
|
with pytest.raises(payload.PromptIngressError, match="handler identity drifted"):
|
|
payload._assert_read_only_tool_registry(bootstrap, profile)
|
|
|
|
|
|
def test_execute_plan_keeps_secret_out_of_argv_and_writes_private_receipt(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
os.chmod(tmp_path, 0o700)
|
|
output = tmp_path / "result.json"
|
|
plan = _plan(tmp_path)
|
|
remote = _remote_result(plan)
|
|
secret = b"sk-or-v1-" + b"x" * 32
|
|
calls: list[tuple[list[str], bytes | None]] = []
|
|
|
|
def fake_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
input_bytes = kwargs.get("input")
|
|
calls.append((list(argv), input_bytes))
|
|
if "secrets" in argv:
|
|
return subprocess.CompletedProcess(argv, 0, stdout=secret + b"\n", stderr=b"")
|
|
if "scp" in argv:
|
|
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
|
|
if "ssh" in argv:
|
|
return subprocess.CompletedProcess(
|
|
argv,
|
|
0,
|
|
stdout=json.dumps(remote, separators=(",", ":"), sort_keys=True).encode(),
|
|
stderr=b"",
|
|
)
|
|
raise AssertionError(argv)
|
|
|
|
monkeypatch.setattr(operator.iap, "validate_clean_revision", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(operator.iap, "validate_remote_main", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(operator.iap, "_cleanup_local_ssh_key", lambda *_args, **_kwargs: None)
|
|
receipt = operator.execute_plan(
|
|
plan,
|
|
output=output,
|
|
runner=fake_runner,
|
|
git_runner=fake_runner,
|
|
gcloud_env={"PATH": "/usr/bin:/bin"},
|
|
)
|
|
assert receipt["status"] == "pass"
|
|
assert stat_mode(output) == 0o600
|
|
assert json.loads(output.read_text(encoding="utf-8")) == receipt
|
|
flattened_argv = "\n".join("\0".join(argv) for argv, _input in calls).encode()
|
|
assert secret not in flattened_argv
|
|
secret_calls = [argv for argv, _input in calls if "secrets" in argv]
|
|
assert len(secret_calls) == 1
|
|
assert not any(argument.startswith("--out-file") for argument in secret_calls[0])
|
|
ssh_inputs = [input_bytes for argv, input_bytes in calls if "ssh" in argv]
|
|
assert ssh_inputs == [secret + b"\n"]
|
|
assert receipt["provider_secret_name"] == PROVIDER_SECRET_NAME
|
|
|
|
|
|
def stat_mode(path: Path) -> int:
|
|
return path.stat().st_mode & 0o777
|
|
|
|
|
|
def test_remote_result_rejects_service_or_prompt_drift(tmp_path: Path) -> None:
|
|
plan = _plan(tmp_path)
|
|
result = _remote_result(plan)
|
|
operator._validate_remote_result(plan, result)
|
|
result["turns"] = list(reversed(result["turns"]))
|
|
with pytest.raises(operator.PromptIngressPlanError, match="prompt set"):
|
|
operator._validate_remote_result(plan, result)
|
|
|
|
|
|
def test_transport_failure_runs_request_bound_cleanup_once(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
os.chmod(tmp_path, 0o700)
|
|
plan = _plan(tmp_path)
|
|
secret = b"sk-or-v1-" + b"x" * 32
|
|
calls: list[tuple[list[str], bytes | None]] = []
|
|
|
|
def fake_runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
input_bytes = kwargs.get("input")
|
|
calls.append((list(argv), input_bytes))
|
|
if "secrets" in argv:
|
|
return subprocess.CompletedProcess(argv, 0, stdout=secret + b"\n", stderr=b"")
|
|
if "scp" in argv:
|
|
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
|
|
if "ssh" in argv and "cleanup did not prove container absence" in argv[-1]:
|
|
return subprocess.CompletedProcess(argv, 0, stdout=b"", stderr=b"")
|
|
if "ssh" in argv:
|
|
return subprocess.CompletedProcess(argv, 255, stdout=b"", stderr=b"transport interrupted")
|
|
raise AssertionError(argv)
|
|
|
|
monkeypatch.setattr(operator.iap, "validate_clean_revision", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(operator.iap, "validate_remote_main", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(operator.iap, "_cleanup_local_ssh_key", lambda *_args, **_kwargs: None)
|
|
with pytest.raises(operator.PromptIngressPlanError, match="IAP prompt ingress failed"):
|
|
operator.execute_plan(
|
|
plan,
|
|
output=tmp_path / "result.json",
|
|
runner=fake_runner,
|
|
git_runner=fake_runner,
|
|
gcloud_env={"PATH": "/usr/bin:/bin"},
|
|
)
|
|
ssh_calls = [(argv, value) for argv, value in calls if "ssh" in argv]
|
|
assert len(ssh_calls) == 2
|
|
assert ssh_calls[0][1] == secret + b"\n"
|
|
assert ssh_calls[1][1] is None
|
|
assert REQUEST_ID in ssh_calls[1][0][-1]
|
|
|
|
|
|
def test_remote_compact_receipt_hash_is_accepted_across_boundary(tmp_path: Path) -> None:
|
|
plan = _plan(tmp_path)
|
|
result = _remote_result(plan)
|
|
stable = {key: item for key, item in result.items() if key != "receipt_sha256"}
|
|
remote_hash = operator._sha256_bytes(
|
|
(json.dumps(stable, allow_nan=False, separators=(",", ":"), sort_keys=True) + "\n").encode()
|
|
)
|
|
assert result["receipt_sha256"] == remote_hash
|
|
operator._validate_remote_result(plan, result)
|
|
|
|
|
|
def test_remote_failure_requires_hash_cleanup_authority_and_service_stability(tmp_path: Path) -> None:
|
|
plan = _plan(tmp_path)
|
|
service = _remote_result(plan)["service_before"]
|
|
stable = {
|
|
"schema": operator.REMOTE_FAILURE_SCHEMA,
|
|
"status": "fail",
|
|
"request_id": REQUEST_ID,
|
|
"failure_stage": "prompt_dc01",
|
|
"failure_reason": "prompt_ingress/database status read failed",
|
|
"service_before": service,
|
|
"service_after": service,
|
|
"completed_prompt_ids": [],
|
|
"turn_failure": None,
|
|
"cleanup": {
|
|
"transient_containers_remaining": [],
|
|
"staged_payload_removed": True,
|
|
},
|
|
"authority": dict(operator.NO_SEND_AUTHORITY),
|
|
}
|
|
failure = {
|
|
**stable,
|
|
"receipt_sha256": operator._sha256_bytes(operator._canonical_json(stable)),
|
|
}
|
|
operator._validate_remote_failure(plan, failure)
|
|
failure["cleanup"] = {
|
|
"transient_containers_remaining": ["orphan"],
|
|
"staged_payload_removed": True,
|
|
}
|
|
with pytest.raises(operator.PromptIngressPlanError, match="cleanup"):
|
|
operator._validate_remote_failure(plan, failure)
|
|
|
|
|
|
def test_payload_claims_only_observation_consistency_and_no_write_surface() -> None:
|
|
source = Path(payload.__file__).read_text(encoding="utf-8")
|
|
assert '"database_observation_consistent": True' in source
|
|
assert '"database_write_path_exposed": False' in source
|
|
assert '"database_unchanged"' not in source
|
|
assert "RUNTIME_READ_ONLY_PGOPTIONS" in Path(
|
|
"hermes-agent/leoclean-bin/cloudsql_memory_tool.py"
|
|
).read_text(encoding="utf-8")
|