3933b0f397
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
443 lines
16 KiB
Python
443 lines
16 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
import textwrap
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
|
|
|
|
def _read_workload_db_identity_shell() -> str:
|
|
workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
|
|
start = workflow.index(" read_workload_db_identity_once() {")
|
|
end = workflow.index(" json_field() {", start)
|
|
return textwrap.dedent(workflow[start:end])
|
|
|
|
|
|
def _api_http_readiness_and_concurrency_python() -> str:
|
|
workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
|
|
marker = " API_HTTP_CONCURRENCY_VERIFIED=$(python3 - <<'PY'"
|
|
start = workflow.index(marker) + len(marker)
|
|
end = workflow.index("\n PY\n )", start)
|
|
return textwrap.dedent(workflow[start:end]).lstrip("\n")
|
|
|
|
|
|
class _FakeHTTPResponse:
|
|
def __init__(self, status: int) -> None:
|
|
self.status = status
|
|
|
|
def __enter__(self) -> "_FakeHTTPResponse":
|
|
return self
|
|
|
|
def __exit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
|
|
def _run_workload_db_identity_contract(
|
|
tmp_path: Path,
|
|
*,
|
|
exec_failures: int,
|
|
) -> tuple[subprocess.CompletedProcess[str], dict[str, int]]:
|
|
state_dir = tmp_path / "state"
|
|
state_dir.mkdir()
|
|
fake_kubectl = tmp_path / "kubectl"
|
|
fake_kubectl.write_text(
|
|
textwrap.dedent(
|
|
"""\
|
|
#!/usr/bin/env bash
|
|
set -eu
|
|
|
|
increment() {
|
|
counter_file="$FAKE_KUBECTL_STATE/$1.count"
|
|
COUNT=0
|
|
if [ -f "$counter_file" ]; then
|
|
COUNT=$(cat "$counter_file")
|
|
fi
|
|
COUNT=$((COUNT + 1))
|
|
printf '%s' "$COUNT" > "$counter_file"
|
|
}
|
|
|
|
case "${1:-}" in
|
|
rollout)
|
|
increment rollout
|
|
exit 0
|
|
;;
|
|
get)
|
|
increment get
|
|
if [[ " $* " == *" -o name "* ]]; then
|
|
echo "pod/awoooi-worker-ready"
|
|
else
|
|
echo "true"
|
|
fi
|
|
exit 0
|
|
;;
|
|
wait)
|
|
increment wait
|
|
exit 0
|
|
;;
|
|
exec)
|
|
increment exec
|
|
if [ "$COUNT" -le "$FAKE_KUBECTL_EXEC_FAILURES" ]; then
|
|
echo 'container not found ("worker")' >&2
|
|
exit 1
|
|
fi
|
|
cat <<'JSON'
|
|
{"role_fingerprint":"worker-fingerprint","connection_limit":8,"required_connection_budget":5,"active_role_connection_count":1,"available_connection_headroom":7,"representative_connection_probe_count":1,"representative_preflight_passed":true,"table_privilege_gap_count":0,"sequence_privilege_gap_count":0,"error_type":null}
|
|
JSON
|
|
exit 0
|
|
;;
|
|
esac
|
|
exit 2
|
|
"""
|
|
)
|
|
)
|
|
fake_kubectl.chmod(0o755)
|
|
shell = "\n".join(
|
|
(
|
|
"set -e",
|
|
'KUBECTL="$FAKE_KUBECTL"',
|
|
"AWOOOI_DB_IDENTITY_RETRY_BACKOFF_SECONDS=0",
|
|
_read_workload_db_identity_shell(),
|
|
"if result=$(read_workload_db_identity " "awoooi-worker worker 1 5); then",
|
|
" status=0",
|
|
"else",
|
|
" status=$?",
|
|
"fi",
|
|
'printf \'status=%s\\n%s\\n\' "$status" "$result"',
|
|
)
|
|
)
|
|
env = {
|
|
**os.environ,
|
|
"FAKE_KUBECTL": str(fake_kubectl),
|
|
"FAKE_KUBECTL_STATE": str(state_dir),
|
|
"FAKE_KUBECTL_EXEC_FAILURES": str(exec_failures),
|
|
}
|
|
completed = subprocess.run(
|
|
["bash", "-c", shell],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
counts = {
|
|
name: int((state_dir / f"{name}.count").read_text())
|
|
for name in ("rollout", "get", "wait", "exec")
|
|
}
|
|
return completed, counts
|
|
|
|
|
|
def _deployment_container(path: str, container_name: str) -> dict:
|
|
documents = yaml.safe_load_all((REPO_ROOT / path).read_text())
|
|
deployment = next(
|
|
document
|
|
for document in documents
|
|
if document and document.get("kind") == "Deployment"
|
|
)
|
|
return next(
|
|
container
|
|
for container in deployment["spec"]["template"]["spec"]["containers"]
|
|
if container["name"] == container_name
|
|
)
|
|
|
|
|
|
def _env_by_name(container: dict) -> dict[str, dict]:
|
|
return {entry["name"]: entry for entry in container.get("env", [])}
|
|
|
|
|
|
def test_runtime_workloads_use_distinct_database_secret_projections() -> None:
|
|
workload_specs = {
|
|
"api": (
|
|
"k8s/awoooi-prod/06-deployment-api.yaml",
|
|
"api",
|
|
"awoooi-api-db",
|
|
),
|
|
"worker": (
|
|
"k8s/awoooi-prod/08-deployment-worker.yaml",
|
|
"worker",
|
|
"awoooi-worker-db",
|
|
),
|
|
"broker": (
|
|
"k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml",
|
|
"broker",
|
|
"awoooi-broker-db",
|
|
),
|
|
}
|
|
observed_database_secret_refs: set[str] = set()
|
|
|
|
for workload_id, (path, container_name, expected_secret) in workload_specs.items():
|
|
container = _deployment_container(path, container_name)
|
|
secret_env_from = [
|
|
source for source in container.get("envFrom", []) if source.get("secretRef")
|
|
]
|
|
assert secret_env_from == [], workload_id
|
|
|
|
env_names = [entry["name"] for entry in container.get("env", [])]
|
|
assert len(env_names) == len(set(env_names)), workload_id
|
|
env = _env_by_name(container)
|
|
database_ref = env["DATABASE_URL"]["valueFrom"]["secretKeyRef"]
|
|
assert database_ref == {
|
|
"name": expected_secret,
|
|
"key": "DATABASE_URL",
|
|
}
|
|
observed_database_secret_refs.add(database_ref["name"])
|
|
assert env["DATABASE_NULL_POOL"]["value"] == "true"
|
|
|
|
assert observed_database_secret_refs == {
|
|
"awoooi-api-db",
|
|
"awoooi-worker-db",
|
|
"awoooi-broker-db",
|
|
}
|
|
|
|
|
|
def test_api_and_worker_project_shared_secrets_explicitly() -> None:
|
|
for path, container_name in (
|
|
("k8s/awoooi-prod/06-deployment-api.yaml", "api"),
|
|
("k8s/awoooi-prod/08-deployment-worker.yaml", "worker"),
|
|
):
|
|
env = _env_by_name(_deployment_container(path, container_name))
|
|
for env_name in (
|
|
"REDIS_URL",
|
|
"GEMINI_API_KEY",
|
|
"OPENCLAW_TG_BOT_TOKEN",
|
|
"SRE_GROUP_CHAT_ID",
|
|
"WEBHOOK_HMAC_SECRET",
|
|
"AWOOOP_OPERATOR_API_KEY",
|
|
):
|
|
secret_ref = env[env_name]["valueFrom"]["secretKeyRef"]
|
|
assert secret_ref["name"] == "awoooi-secrets"
|
|
assert secret_ref["key"] == env_name
|
|
assert secret_ref["optional"] is True
|
|
|
|
|
|
def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None:
|
|
script = (
|
|
REPO_ROOT / "scripts/ops/awoooi-workload-db-identity-bootstrap.sh"
|
|
).read_text()
|
|
|
|
assert "set +x" in script
|
|
assert "openssl rand -hex 32" in script
|
|
assert "--from-file=DATABASE_URL=/dev/stdin" in script
|
|
assert "--from-literal=DATABASE_URL" not in script
|
|
assert "docker exec -i -u postgres" in script
|
|
assert "table_privilege_gap_count" in script
|
|
assert "sequence_privilege_gap_count" in script
|
|
assert "REPRESENTATIVE_CONNECTION_PROBE_COUNT" in script
|
|
assert "REQUIRED_CONCURRENT_CONNECTIONS" not in script
|
|
assert "representative_connection_probe_count" in script
|
|
assert "active_role_connection_count" in script
|
|
assert "available_connection_headroom" in script
|
|
assert "connection_headroom_below_required" in script
|
|
assert "FROM pg_stat_activity" in script
|
|
assert "pid <> pg_backend_pid()" in script
|
|
assert "for preflight_attempt in 1 2 3" in script
|
|
assert "preflight_attempts_exhausted=3" in script
|
|
assert '"connection_headroom_below_required"' in script
|
|
assert 'error_code = "workload_database_preflight_failed"' in script
|
|
assert '"error_code": error_code' in script
|
|
assert 'text("SELECT pg_sleep(0.25)")' in script
|
|
assert "api) printf '12'" in script
|
|
assert "worker) printf '8'" in script
|
|
assert "broker) printf '4'" in script
|
|
assert '"incidents"' in script
|
|
assert '"knowledge_entries"' in script
|
|
assert '"awooop_run_state"' in script
|
|
assert "ALTER ROLE %I NOLOGIN" in script
|
|
assert "DROP ROLE" not in script.upper()
|
|
assert "base64 -d" not in script
|
|
assert "GITHUB_RUN_ID" not in script
|
|
assert "check|apply|verify|disable" in script
|
|
assert '"secret_value_exposed": False' in script
|
|
assert '"secret_value_logged": False' in script
|
|
assert "secret_value_read" not in script
|
|
|
|
|
|
def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
|
|
workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
|
|
|
|
assert "WORKLOAD_DB_ROLLBACK_FILE" in workflow
|
|
assert "WORKLOAD_DB_SECRET_PREEXISTENCE_FILE" in workflow
|
|
assert "restore_workload_database_projection" in workflow
|
|
assert "bash -s -- apply" in workflow
|
|
assert "bash -s -- verify" in workflow
|
|
assert workflow.index("bash -s -- apply") < workflow.index("bash -s -- verify")
|
|
assert '"spec": item["spec"]' in workflow
|
|
assert '"data": item' not in workflow
|
|
assert "get secret '${secret_name}' -n awoooi-prod -o jsonpath" not in workflow
|
|
assert "get secret awoooi-api-db -n awoooi-prod -o jsonpath" not in workflow
|
|
assert "HEAD^" not in workflow
|
|
assert 'while [ "$attempt" -le 3 ]' in workflow
|
|
assert 'rollout status "deployment/$workload"' in workflow
|
|
assert 'wait --for=condition=Ready "$ready_pod_ref"' in workflow
|
|
assert 'exec -i "$ready_pod_ref"' in workflow
|
|
assert "container_not_ready" in workflow
|
|
assert "WorkloadDbIdentityAttemptsExhausted" in workflow
|
|
assert "workload DB identity attempts exhausted before receipt write" in workflow
|
|
assert "API_PUBLIC_READINESS_ATTEMPTS" in workflow
|
|
assert "API_HTTP_CONCURRENCY_ATTEMPTS" in workflow
|
|
assert "API sequential readiness permanent HTTP status" in workflow
|
|
assert "api/v1/health/ready" in workflow
|
|
assert "events/dossier/coverage?project_id=awoooi&limit=1" not in workflow
|
|
assert "api/v1/platform/cicd/events" in workflow
|
|
assert "?project_id=awoooi&limit=1" in workflow
|
|
assert "concurrency_urls = (db_probe_url,) * 8" in workflow
|
|
assert "concurrency_urls = urls * 6" not in workflow
|
|
assert "API_RESTART_COUNT_BEFORE=$(api_restart_total)" in workflow
|
|
assert "API_RESTART_COUNT_AFTER=$(api_restart_total)" in workflow
|
|
assert '"$API_PROBE_RESTART_FREE" != "true"' in workflow
|
|
assert "API concurrency probe destabilized rollout" in workflow
|
|
assert "api_http_probe_restart_free" in workflow
|
|
assert workflow.index("API_HTTP_CONCURRENCY_VERIFIED=$(python3") < workflow.index(
|
|
"$KUBECTL create configmap awoooi-executor-boundary-verification"
|
|
)
|
|
|
|
|
|
def test_cd_workload_db_identity_retries_transient_worker_rollout_race(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
completed, counts = _run_workload_db_identity_contract(
|
|
tmp_path,
|
|
exec_failures=1,
|
|
)
|
|
|
|
lines = completed.stdout.strip().splitlines()
|
|
assert lines[0] == "status=0"
|
|
payload = json.loads(lines[-1])
|
|
assert payload["role_fingerprint"] == "worker-fingerprint"
|
|
assert payload["available_connection_headroom"] == 7
|
|
assert payload["representative_connection_probe_count"] == 1
|
|
assert payload["representative_preflight_passed"] is True
|
|
assert counts == {"rollout": 2, "get": 4, "wait": 2, "exec": 2}
|
|
assert "reason=kubectl_exec_failed" in completed.stderr
|
|
assert "workload_db_identity_exec_retry=1/3" in completed.stderr
|
|
assert "workload_db_identity_blocked" not in completed.stderr
|
|
|
|
|
|
def test_cd_workload_db_identity_blocks_after_three_worker_exec_failures(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
completed, counts = _run_workload_db_identity_contract(
|
|
tmp_path,
|
|
exec_failures=3,
|
|
)
|
|
|
|
lines = completed.stdout.strip().splitlines()
|
|
assert lines[0] == "status=1"
|
|
payload = json.loads(lines[-1])
|
|
assert payload["role_fingerprint"] == ""
|
|
assert payload["representative_preflight_passed"] is False
|
|
assert payload["error_type"] == "WorkloadDbIdentityAttemptsExhausted"
|
|
assert counts == {"rollout": 3, "get": 6, "wait": 3, "exec": 3}
|
|
assert "workload_db_identity_exec_retry=3/3" in completed.stderr
|
|
assert "workload_db_identity_exec_exhausted=3" in completed.stderr
|
|
|
|
|
|
def test_cd_api_readiness_retries_http_503_then_runs_all_200_concurrency(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
responses = [503, 200, *([200] * 8)]
|
|
calls: list[str] = []
|
|
|
|
def fake_urlopen(url: str, *, timeout: int) -> _FakeHTTPResponse:
|
|
calls.append(f"{url}|{timeout}")
|
|
status = responses.pop(0)
|
|
if status == 503:
|
|
raise urllib.error.HTTPError(url, status, "unavailable", None, None)
|
|
return _FakeHTTPResponse(status)
|
|
|
|
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
|
|
monkeypatch.setattr(time, "sleep", lambda _seconds: None)
|
|
monkeypatch.setenv("API_PUBLIC_READINESS_ATTEMPTS", "3")
|
|
monkeypatch.setenv("API_PUBLIC_READINESS_DELAY_SECONDS", "0")
|
|
monkeypatch.setenv("API_HTTP_CONCURRENCY_ATTEMPTS", "2")
|
|
monkeypatch.setenv("API_HTTP_CONCURRENCY_DELAY_SECONDS", "0")
|
|
|
|
exec(
|
|
compile(
|
|
_api_http_readiness_and_concurrency_python(),
|
|
"<cd-api-http-verifier>",
|
|
"exec",
|
|
),
|
|
{},
|
|
)
|
|
|
|
output = capsys.readouterr()
|
|
assert output.out.strip() == "true"
|
|
assert "api_sequential_readiness=ready;attempt=2" in output.err
|
|
assert len(calls) == 10
|
|
assert calls[0].startswith("https://awoooi.wooo.work/api/v1/health/ready|")
|
|
assert calls[1].startswith("https://awoooi.wooo.work/api/v1/health/ready|")
|
|
assert all("platform/cicd/events" in call for call in calls[2:])
|
|
assert responses == []
|
|
|
|
|
|
def test_cd_api_readiness_exhaustion_stops_before_green_receipt_write(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
calls: list[str] = []
|
|
|
|
def always_unavailable(url: str, *, timeout: int) -> _FakeHTTPResponse:
|
|
calls.append(f"{url}|{timeout}")
|
|
raise urllib.error.HTTPError(url, 503, "unavailable", None, None)
|
|
|
|
monkeypatch.setattr(urllib.request, "urlopen", always_unavailable)
|
|
monkeypatch.setattr(time, "sleep", lambda _seconds: None)
|
|
monkeypatch.setenv("API_PUBLIC_READINESS_ATTEMPTS", "3")
|
|
monkeypatch.setenv("API_PUBLIC_READINESS_DELAY_SECONDS", "0")
|
|
namespace = {"receipt_write_reached": False}
|
|
verifier = _api_http_readiness_and_concurrency_python()
|
|
|
|
with pytest.raises(SystemExit, match="API sequential readiness failed"):
|
|
exec(
|
|
compile(
|
|
verifier + "\nreceipt_write_reached = True\n",
|
|
"<cd-api-http-verifier>",
|
|
"exec",
|
|
),
|
|
namespace,
|
|
)
|
|
|
|
output = capsys.readouterr()
|
|
assert output.out == ""
|
|
assert namespace["receipt_write_reached"] is False
|
|
assert len(calls) == 3
|
|
assert all("api/v1/health/ready" in call for call in calls)
|
|
|
|
|
|
def test_cd_api_readiness_permanent_http_error_fails_without_retry(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
calls: list[str] = []
|
|
|
|
def unauthorized(url: str, *, timeout: int) -> _FakeHTTPResponse:
|
|
calls.append(f"{url}|{timeout}")
|
|
raise urllib.error.HTTPError(url, 401, "unauthorized", None, None)
|
|
|
|
monkeypatch.setattr(urllib.request, "urlopen", unauthorized)
|
|
monkeypatch.setattr(time, "sleep", lambda _seconds: None)
|
|
namespace = {"receipt_write_reached": False}
|
|
verifier = _api_http_readiness_and_concurrency_python()
|
|
|
|
with pytest.raises(SystemExit, match="permanent HTTP status: 401"):
|
|
exec(
|
|
compile(
|
|
verifier + "\nreceipt_write_reached = True\n",
|
|
"<cd-api-http-verifier>",
|
|
"exec",
|
|
),
|
|
namespace,
|
|
)
|
|
|
|
output = capsys.readouterr()
|
|
assert output.out == ""
|
|
assert namespace["receipt_write_reached"] is False
|
|
assert len(calls) == 1
|