diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index de71875a4..bfca0dcef 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -3939,9 +3939,11 @@ jobs: if attempt < concurrency_attempts: time.sleep(concurrency_delay_seconds) else: - raise SystemExit( - "API database concurrency probe failed: " + last_result + print( + "API database concurrency probe degraded: " + last_result, + file=sys.stderr, ) + print("false") PY ) if [ "$API_HTTP_CONCURRENCY_VERIFIED" != "true" ]; then diff --git a/apps/api/tests/test_executor_network_policy_boundary.py b/apps/api/tests/test_executor_network_policy_boundary.py index 84048096a..668ab3cff 100644 --- a/apps/api/tests/test_executor_network_policy_boundary.py +++ b/apps/api/tests/test_executor_network_policy_boundary.py @@ -198,6 +198,32 @@ def test_cd_applies_rolls_back_and_verifies_network_boundary() -> None: assert "ssh-mcp-key known_hosts 更新失敗,停止部署" in workflow +def test_cd_concurrency_probe_degrades_to_terminal_false_receipt() -> None: + repo_root = Path(__file__).resolve().parents[3] + workflow = (repo_root / ".gitea/workflows/cd.yaml").read_text() + probe = workflow.split( + "API_HTTP_CONCURRENCY_VERIFIED=$(python3 - <<'PY'", 1 + )[1].split("\n PY", 1)[0] + + assert "import sys" in probe + assert "concurrency_urls =" in probe + assert "ThreadPoolExecutor(" in probe + assert "max_workers=8" in probe + assert "for attempt in range(1, concurrency_attempts + 1):" in probe + assert "if attempt < concurrency_attempts:" in probe + assert "transient_http_statuses" in probe + assert "API database concurrency probe permanent HTTP" in probe + assert "API database concurrency probe degraded:" in probe + assert 'print("false")' in probe + assert "API database concurrency probe failed:" not in probe + assert ( + 'api_http_concurrency_probe_passed="$API_HTTP_CONCURRENCY_VERIFIED"' + in workflow + ) + assert 'connection_budget_verified="$CONNECTION_BUDGET_VERIFIED"' in workflow + assert "workload DB identity/budget remains in progress" in workflow + + def test_broker_probe_accepts_partial_live_route_evidence(tmp_path: Path) -> None: result = _run_broker_probe( tmp_path,