import json import os from pathlib import Path import socket import subprocess import tempfile import urllib.error import urllib.request import uuid import sys class StackCheck: def __init__(self, root, workspace): self.root = Path(root).resolve() self.workspace = Path(workspace).resolve() self.project = "post76-ci-" + uuid.uuid4().hex self.env_file = self.workspace / "stack.env" self.environment = { "PATH": os.environ["PATH"], "DOCKER_HOST": "unix:///var/run/docker.sock", "DOCKER_CONFIG": str(self.workspace / "docker-config"), "COMPOSE_DISABLE_ENV_FILE": "1", } def command(self, arguments, capture=True): return subprocess.run(arguments, cwd=self.root, env=self.environment, check=True, capture_output=capture, text=True) def compose(self, *arguments): return self.command(["docker", "compose", "--env-file", str(self.env_file), "--project-name", self.project, "-f", str(self.root / "compose.yaml"), *arguments]) def request(self, port, path, domain="web.post76.localhost"): request = urllib.request.Request(f"http://127.0.0.1:{port}{path}", headers={"Host": domain}) opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) with opener.open(request, timeout=15) as response: return response.status, response.read(), response.headers def run(self): secret_dir = self.workspace / "fixtures" self.command(["php", str(self.root / "docker/create-secrets.php"), str(secret_dir)]) with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) port = listener.getsockname()[1] self.env_file.write_text(f"POST76_SECRETS_DIR={secret_dir}\nPOST76_IMAGE_TAG=ci\nPOST76_HTTP_PORT={port}\n" f"POST76_APP_URL=http://web.post76.localhost:{port}\nPOST76_REVERB_PORT={port}\nPOST76_METRICS_ENABLED=true\n") self.env_file.chmod(0o600) model = json.loads(self.compose("config", "--format", "json").stdout) assert len(model["services"]) == 9 assert model["networks"]["backend"]["internal"] assert all(not value.get("ports") for name, value in model["services"].items() if name != "nginx") result = {"result": "failed", "services": 9} try: self.compose("up", "--detach", "--no-build", "--wait", "--wait-timeout", "300") status, body, _ = self.request(port, "/up") assert status == 200 status, body, _ = self.request(port, "/v1/app/config/get", "api.post76.localhost") assert status == 200 and json.loads(body)["status"] == "success" for path in ["/.env", "/storage/private.txt"]: try: self.request(port, path) raise AssertionError("Private paths must not be served") except urllib.error.HTTPError as error: assert error.code == 404 self.compose("exec", "-T", "app", "php", "-r", 'if (file_exists(".env") || is_link("public/storage")) { exit(1); } file_put_contents("storage/app/private/ci-persistence", "ci-only");') self.compose("up", "--detach", "--no-build", "--no-deps", "--wait", "--force-recreate", "app") self.compose("exec", "-T", "app", "php", "-r", 'if (file_get_contents("storage/app/private/ci-persistence") !== "ci-only") { exit(1); }') self.check_backups() self.check_monitoring() result = {"result": "passed", "services": 9, "checks": ["health", "api", "private-paths", "uploads-recreate", "encrypted-backup", "isolated-mysql-restore", "aggregate-metrics", "alert-rules", "metrics-exporter-config"]} print("Synthetic container health, API, private paths and upload persistence passed") except Exception: try: statuses = json.loads(self.compose("ps", "--all", "--format", "json").stdout) result["status"] = [{key: value.get(key) for key in ["Service", "State", "Health", "ExitCode"]} for value in statuses] if isinstance(statuses, list) else [] except Exception: result["status"] = [] raise finally: results = self.root / "storage/app/private/ci-check/container-results" results.mkdir(parents=True, exist_ok=True) (results / "summary.json").write_text(json.dumps(result, indent=2)) self.compose("down", "--volumes", "--remove-orphans", "--timeout", "30") def check_backups(self): repository = self.workspace / "repository" work = self.workspace / "operations" restore = self.workspace / "recovery" arguments = [sys.executable, "ops/backup.py", "--project", self.project, "--env-file", str(self.env_file), "--work-dir", str(work), "--local-repository", str(repository)] self.command([*arguments, "init-repository"]) completed = self.command([*arguments, "backup", "--quiesce"]) identity = completed.stdout.strip().split()[-1] self.command([*arguments, "restore-drill", "--snapshot", identity, "--destination", str(restore)]) outcome = json.loads((restore / "database-drill-result.json").read_text()) assert outcome["database_imported"] and outcome["tables"] > 0 and not outcome["production_database_modified"] manifest = json.loads((restore / "data/capture/manifest.json").read_text()) assert manifest["project"] == self.project assert not (restore / "data/keys/backup_password").exists() assert not (restore / "data/keys/backup_s3_credentials").exists() assert (restore / "data/redis/dump.rdb").stat().st_size > 0 def check_monitoring(self): work = self.workspace / "operations" self.command([sys.executable, "ops/metrics.py", "--project", self.project, "--env-file", str(self.env_file), "--work-dir", str(work)]) values = (work / "metrics/metrics.prom").read_text() assert 'post76_collector_success{project="' + self.project + '"} 1' in values assert 'volume="mysql"' in values and 'queue="mail"' in values mounts = ["--mount", "type=bind,source=" + str(self.root / "docker/monitoring/post76.rules.yml") + ",target=/fixture/docker/monitoring/post76.rules.yml,readonly", "--mount", "type=bind,source=" + str(self.root / "ops/monitoring_rules.test.yml") + ",target=/fixture/ops/monitoring_rules.test.yml,readonly"] self.command(["docker", "run", "--rm", "--network", "none", "--read-only", *mounts, "--entrypoint", "promtool", "prom/prometheus:v3.13.4", "test", "rules", "/fixture/ops/monitoring_rules.test.yml"]) self.command(["docker", "run", "--rm", "--network", "none", "--read-only", "--mount", "type=bind,source=" + str(self.root / "docker/monitoring/alertmanager.yml") + ",target=/fixture/alerts.yml,readonly", "--entrypoint", "amtool", "prom/alertmanager:v0.34.1", "check-config", "/fixture/alerts.yml"]) key = self.workspace / "exporter-fixture-key" key.write_text("a" * 64) key.chmod(0o444) self.command(["docker", "run", "--rm", "--network", "none", "--read-only", "--user", "101:101", "--cap-drop", "ALL", "--security-opt", "no-new-privileges:true", "--tmpfs", "/run/post76:rw,noexec,nosuid,size=8m,uid=101,gid=101,mode=0700", "--mount", "type=bind,source=" + str(self.root / "docker/monitoring/exporter.sh") + ",target=/configuration/exporter.sh,readonly", "--mount", "type=bind,source=" + str(key) + ",target=/run/secrets/metrics_access_key,readonly", "--entrypoint", "sh", "nginx:1.30.5-alpine", "/configuration/exporter.sh", "--test"]) if __name__ == "__main__": root = Path(__file__).resolve().parent.parent workspace_root = root / "storage/app/private/ci-check" workspace_root.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="container-", dir=workspace_root) as workspace: StackCheck(root, workspace).run()