Spaces:
Sleeping
Sleeping
File size: 6,434 Bytes
7098eeb 9fa3c4f 7098eeb b0593e2 7098eeb 9fa3c4f 7098eeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | import importlib.util
from pathlib import Path
SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "docker_smoke.py"
SPEC = importlib.util.spec_from_file_location("docker_smoke", SCRIPT_PATH)
docker_smoke = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
SPEC.loader.exec_module(docker_smoke)
def write_dockerignore(root: Path) -> None:
(root / ".dockerignore").write_text(
"\n".join(
[
".git/",
".env",
".env.*",
"!.env.example",
"!.env.final.example",
".env.final.local",
".venv/",
"var/",
"output/",
]
)
+ "\n",
encoding="utf-8",
)
def fake_runner(args, cwd, timeout, env=None):
if args[:2] == ["docker", "info"]:
return {"ok": True, "returncode": 0, "stdout_tail": '"29.3.0"', "stderr_tail": ""}
if args[:2] == ["docker", "build"]:
return {"ok": True, "returncode": 0, "stdout_tail": "Successfully tagged", "stderr_tail": ""}
if args[:2] == ["docker", "run"]:
return {"ok": True, "returncode": 0, "stdout_tail": "container-id", "stderr_tail": ""}
if args[:2] == ["docker", "logs"]:
return {"ok": True, "returncode": 0, "stdout_tail": "Uvicorn running", "stderr_tail": ""}
if args[:3] == ["docker", "rm", "-f"]:
return {"ok": True, "returncode": 0, "stdout_tail": "removed", "stderr_tail": ""}
if args[-1].startswith("http://127.0.0.1:"):
assert env and env["PYTHONPATH"] == "src"
return {"ok": True, "returncode": 0, "stdout_tail": '{"ok": true}', "stderr_tail": ""}
raise AssertionError(f"unexpected command: {args}")
def recording_runner(calls, overrides=None):
overrides = overrides or {}
def runner(args, cwd, timeout, env=None):
calls.append(args)
command_id = "_".join(args[:2]) if args[:2] != ["docker", "rm"] else "docker_rm"
if command_id in overrides:
return overrides[command_id]
return fake_runner(args, cwd, timeout, env)
return runner
def healthy_fetcher(url, timeout):
return {
"ok": True,
"status": 200,
"json": {
"ready": True,
"storage_backend": "local",
"generation_backend": "mock",
"b2_configured": False,
"genblaze_configured": False,
},
"error": None,
}
def test_fetch_health_treats_connection_reset_as_transient(monkeypatch):
def broken_urlopen(request, timeout):
raise ConnectionResetError("reset by peer")
monkeypatch.setattr(docker_smoke, "urlopen", broken_urlopen)
result = docker_smoke.fetch_health("http://127.0.0.1:18088/api/health", 1)
assert result["ok"] is False
assert "reset by peer" in result["error"]
def test_docker_smoke_report_passes_with_fake_runner(tmp_path):
write_dockerignore(tmp_path)
report = docker_smoke.build_report(
root=tmp_path,
command_runner=fake_runner,
health_fetcher=healthy_fetcher,
sleep_fn=lambda seconds: None,
)
assert report["schema"] == "proofframe.docker_smoke.v1"
assert report["ok"] is True
assert report["mode"] == "docker_smoke_ready"
assert report["dockerignore"]["missing_required_patterns"] == []
assert report["dockerignore"]["missing_allow_patterns"] == []
assert {check["id"] for check in report["checks"]} >= {
"dockerignore_secret_exclusions",
"docker_build",
"docker_health",
"api_smoke",
}
assert "does not read .env.final.local" in report["secret_policy"]
def test_docker_smoke_cleans_up_after_health_exception(tmp_path):
write_dockerignore(tmp_path)
calls = []
def broken_fetcher(url, timeout):
raise RuntimeError("health exploded")
report = docker_smoke.build_report(
root=tmp_path,
command_runner=recording_runner(calls),
health_fetcher=broken_fetcher,
sleep_fn=lambda seconds: None,
)
failed = {check["id"] for check in report["checks"] if not check["ok"]}
assert report["ok"] is False
assert "docker_health" in failed
assert ["docker", "rm", "-f", "proofframe-submission-smoke"] in calls
cleanup = next(command for command in report["commands"] if command["id"] == "docker_cleanup")
assert cleanup["ok"] is True
logs = next(command for command in report["commands"] if command["id"] == "docker_logs")
assert "health exploded" in logs["stderr_tail"]
def test_docker_smoke_cleans_up_when_api_smoke_fails(tmp_path):
write_dockerignore(tmp_path)
calls = []
report = docker_smoke.build_report(
root=tmp_path,
command_runner=recording_runner(
calls,
overrides={
f"{docker_smoke.sys.executable}_scripts/api_smoke.py": {
"ok": False,
"returncode": 1,
"stdout_tail": "",
"stderr_tail": "api failed",
}
},
),
health_fetcher=healthy_fetcher,
sleep_fn=lambda seconds: None,
)
failed = {check["id"] for check in report["checks"] if not check["ok"]}
assert report["ok"] is False
assert "api_smoke" in failed
assert ["docker", "rm", "-f", "proofframe-submission-smoke"] in calls
def test_docker_smoke_fails_closed_without_dockerignore(tmp_path):
report = docker_smoke.build_report(
root=tmp_path,
command_runner=fake_runner,
health_fetcher=healthy_fetcher,
sleep_fn=lambda seconds: None,
)
failed = {check["id"] for check in report["checks"] if not check["ok"]}
assert report["ok"] is False
assert "dockerignore_secret_exclusions" in failed
def test_docker_smoke_writes_json_and_markdown(tmp_path):
write_dockerignore(tmp_path)
report = docker_smoke.build_report(
root=tmp_path,
command_runner=fake_runner,
health_fetcher=healthy_fetcher,
sleep_fn=lambda seconds: None,
)
json_path = tmp_path / "docker-smoke.json"
markdown_path = tmp_path / "docker-smoke.md"
docker_smoke.write_outputs(report, json_path, markdown_path)
assert json_path.read_text(encoding="utf-8").startswith("{")
assert "# ProofFrame Docker Smoke Report" in markdown_path.read_text(encoding="utf-8")
|