219 lines
8 KiB
Python
219 lines
8 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from scripts import run_gcp_leoclean_nosend_oci_smoke as smoke
|
|
|
|
|
|
def _runtime_image(*, user: str | None = "", start_interval: int | None = 0) -> dict:
|
|
healthcheck = {
|
|
**smoke.EXPECTED_HEALTHCHECK,
|
|
"StartInterval": start_interval,
|
|
}
|
|
return {
|
|
"Config": {
|
|
"User": user,
|
|
"Entrypoint": list(smoke.EXPECTED_ENTRYPOINT),
|
|
"Cmd": list(smoke.EXPECTED_COMMAND),
|
|
"Healthcheck": healthcheck,
|
|
"StopSignal": smoke.EXPECTED_STOP_SIGNAL,
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("user", [None, ""])
|
|
@pytest.mark.parametrize("start_interval", [None, 0])
|
|
def test_runtime_configuration_accepts_portable_unset_values(user: str | None, start_interval: int | None) -> None:
|
|
receipt = smoke._validate_image_runtime_configuration(
|
|
_runtime_image(user=user, start_interval=start_interval)
|
|
)
|
|
|
|
assert receipt["bootstrap_user"] == "root"
|
|
assert receipt["entrypoint"] == smoke.EXPECTED_ENTRYPOINT
|
|
assert receipt["command"] == ["service"]
|
|
assert receipt["stop_signal"] == "SIGTERM"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "message"),
|
|
[
|
|
("User", "65532:65532", "root bootstrap user"),
|
|
("Entrypoint", ["/bin/sh"], "entrypoint drifted"),
|
|
("Cmd", ["probe"], "command drifted"),
|
|
("StopSignal", "SIGKILL", "stop signal drifted"),
|
|
],
|
|
)
|
|
def test_runtime_configuration_rejects_authoritative_field_drift(field: str, value: object, message: str) -> None:
|
|
image = _runtime_image()
|
|
image["Config"][field] = value
|
|
|
|
with pytest.raises(smoke.SmokeError, match=message):
|
|
smoke._validate_image_runtime_configuration(image)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[
|
|
("Test", ["CMD-SHELL", "true"]),
|
|
("Interval", 1),
|
|
("Timeout", 1),
|
|
("StartPeriod", 1),
|
|
("Retries", 1),
|
|
("StartInterval", 1),
|
|
("StartInterval", False),
|
|
("Unexpected", 1),
|
|
],
|
|
)
|
|
def test_runtime_configuration_rejects_healthcheck_drift(field: str, value: object) -> None:
|
|
image = _runtime_image()
|
|
image["Config"]["Healthcheck"] = copy.deepcopy(image["Config"]["Healthcheck"])
|
|
image["Config"]["Healthcheck"][field] = value
|
|
|
|
with pytest.raises(smoke.SmokeError, match="healthcheck"):
|
|
smoke._validate_image_runtime_configuration(image)
|
|
|
|
|
|
def test_build_command_is_one_platform_pinned_local_build(tmp_path: Path) -> None:
|
|
command = smoke._build_command(
|
|
context=tmp_path,
|
|
tag="livingip-leoclean-nosend-smoke:test",
|
|
smoke_label="livingip.leoclean-nosend-smoke=test",
|
|
build_args={"TELEO_REVISION": "1" * 40, "INPUT_SHA256": "2" * 64},
|
|
)
|
|
|
|
assert command[:6] == ["docker", "build", "--platform", "linux/amd64", "--pull=false", "--tag"]
|
|
assert command.count("build") == 1
|
|
assert "--push" not in command
|
|
assert "--load" not in command
|
|
assert command[-1] == str(tmp_path)
|
|
assert command.index("INPUT_SHA256=" + "2" * 64) < command.index("TELEO_REVISION=" + "1" * 40)
|
|
|
|
|
|
def test_run_command_has_no_port_secret_or_host_mount_surface() -> None:
|
|
command = smoke._run_command(
|
|
tag="livingip-leoclean-nosend-smoke:test",
|
|
name="livingip-leoclean-nosend-smoke-test",
|
|
smoke_label="livingip.leoclean-nosend-smoke=test",
|
|
)
|
|
encoded = " ".join(command).casefold()
|
|
|
|
for required in (
|
|
"--read-only",
|
|
"--cap-drop=all",
|
|
"--security-opt=no-new-privileges:true",
|
|
"--network=bridge",
|
|
"--pids-limit=512",
|
|
"--pull=never",
|
|
):
|
|
assert required in encoded
|
|
assert "--publish" not in encoded
|
|
assert " -p " not in encoded
|
|
assert "--env" not in encoded
|
|
assert "--volume" not in encoded
|
|
assert "docker.sock" not in encoded
|
|
assert "telegram" not in encoded
|
|
|
|
|
|
def test_unsupported_send_requires_exact_redacted_fail_closed_result(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
result = subprocess.CompletedProcess(
|
|
args=["docker"],
|
|
returncode=65,
|
|
stdout="",
|
|
stderr=json.dumps({"status": "fail", "error": "container_contract_failed"}) + "\n",
|
|
)
|
|
monkeypatch.setattr(smoke, "_run", lambda *_args, **_kwargs: result)
|
|
|
|
receipt = smoke._verify_unsupported_send("fixture")
|
|
|
|
assert receipt == {
|
|
"exit_code": 65,
|
|
"stdout_empty": True,
|
|
"redacted_error": {"status": "fail", "error": "container_contract_failed"},
|
|
}
|
|
|
|
|
|
def test_claim_ceiling_does_not_overstate_entrypoint_dispatch_probe() -> None:
|
|
assert "does not prove a real Hermes tool-call denial" in smoke.SUCCESS_CLAIM_CEILING
|
|
assert "general network-egress denial" in smoke.SUCCESS_CLAIM_CEILING
|
|
assert "Cloud SQL permissions" in smoke.SUCCESS_CLAIM_CEILING
|
|
assert "production readiness" in smoke.SUCCESS_CLAIM_CEILING
|
|
assert "base images and build cache may remain" in smoke.SUCCESS_CLAIM_CEILING
|
|
assert "makes no affirmative image-build" in smoke.FAILURE_CLAIM_CEILING
|
|
|
|
|
|
def test_health_wait_rejects_container_running_a_different_image(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
inspected = [{"Image": "sha256:wrong", "State": {"Status": "running", "Health": {"Status": "healthy"}}}]
|
|
result = subprocess.CompletedProcess(args=["docker"], returncode=0, stdout=json.dumps(inspected), stderr="")
|
|
monkeypatch.setattr(smoke, "_run", lambda *_args, **_kwargs: result)
|
|
|
|
with pytest.raises(smoke.SmokeError, match="image ID differs"):
|
|
smoke._wait_healthy("fixture", expected_image_id="sha256:expected", timeout=0.1)
|
|
|
|
|
|
def test_cleanup_fails_when_docker_inventory_is_indeterminate(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls = 0
|
|
|
|
def fake_run(*_args, **_kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls <= 2:
|
|
return subprocess.CompletedProcess(args=["docker"], returncode=1, stdout="", stderr="not found")
|
|
raise smoke.SmokeError("Docker inventory unavailable")
|
|
|
|
monkeypatch.setattr(smoke, "_run", fake_run)
|
|
|
|
with pytest.raises(smoke.SmokeError, match="inventory unavailable"):
|
|
smoke._cleanup(
|
|
name="fixture",
|
|
tag="livingip-leoclean-nosend-smoke:fixture",
|
|
image_id="sha256:" + "1" * 64,
|
|
smoke_label="livingip.leoclean-nosend-smoke=fixture",
|
|
)
|
|
|
|
|
|
def test_failed_receipt_never_carries_the_success_claim(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(smoke.shutil, "which", lambda _name: None)
|
|
monkeypatch.setattr(smoke, "_cleanup", lambda **_kwargs: {"status": "pass"})
|
|
output = tmp_path / "receipt.json"
|
|
|
|
report, exit_code = smoke.run_smoke(
|
|
repo_root=tmp_path,
|
|
artifact=tmp_path / "artifact",
|
|
runtime_receipt=tmp_path / "runtime.json",
|
|
output=output,
|
|
)
|
|
|
|
assert exit_code == 65
|
|
assert report["status"] == "fail"
|
|
assert report["claim_ceiling"] == smoke.FAILURE_CLAIM_CEILING
|
|
assert smoke.SUCCESS_CLAIM_CEILING not in output.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_runtime_canary_can_retain_only_a_successfully_verified_artifact() -> None:
|
|
source = (Path(__file__).resolve().parents[1] / "scripts" / "run_leoclean_nosend_runtime_canary.sh").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
verification = source.index('receipt.get("status") != "pass"')
|
|
retention = source.index('if [[ -n "$artifact_output" ]]')
|
|
assert verification < retention
|
|
assert "retained artifact output already exists" in source
|
|
assert 'cp -pR "$artifact" "$artifact_output_partial"' in source
|
|
assert 'mv "$artifact_output_partial" "$artifact_output"' in source
|
|
|
|
|
|
def test_ci_preloads_the_exact_target_architecture_and_checks_inventory() -> None:
|
|
workflow = (Path(__file__).resolve().parents[1] / ".github" / "workflows" / "ci.yml").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
assert workflow.count("docker pull --platform=linux/amd64") == 3
|
|
assert "container_inventory=\"$(docker ps" in workflow
|
|
assert "image_inventory=\"$(docker images" in workflow
|
|
assert "remaining_containers=\"$(docker ps" in workflow
|
|
assert "remaining_images=\"$(docker images" in workflow
|