Homepage (TDD slice 1): the status checker core
Full TDD (Q36): red test first, then the minimal implementation. - tests/homepage_checker.bats: seam = homepage-status.py <services.json> <out.json>. Asserts an open port is up, a closed port is down, a wip service is wip (not probed), and the output carries generated_at. - scripts/homepage-status.py: reads the config, probes each backend (tcp, 5 s timeout), skips wip, writes status.json atomically (temp + rename). Remaining slices: http probe method, services.json schema, the pages + theme + toggle, the Caddyfile (validate + two hashes), the git-driven deploy.
This commit is contained in:
Executable
+69
@@ -0,0 +1,69 @@
|
||||
#!/usr/bin/env python3
|
||||
"""homepage-status.py — the homepage status checker.
|
||||
|
||||
Reads a service config (a JSON array of service entries), probes each backend,
|
||||
and writes a status.json atomically. A wip service is not probed.
|
||||
|
||||
Usage: homepage-status.py <services.json> <out.json>
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import tempfile
|
||||
from datetime import datetime, timezone
|
||||
|
||||
TIMEOUT = 5 # seconds, per probe (Q11 / Q40)
|
||||
|
||||
|
||||
def probe(host, port, method="tcp"):
|
||||
"""Return 'up' if the backend answers, else 'down'."""
|
||||
try:
|
||||
with socket.create_connection((host, port), timeout=TIMEOUT):
|
||||
return "up"
|
||||
except Exception:
|
||||
return "down"
|
||||
|
||||
|
||||
def main(argv):
|
||||
if len(argv) != 3:
|
||||
sys.stderr.write("usage: homepage-status.py <services.json> <out.json>\n")
|
||||
return 2
|
||||
services_path, out_path = argv[1], argv[2]
|
||||
with open(services_path) as f:
|
||||
services = json.load(f)
|
||||
|
||||
results = []
|
||||
for svc in services:
|
||||
if svc.get("wip"):
|
||||
state = "wip"
|
||||
else:
|
||||
host, _, port = svc["probe"].rpartition(":")
|
||||
state = probe(host, int(port), svc.get("method", "tcp"))
|
||||
results.append({
|
||||
"name": svc.get("name"),
|
||||
"category": svc.get("category", "user"),
|
||||
"state": state,
|
||||
})
|
||||
|
||||
doc = {
|
||||
"generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
||||
"services": results,
|
||||
}
|
||||
|
||||
# Atomic write: a temp file in the same directory, then rename over the target.
|
||||
out_dir = os.path.dirname(os.path.abspath(out_path))
|
||||
fd, tmp = tempfile.mkstemp(dir=out_dir, prefix=".status.", suffix=".tmp")
|
||||
try:
|
||||
with os.fdopen(fd, "w") as f:
|
||||
json.dump(doc, f)
|
||||
os.replace(tmp, out_path)
|
||||
except Exception:
|
||||
if os.path.exists(tmp):
|
||||
os.unlink(tmp)
|
||||
raise
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv))
|
||||
Reference in New Issue
Block a user