scripts/test-microservice-runtime.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Live OCI -> MicroVM -> HTTP service test. Run on the Linux Firecrab host.
Requires a running API, helper, /dev/kvm and a fixture image from an account
connected in Pipe Auth. The image must serve HTTP on port 8080, return the
GREETING environment value at /health and a non-2xx response at /missing.
"""
import argparse
import json
import subprocess
import time
import uuid
from urllib.error import HTTPError
from urllib.request import Request, urlopen
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--api", default="http://127.0.0.1:5523")
parser.add_argument("--image", required=True, help="Fixture image in a Pipe Auth-connected GHCR namespace or ECR registry")
parser.add_argument("--host-port", type=int, default=30170)
parser.add_argument("--restart-unit", help="Optional QA API unit to restart after successful deployment")
parser.add_argument("--keep-running", action="store_true", help="Keep the successful service for manual inspection")
parser.add_argument("--with-secrets", action="store_true", help="Verify encrypted Secret injection, redaction, rotation isolation and VM restart")
args = parser.parse_args()
owned = []
kept = None
secret = None
greeting = "FIRECRAB_LIVE_SERVICE_OK"
if args.with_secrets:
greeting += "_" + uuid.uuid4().hex
def request(method, path, body=None):
payload = None if body is None else json.dumps(body).encode()
req = Request(args.api.rstrip("/") + path, data=payload, method=method)
if payload is not None:
req.add_header("Content-Type", "application/json")
try:
with urlopen(req, timeout=15) as response:
content = response.read()
return json.loads(content) if content else None
except HTTPError as error:
raise RuntimeError(f"{method} {path}: HTTP {error.code}: {error.read().decode()}") from error
def wait(service_id, terminal, timeout=600):
deadline = time.monotonic() + timeout
previous = None
while time.monotonic() < deadline:
status = request("GET", f"/api/micro-services/{service_id}/deployment")
if status["phase"] != previous:
print(f"{service_id}: {status['phase']}", flush=True)
previous = status["phase"]
if status["phase"] in terminal:
return status
if status["phase"] in ("failed", "degraded"):
raise AssertionError(json.dumps(status, indent=2))
time.sleep(1)
raise TimeoutError(f"Service {service_id} did not reach {terminal}")
def check_http(status):
for instance in status["instances"]:
url = f"http://127.0.0.1:{instance['hostPort']}/health"
with urlopen(url, timeout=5) as response:
assert response.read().decode().strip() == greeting
def check_redaction(service_id, status):
for payload in [status, request("GET", f"/api/micro-services/{service_id}"), request("GET", "/api/vms")]:
assert greeting not in json.dumps(payload), "Secret leaked through a public response"
def wait_vm(vm_id, expected):
deadline = time.monotonic() + 180
while time.monotonic() < deadline:
vm = request("GET", f"/api/vms/{vm_id}")
if vm["state"] == expected:
return
assert vm["state"] not in ("failed", "crashed"), "replica lifecycle failed"
time.sleep(1)
raise TimeoutError("replica did not reach " + expected)
def check_removed(service_id):
prefix = f"ms-{service_id.replace('-', '')}-"
assert not any(vm["name"].startswith(prefix) for vm in request("GET", "/api/vms"))
try:
if args.with_secrets:
secret = request("POST", "/api/secrets", {"name": "qa-service-" + uuid.uuid4().hex, "value": greeting})
for suffix, health, replicas in [("healthy", "/health", 2), ("unhealthy", "/missing", 1)]:
spec = {
"name": f"live-{suffix}-{int(time.time())}", "image": args.image,
"port": 8080, "ram": 128, "cpu": 1, "replicas": replicas,
"healthPath": health, "startupTimeout": 30 if suffix == "healthy" else 3,
"drainTimeout": 1,
"env": {} if secret else {"GREETING": greeting},
"secrets": {"GREETING": secret["id"]} if secret else {},
}
service = request("POST", "/api/micro-services", spec)
service_id = service["id"]
owned.append(service_id)
request("POST", f"/api/micro-services/{service_id}/deployment", {"hostPort": args.host_port + (10 if suffix == "unhealthy" else 0)})
status = wait(service_id, {"running", "failed"})
if suffix == "unhealthy":
assert status["phase"] == "failed", status
assert status["instances"] == [], status
assert "health" in status["error"].lower(), status
check_removed(service_id)
print("PASS: failed app health leaves no service VMs", flush=True)
continue
assert status["phase"] == "running", status
assert len(status["instances"]) == replicas, status
assert status["digest"].startswith("sha256:"), status
for instance in status["instances"]:
assert instance["state"] == "running" and instance["healthy"], instance
check_http(status)
print("PASS: two real MicroVMs answer HTTP with injected environment", flush=True)
if secret:
check_redaction(service_id, status)
request("PUT", "/api/secrets/" + secret["id"], {
"name": secret["name"], "value": "rotated-" + uuid.uuid4().hex, "version": secret["version"],
})
vm_id = status["instances"][0]["vmId"]
request("POST", f"/api/vms/{vm_id}/stop")
wait_vm(vm_id, "stopped")
request("POST", f"/api/vms/{vm_id}/start")
wait_vm(vm_id, "running")
deadline = time.monotonic() + 60
while True:
try:
check_http(status)
break
except OSError:
if time.monotonic() >= deadline:
raise
time.sleep(1)
check_redaction(service_id, request("GET", f"/api/micro-services/{service_id}/deployment"))
print("PASS: Secret values stay private and VM restart retains the deployment snapshot after rotation", flush=True)
if args.restart_unit:
with subprocess.Popen(["systemctl", "restart", args.restart_unit]) as restart:
while True:
check_http(status)
if restart.poll() is not None:
assert restart.returncode == 0
break
time.sleep(0.05)
deadline = time.monotonic() + 45
while True:
try:
status = request("GET", f"/api/micro-services/{service_id}/deployment")
assert status["phase"] == "running", status
assert all(instance["healthy"] for instance in status["instances"]), status
break
except (OSError, RuntimeError):
if time.monotonic() >= deadline:
raise
time.sleep(1)
print("PASS: API restart re-adopts the service and preserves HTTP health", flush=True)
kept = service_id
if not args.keep_running:
request("POST", f"/api/micro-services/{service_id}/stop")
stopped = wait(service_id, {"stopped"})
assert stopped["instances"] == []
check_removed(service_id)
print("PASS: Stop removes owned VMs and keeps service configuration", flush=True)
kept = None
finally:
for service_id in owned:
if args.keep_running and service_id == kept:
print(f"Kept live service: {service_id}", flush=True)
continue
status = request("GET", f"/api/micro-services/{service_id}/deployment")
if status and status["instances"]:
if status["phase"] in ("running", "degraded", "failed"):
request("POST", f"/api/micro-services/{service_id}/stop")
wait(service_id, {"stopped"})
else:
raise RuntimeError(f"Deployment still active; clean up service {service_id} after it finishes")
request("DELETE", f"/api/micro-services/{service_id}")
if secret and not (args.keep_running and kept):
request("DELETE", "/api/secrets/" + secret["id"])
if __name__ == "__main__":
main()