149 lines
4.9 KiB
Python
Executable File
149 lines
4.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Failure-injection checks for CI quality gates."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
GATE = ROOT / "ci_quality_gate.py"
|
|
HEALTH = ROOT / "validate_health_json.py"
|
|
|
|
|
|
def run(command):
|
|
return subprocess.run(command, capture_output=True, text=True)
|
|
|
|
|
|
def write_fixtures(
|
|
directory,
|
|
coverage_lines=(
|
|
"github.com/gochat/gochat/internal/ws/file.go:1.1,1.2 2 1\n"
|
|
"github.com/gochat/gochat/internal/ws/file.go:2.1,2.2 2 1\n"
|
|
),
|
|
test_lines=None,
|
|
duration="1",
|
|
):
|
|
(directory / "coverage.out").write_text("mode: atomic\n" + coverage_lines, encoding="utf-8")
|
|
events = test_lines or [
|
|
{"Action": "pass", "Package": "github.com/gochat/gochat/internal/ws", "Test": "TestRelay"}
|
|
]
|
|
(directory / "test.jsonl").write_text(
|
|
"\n".join(json.dumps(event) for event in events) + "\n", encoding="utf-8"
|
|
)
|
|
(directory / "duration").write_text(duration, encoding="utf-8")
|
|
|
|
|
|
def gate(directory, budget=5, exit_code=0):
|
|
return run(
|
|
[
|
|
sys.executable,
|
|
str(GATE),
|
|
"--coverage",
|
|
str(directory / "coverage.out"),
|
|
"--test-json",
|
|
str(directory / "test.jsonl"),
|
|
"--duration",
|
|
str(directory / "duration"),
|
|
"--report",
|
|
str(directory / "report.json"),
|
|
"--min-coverage",
|
|
"65",
|
|
"--max-duration",
|
|
str(budget),
|
|
"--test-exit-code",
|
|
str(exit_code),
|
|
"--critical-package",
|
|
"github.com/gochat/gochat/internal/ws",
|
|
]
|
|
)
|
|
|
|
|
|
def assert_auditable_failure(result, directory, name, expected):
|
|
report = json.loads((directory / "report.json").read_text(encoding="utf-8"))
|
|
assert result.returncode != 0, f"{name} must fail"
|
|
assert report["status"] == "failed"
|
|
assert expected in report["failures"]
|
|
assert expected in result.stdout
|
|
print(result.stdout.strip())
|
|
|
|
if evidence := os.getenv("GOCHAT_QUALITY_DIR"):
|
|
target = Path(evidence)
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
(target / f"{name}.json").write_text(
|
|
json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
|
|
|
|
def main():
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
directory = Path(temp)
|
|
write_fixtures(directory)
|
|
assert gate(directory).returncode == 0
|
|
|
|
write_fixtures(
|
|
directory,
|
|
"github.com/gochat/gochat/internal/ws/file.go:1.1,1.2 2 1\n"
|
|
"github.com/gochat/gochat/internal/ws/file.go:2.1,2.2 2 0\n",
|
|
)
|
|
assert gate(directory).returncode != 0, "coverage regression must fail"
|
|
|
|
write_fixtures(
|
|
directory,
|
|
test_lines=[
|
|
{"Action": "skip", "Package": "github.com/gochat/gochat/internal/ws", "Test": "TestRelay"}
|
|
],
|
|
)
|
|
assert gate(directory).returncode != 0, "all-skipped critical suite must fail"
|
|
|
|
write_fixtures(
|
|
directory,
|
|
"github.com/gochat/gochat/internal/wsevent/file.go:1.1,1.2 2 1\n",
|
|
[{"Action": "pass", "Package": "github.com/gochat/gochat/internal/wsevent", "Test": "TestBridge"}],
|
|
)
|
|
assert gate(directory).returncode != 0, "sibling package must not satisfy the ws gate"
|
|
|
|
write_fixtures(directory, duration="6")
|
|
assert gate(directory).returncode != 0, "duration budget must fail"
|
|
|
|
write_fixtures(
|
|
directory,
|
|
test_lines=[
|
|
{"Action": "fail", "Package": "github.com/gochat/gochat/internal/ws", "Test": "TestRelay"}
|
|
],
|
|
)
|
|
assert_auditable_failure(
|
|
gate(directory), directory, "action-fail", "test evidence contains 1 failed test events"
|
|
)
|
|
|
|
write_fixtures(directory)
|
|
assert_auditable_failure(
|
|
gate(directory, exit_code=2), directory, "nonzero-test-exit", "go test exited with 2"
|
|
)
|
|
|
|
health = directory / "health.json"
|
|
health.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "healthy",
|
|
"timestamp": "2026-08-22T00:00:00Z",
|
|
"uptime": "1s",
|
|
"checks": {"database": "healthy"},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
assert run([sys.executable, str(HEALTH), str(health)]).returncode == 0
|
|
health.write_text("{broken", encoding="utf-8")
|
|
assert run([sys.executable, str(HEALTH), str(health)]).returncode != 0, "invalid health JSON must fail"
|
|
health.write_text(json.dumps({"status": "unhealthy"}), encoding="utf-8")
|
|
assert run([sys.executable, str(HEALTH), str(health)]).returncode != 0, "unhealthy contract must fail"
|
|
|
|
print("quality gate failure-injection checks: ok")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|