273 lines
9.2 KiB
Python
273 lines
9.2 KiB
Python
import json
|
|
import subprocess
|
|
import sys
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
from scripts.ops import report_source_deploy_runtime_truth as report
|
|
|
|
|
|
LOCAL_HEAD = "a" * 40
|
|
OTHER_HEAD = "b" * 40
|
|
|
|
|
|
def _write_source_root(tmp_path: Path) -> Path:
|
|
(tmp_path / "config.py").write_text('SYSTEM_VERSION = "V10.725"\n', encoding="utf-8")
|
|
(tmp_path / "proof.txt").write_text("deployed proof\n", encoding="utf-8")
|
|
return tmp_path
|
|
|
|
|
|
def _runner(
|
|
*,
|
|
origin_main: str = LOCAL_HEAD,
|
|
origin_dev: str = LOCAL_HEAD,
|
|
gitea_main: str = LOCAL_HEAD,
|
|
gitea_dev: str = LOCAL_HEAD,
|
|
head_config_version: str = "V10.725",
|
|
tracked_file_status: str = "",
|
|
container_state: dict | None = None,
|
|
):
|
|
def run(args: list[str], cwd: Path) -> str:
|
|
if args == ["git", "rev-parse", "HEAD"]:
|
|
return LOCAL_HEAD
|
|
if args == ["git", "rev-parse", "--abbrev-ref", "HEAD"]:
|
|
return "codex/prod-version-truth-guard"
|
|
if args == ["git", "show", "HEAD:config.py"]:
|
|
return f'SYSTEM_VERSION = "{head_config_version}"\n'
|
|
if args == ["git", "remote", "get-url", "origin"]:
|
|
return "https://gitea.wooo.work/wooo/ewoooc.git"
|
|
if args[:3] == ["git", "ls-remote", "origin"]:
|
|
return "\n".join(
|
|
[
|
|
f"{origin_main}\trefs/heads/main",
|
|
f"{origin_dev}\trefs/heads/dev",
|
|
]
|
|
)
|
|
if args[:3] == ["git", "ls-remote", report.DEFAULT_GITEA_REMOTE]:
|
|
return "\n".join(
|
|
[
|
|
f"{gitea_main}\trefs/heads/main",
|
|
f"{gitea_dev}\trefs/heads/dev",
|
|
]
|
|
)
|
|
if args[:3] == ["git", "status", "--porcelain"]:
|
|
return tracked_file_status
|
|
if args[:4] == ["docker", "inspect", "--format", "{{json .State}}"]:
|
|
state = container_state or {
|
|
"Status": "running",
|
|
"Running": True,
|
|
"Health": {"Status": "healthy"},
|
|
}
|
|
return json.dumps(state)
|
|
raise AssertionError(f"unexpected command: {args}")
|
|
|
|
return run
|
|
|
|
|
|
def _health(version: str = "V10.725", status: str = "healthy"):
|
|
def fetch(url: str, timeout: float) -> dict:
|
|
return {"status": status, "database": "postgresql", "version": version}
|
|
|
|
return fetch
|
|
|
|
|
|
def test_report_passes_when_source_deploy_runtime_truth_aligns(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "proof.txt"),
|
|
container_name="momo-pro-system",
|
|
runner=_runner(),
|
|
health_fetcher=_health(),
|
|
)
|
|
|
|
assert payload["result"] == "PASS"
|
|
assert payload["summary"]["source_control_ok"] is True
|
|
assert payload["summary"]["tracked_files_committed"] is True
|
|
assert payload["summary"]["deployment_hash_readback_ok"] is True
|
|
assert payload["summary"]["production_health_ok"] is True
|
|
assert payload["summary"]["truth_layers_separated"] is True
|
|
assert payload["runtime"]["container"]["health_status"] == "healthy"
|
|
assert payload["safety_gates"]["github_allowed_actions"] == 0
|
|
assert payload["safety_gates"]["database_write_performed"] is False
|
|
|
|
|
|
def test_report_blocks_when_gitea_main_differs_from_local_head(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "proof.txt"),
|
|
runner=_runner(gitea_main=OTHER_HEAD),
|
|
health_fetcher=_health(),
|
|
)
|
|
|
|
assert payload["result"] == "BLOCKED"
|
|
assert payload["summary"]["source_control_ok"] is False
|
|
assert any("Gitea SSH main/dev are not aligned" in error for error in payload["errors"])
|
|
|
|
|
|
def test_report_blocks_when_tracked_deployment_file_is_not_committed(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "proof.txt"),
|
|
runner=_runner(tracked_file_status=" M proof.txt"),
|
|
health_fetcher=_health(),
|
|
)
|
|
|
|
assert payload["result"] == "BLOCKED"
|
|
assert payload["summary"]["tracked_files_committed"] is False
|
|
assert any("uncommitted source-control changes" in error for error in payload["errors"])
|
|
|
|
|
|
def test_report_blocks_when_production_version_differs_from_head(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "proof.txt"),
|
|
runner=_runner(),
|
|
health_fetcher=_health(version="V10.724"),
|
|
)
|
|
|
|
assert payload["result"] == "BLOCKED"
|
|
assert payload["summary"]["production_version_matches_head"] is False
|
|
assert payload["summary"]["version_bump_detected"] is True
|
|
assert any("production /health version does not match HEAD config.py" in error for error in payload["errors"])
|
|
|
|
|
|
def test_missing_deployment_file_blocks_only_the_deployment_layer(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "missing.txt"),
|
|
runner=_runner(),
|
|
health_fetcher=_health(),
|
|
)
|
|
|
|
assert payload["result"] == "BLOCKED"
|
|
assert payload["summary"]["source_control_ok"] is True
|
|
assert payload["summary"]["deployment_hash_readback_ok"] is False
|
|
assert any("tracked deployment files" in error for error in payload["errors"])
|
|
|
|
|
|
def test_text_output_exposes_source_deployment_and_runtime_layers(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "proof.txt"),
|
|
runner=_runner(),
|
|
health_fetcher=_health(),
|
|
)
|
|
|
|
text = report.format_text(payload)
|
|
|
|
assert "origin_main:" in text
|
|
assert "gitea_main:" in text
|
|
assert "tracked_files_committed: true" in text
|
|
assert "production_health: healthy postgresql V10.725" in text
|
|
assert "deployment_files_hashed: 2" in text
|
|
assert "truth_layers_separated: true" in text
|
|
|
|
|
|
def test_report_can_use_source_overrides_for_no_git_deployment_tree(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "proof.txt"),
|
|
source_override={
|
|
"branch": "main",
|
|
"head": LOCAL_HEAD,
|
|
"origin_main": LOCAL_HEAD,
|
|
"origin_dev": LOCAL_HEAD,
|
|
"gitea_main": LOCAL_HEAD,
|
|
"gitea_dev": LOCAL_HEAD,
|
|
"head_config_version": "V10.725",
|
|
"tracked_files_committed": True,
|
|
},
|
|
runner=lambda args, cwd: (_ for _ in ()).throw(AssertionError("git should not run")),
|
|
health_fetcher=_health(),
|
|
)
|
|
|
|
assert payload["result"] == "PASS"
|
|
assert payload["source_control"]["source_mode"] == "override_for_no_git_deployment_tree"
|
|
assert payload["summary"]["source_control_ok"] is True
|
|
assert payload["summary"]["tracked_files_committed"] is True
|
|
|
|
|
|
def test_source_override_still_requires_committed_file_confirmation(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
|
|
payload = report.build_report(
|
|
root=source_root,
|
|
tracked_files=("config.py", "proof.txt"),
|
|
source_override={
|
|
"branch": "main",
|
|
"head": LOCAL_HEAD,
|
|
"origin_main": LOCAL_HEAD,
|
|
"origin_dev": LOCAL_HEAD,
|
|
"gitea_main": LOCAL_HEAD,
|
|
"gitea_dev": LOCAL_HEAD,
|
|
"head_config_version": "V10.725",
|
|
"tracked_files_committed": False,
|
|
},
|
|
runner=lambda args, cwd: (_ for _ in ()).throw(AssertionError("git should not run")),
|
|
health_fetcher=_health(),
|
|
)
|
|
|
|
assert payload["result"] == "BLOCKED"
|
|
assert payload["summary"]["tracked_files_committed"] is False
|
|
assert any("uncommitted source-control changes" in error for error in payload["errors"])
|
|
|
|
|
|
def test_script_runs_by_path_when_called_from_outside_repo(tmp_path):
|
|
source_root = _write_source_root(tmp_path)
|
|
health_payload = urllib.parse.quote(
|
|
json.dumps({"status": "healthy", "database": "postgresql", "version": "V10.725"})
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(Path(report.__file__).resolve()),
|
|
"--source-root",
|
|
str(source_root),
|
|
"--health-url",
|
|
f"data:application/json,{health_payload}",
|
|
"--tracked-file",
|
|
"config.py",
|
|
"--tracked-file",
|
|
"proof.txt",
|
|
"--source-head",
|
|
LOCAL_HEAD,
|
|
"--source-branch",
|
|
"main",
|
|
"--origin-main",
|
|
LOCAL_HEAD,
|
|
"--origin-dev",
|
|
LOCAL_HEAD,
|
|
"--gitea-main",
|
|
LOCAL_HEAD,
|
|
"--gitea-dev",
|
|
LOCAL_HEAD,
|
|
"--head-config-version",
|
|
"V10.725",
|
|
"--tracked-files-committed",
|
|
"--json",
|
|
],
|
|
cwd="/tmp",
|
|
check=True,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
|
|
payload = json.loads(result.stdout)
|
|
assert payload["result"] == "PASS"
|
|
assert payload["summary"]["truth_layers_separated"] is True
|