#!/usr/bin/env python3 from __future__ import annotations import re from pathlib import Path ROOT = Path(__file__).resolve().parents[2] CD_WORKFLOW = ROOT / ".gitea" / "workflows" / "cd.yaml" HARBOR_110_REPAIR_WORKFLOW = ( ROOT / ".gitea" / "workflows" / "harbor-110-local-repair.yaml" ) WAIT_HOST_PRESSURE = ROOT / "scripts" / "ci" / "wait-host-web-build-pressure.sh" PROD_API_DEPLOYMENT = ROOT / "k8s" / "awoooi-prod" / "06-deployment-api.yaml" PROD_WORKER_DEPLOYMENT = ROOT / "k8s" / "awoooi-prod" / "08-deployment-worker.yaml" PROD_CANARY_DEPLOYMENT = ( ROOT / "k8s" / "awoooi-prod" / "10-deployment-auto-repair-canary.yaml" ) CD_LANE_DRAIN_SERVICE = ROOT / "ops" / "runner" / "awoooi-cd-lane-drain.service" def _workflow_text() -> str: return CD_WORKFLOW.read_text(encoding="utf-8") def test_cd_lane_drain_service_source_keeps_controlled_guardrails() -> None: text = CD_LANE_DRAIN_SERVICE.read_text(encoding="utf-8") expected = [ "ConditionPathExists=/home/wooo/awoooi-cd-lane-drain/data/.runner", "WorkingDirectory=/home/wooo/awoooi-cd-lane-drain/data", "Environment=AWOOOI_CONTROLLED_RUNNER_OPEN=1", "Environment=HOST_WEB_BUILD_PRESSURE_ATTEMPTS=1", "Environment=HOST_WEB_BUILD_PRESSURE_SLEEP_SECONDS=1", "ExecStart=/home/wooo/awoooi-cd-lane-drain/awoooi_cd_lane_controlled daemon --config /home/wooo/awoooi-cd-lane-drain/config.yaml", "SuccessExitStatus=0 130 143", "CPUAccounting=true", "CPUQuota=250%", "MemoryAccounting=true", "MemoryHigh=8G", "MemoryMax=12G", "TasksAccounting=true", "TasksMax=512", "IOAccounting=true", "IOWeight=100", "NoNewPrivileges=true", ] for line in expected: assert line in text forbidden = [ "ubuntu-latest", "self-hosted", "stockplatform", "headless", "playwright", ] for label in forbidden: assert label not in text def test_web_changes_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() assert "apps/web/*)" in text assert "UI-only changes are verified by the" in text def test_product_manifest_changes_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() assert "product.awoooi.yaml)" in text assert "apps/api/Dockerfile)" in text assert "docs/schemas/product_awoooi_manifest_v1.schema.json)" in text assert "apps/api/src/services/product_awoooi_manifest_standard.py)" in text assert "tests/test_product_awoooi_manifest_standard_api.py" in text def test_deploy_marker_k8s_files_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() assert "build-and-deploy writes only these GitOps" in text assert "k8s/awoooi-prod/04-configmap.yaml)" in text assert "k8s/awoooi-prod/06-deployment-api.yaml)" in text assert "k8s/awoooi-prod/08-deployment-worker.yaml)" in text assert "k8s/awoooi-prod/10-deployment-auto-repair-canary.yaml)" in text assert "k8s/awoooi-prod/kustomization.yaml)" in text def test_public_maintenance_fallback_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() assert "public maintenance fallback Nginx" in text expected_sources = [ "docs/runbooks/PUBLIC-MAINTENANCE-FALLBACK-RUNBOOK.md)", "docs/workplans/2026-07-02-commander-inserted-requirements-priority-ledger.md)", "k8s/nginx/awoooi-prod.conf)", "ops/maintenance/maintenance.html)", "ops/maintenance/nginx-502-maintenance-snippet.conf)", "ops/nginx/awoooi.wooo.work.conf)", "scripts/reboot-recovery/tests/test_reboot_p0_operational_contract.py)", ] for source in expected_sources: assert source in text def test_prod_db_pool_recovery_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/core/config.py)", "apps/api/src/db/base.py)", "apps/api/tests/test_config_url_validation.py)", "apps/api/tests/test_runtime_bootstrap_guards.py)", "src/core/config.py", "src/db/base.py", "tests/test_config_url_validation.py", "tests/test_runtime_bootstrap_guards.py", ] for source in expected_sources: assert source in text def test_prod_db_pool_budget_and_non_overlap_rollouts_are_source_controlled() -> None: api = PROD_API_DEPLOYMENT.read_text(encoding="utf-8") worker = PROD_WORKER_DEPLOYMENT.read_text(encoding="utf-8") canary = PROD_CANARY_DEPLOYMENT.read_text(encoding="utf-8") assert "DATABASE_POOL_SIZE" in api assert "value: \"1\"" in api assert "DATABASE_MAX_OVERFLOW" in api assert "value: \"0\"" in api assert "DATABASE_POOL_SIZE" in worker assert "DATABASE_MAX_OVERFLOW" in worker assert "maxSurge: 0" in worker assert "maxUnavailable: 1" in worker assert "post-reboot DB connection pressure recovery" in canary assert "maxSurge: 0" in canary assert "maxUnavailable: 1" in canary def test_workflow_secret_transport_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() assert "workflow secret-transport and guard-only" in text expected_sources = [ ".gitea/workflows/cd-dev.yaml)", ".gitea/workflows/code-review.yaml)", ".gitea/workflows/deploy-alerts.yaml)", ".gitea/workflows/e2e-health.yaml)", ".gitea/workflows/ansible-lint.yml)", ".gitea/workflows/harbor-110-local-repair.yaml)", ".gitea/workflows/run-migration.yml)", "scripts/ci/check-gitea-step-env-secrets.js)", ] for source in expected_sources: assert source in text def test_backup_freshness_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "scripts/backup/backup-awoooi-frequent.sh)", "scripts/backup/backup-status.sh)", "scripts/backup/gitea-repo-bundle-backup.sh)", "scripts/backup/gitea-bundle-sample-restore-dry-run.sh)", "scripts/backup/tests/test_backup_status_contract.py)", "scripts/backup/tests/test_gitea_bundle_sample_restore_dry_run.py)", "scripts/ops/backup-alert-label-contract-check.py)", "scripts/ops/backup-health-textfile-exporter.py)", "scripts/ops/tests/test_backup_health_textfile_exporter.py)", "scripts/ops/gitea-queue-hook-backlog-playbook.py)", "scripts/ops/host-runaway-process-exporter.py)", "scripts/ops/host-sustained-load-controller.py)", "scripts/ops/host-sustained-load-evidence.py)", "scripts/ops/deploy-alertmanager-config.sh)", "scripts/ops/tests/test_gitea_queue_hook_backlog_playbook.py)", "scripts/ops/tests/test_host_runaway_process_exporter.py)", "scripts/ops/tests/test_host_pressure_alert_contract.py)", "scripts/ops/tests/test_alertmanager_webhook_config.py)", "../../scripts/backup/backup-awoooi-frequent.sh", "../../scripts/backup/backup-status.sh", "../../scripts/backup/gitea-bundle-sample-restore-dry-run.sh", "../../scripts/backup/tests/test_backup_status_contract.py", "../../scripts/backup/tests/test_gitea_bundle_sample_restore_dry_run.py", "../../scripts/ops/backup-alert-label-contract-check.py", "../../scripts/ops/gitea-queue-hook-backlog-playbook.py", "../../scripts/ops/host-runaway-process-exporter.py", "../../scripts/ops/host-sustained-load-controller.py", "../../scripts/ops/host-sustained-load-evidence.py", "../../scripts/ops/deploy-alertmanager-config.sh", "../../scripts/ops/tests/test_backup_health_textfile_exporter.py", "../../scripts/ops/tests/test_gitea_queue_hook_backlog_playbook.py", "../../scripts/ops/tests/test_host_runaway_process_exporter.py", "../../scripts/ops/tests/test_host_pressure_alert_contract.py", "../../scripts/ops/tests/test_alertmanager_webhook_config.py", ] for source in expected_sources: assert source in text def test_log_controlled_writeback_consumer_apply_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py)", "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_apply.py)", "apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py)", "apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py)", "src/services/ai_agent_log_controlled_writeback_consumer_readback.py", "src/services/ai_agent_log_controlled_writeback_consumer_apply.py", "tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py", "tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py", ] for source in expected_sources: assert source in text def test_knowledge_readback_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/repositories/knowledge_repository.py)", "apps/api/src/services/knowledge_service.py)", "apps/api/tests/test_knowledge_repository_read_model.py)", "src/repositories/knowledge_repository.py", "src/services/knowledge_service.py", ] for source in expected_sources: assert source in text def test_telegram_alert_ai_automation_matrix_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "docs/awooop/TELEGRAM-INCIDENT-NOTIFICATION-MODEL.md)", "docs/security/TELEGRAM-NOTIFICATION-EGRESS-INVENTORY.md)", "docs/security/TELEGRAM-NOTIFICATION-EGRESS-MIGRATION-PLAN-DRAFT.md)", "docs/security/TELEGRAM-NOTIFICATION-EGRESS-NO-NEW-BYPASS-GUARD.md)", "docs/security/TELEGRAM-NOTIFICATION-EGRESS-OWNER-REQUEST-DRAFT.md)", "docs/security/TELEGRAM-NOTIFICATION-EGRESS-OWNER-RESPONSE-ACCEPTANCE.md)", "docs/security/telegram-notification-egress-inventory.snapshot.json)", "docs/security/telegram-notification-egress-owner-request-draft.snapshot.json)", "docs/security/telegram-notification-egress-migration-plan-draft.snapshot.json)", "docs/security/telegram-notification-egress-no-new-bypass-guard.snapshot.json)", "docs/security/telegram-notification-egress-owner-response-acceptance.snapshot.json)", "apps/api/src/api/v1/platform/operator_runs.py)", "apps/api/src/services/channel_hub.py)", "apps/api/src/services/telegram_alert_ai_automation_matrix.py)", "apps/api/tests/test_channel_hub_grouped_alert_events.py)", "apps/api/tests/test_ai_agent_report_truth_actionability_review.py)", "apps/api/tests/test_ai_agent_report_truth_actionability_review_api.py)", "apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py)", "scripts/ops/backup-from-110.sh)", "scripts/ops/docker-health-monitor.sh)", "scripts/ops/dr-drill.sh)", "scripts/ops/notify-awoooi-ops.sh)", "scripts/ops/pg-backup.sh)", "scripts/security/security-mirror-progress-guard.py)", "scripts/security/telegram-notification-egress-no-new-bypass-guard.py)", "scripts/security/telegram-notification-egress-owner-response-acceptance.py)", "src/services/channel_hub.py", "src/api/v1/platform/operator_runs.py", "src/services/telegram_alert_ai_automation_matrix.py", "tests/test_channel_hub_grouped_alert_events.py", "tests/test_ai_agent_report_truth_actionability_review.py", "tests/test_ai_agent_report_truth_actionability_review_api.py", "tests/test_telegram_alert_ai_automation_matrix_api.py", "../../scripts/security/security-mirror-progress-guard.py", "../../scripts/security/telegram-notification-egress-no-new-bypass-guard.py", "../../scripts/security/telegram-notification-egress-owner-response-acceptance.py", "../../scripts/ops/backup-from-110.sh", "../../scripts/ops/docker-health-monitor.sh", "../../scripts/ops/dr-drill.sh", "../../scripts/ops/notify-awoooi-ops.sh", "../../scripts/ops/pg-backup.sh", ] for source in expected_sources: assert source in text def test_telegram_alert_monitoring_coverage_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/telegram_alert_monitoring_coverage_readback.py)", "apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py)", "src/services/telegram_alert_monitoring_coverage_readback.py", "tests/test_telegram_alert_monitoring_coverage_readback_api.py", ] for source in expected_sources: assert source in text def test_alertmanager_webhook_metrics_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/api/v1/webhooks.py)", "apps/api/tests/test_alertmanager_webhook_metrics.py)", "src/api/v1/webhooks.py", "tests/test_alertmanager_webhook_metrics.py", ] for source in expected_sources: assert source in text def test_cd_requires_production_deploy_readback_after_rollout() -> None: text = _workflow_text() assert "apps/api/tests/test_awoooi_production_deploy_readback_blocker.py)" in text assert "tests/test_awoooi_production_deploy_readback_blocker.py" in text assert "production_deploy_readback_mismatch=" in text assert 'attempts = int(os.environ.get("DEPLOY_READBACK_ATTEMPTS", "36"))' in text assert 'sleep_seconds = int(os.environ.get("DEPLOY_READBACK_SLEEP_SECONDS", "10"))' in text assert "production_deploy_readback_attempt=" in text assert "time.sleep(sleep_seconds)" in text assert "production_deploy_runtime_build_commit_short_sha" in text assert "production_deploy_desired_main_api_image_tag_short_sha" in text assert "production_deploy_desired_main_api_image_tag_readback_status" in text assert "Production deploy readback matches this build and " in text assert "GitOps desired image tag " in text success_block = text.split( '"✅ Production deploy readback matches this build and "', 1 )[0].rsplit("if (", 1)[-1] assert "runtime_short == expected_short" in success_block assert "desired_short == expected_short" in success_block assert 'desired_status == "ok"' in success_block assert "and matches_main" not in success_block assert "DEPLOY_READBACK_EXIT=0" in text assert "production_deploy_readback_matched=true" in text assert "treating as rollout risk, not deploy failure" in text def test_post_deploy_smoke_uses_workspace_playwright_dependency() -> None: text = _workflow_text() block = text.split("- name: E2E Smoke Test", 1)[1] block = block.split("- name: Notify Success", 1)[0] assert "set -euo pipefail" in block assert "tee /tmp/pnpm-install.log | tail -20" in block assert 'pnpm --dir apps/web exec node -e "require.resolve(\'@playwright/test\')"' in block assert "pnpm exec playwright install chromium --with-deps" in block assert "pnpm exec playwright install-deps chromium" in block assert "pnpm exec playwright test tests/e2e/smoke.spec.ts --reporter=line" in block assert "chromium.executablePath()" in block assert '[ ! -x "$BROWSER_PATH" ]' in block assert 'test -x "$BROWSER_PATH"' in block assert "Playwright browser cache missing/stale" in block assert "SMOKE_WORKDIR=/tmp/awoooi-smoke-workspace" in block assert "-v \"$PWD:/source:ro\"" in block assert "-v \"$SMOKE_OUTPUT:/github-output\"" in block assert "-w /tmp" in block assert "-e GITHUB_OUTPUT=/github-output" in block assert "-v \"$PWD:/workspace\"" not in block assert "-w /workspace" not in block assert "GITHUB_OUTPUT=/workspace/.awoooi-smoke-output" not in block assert "rm -rf /workspace/node_modules" not in block assert "npx playwright" not in block def test_harbor_login_has_public_route_retry_and_safe_secret_transport() -> None: text = _workflow_text() block = text.split("- name: Login to Harbor", 1)[1] block = block.split("- name: Wait for Host Web Build Pressure", 1)[0] assert 'LOGIN_ATTEMPTS="${HARBOR_LOGIN_ATTEMPTS:-3}"' in block assert 'LOGIN_SLEEP_SECONDS="${HARBOR_LOGIN_SLEEP_SECONDS:-5}"' in block assert 'WATCHDOG="/usr/local/bin/harbor-watchdog.sh"' in block assert ( 'HARBOR_110_NODE_EXPORTER_URL="${HARBOR_110_NODE_EXPORTER_URL:-http://192.168.0.110:9100/metrics}"' in block ) assert 'HARBOR_110_MAX_LOAD5_PER_CORE="${HARBOR_110_MAX_LOAD5_PER_CORE:-1.25}"' in block assert 'HARBOR_110_MAX_GITEA_CPU_CORES="${HARBOR_110_MAX_GITEA_CPU_CORES:-3.0}"' in block assert "preflight_110_capacity()" in block assert "harbor_110_load5_per_core=" in block assert "BLOCKER harbor_110_host_pressure_high" in block assert "harbor_110_gitea_cpu_cores=" in block assert "BLOCKER harbor_110_gitea_cpu_saturated" in block assert "host_has_110_ip()" in block assert 'grep -q " 192.168.0.110/"' in block assert 'AWOOOI_CD_HARBOR_CONTROLLED_REPAIR:-1' in block assert "harbor_controlled_repair_check_start=1" in block assert "harbor_controlled_repair_once_start=1" in block assert "harbor_controlled_repair_public_registry_v2_status=" in block assert "sudo -n" in block assert '"https://${HARBOR_REGISTRY}/v2/"' in block assert 'registry_status="000"' in block assert '[ "${registry_status}" = "200" ] || [ "${registry_status}" = "401" ]' in block assert "docker login" in block assert "--password-stdin" in block assert "BLOCKER harbor_registry_public_route_unavailable" in block assert "sudo /usr/local/bin/harbor-watchdog.sh --check" in block assert "sudo /usr/local/bin/harbor-watchdog.sh --repair-once" in block assert ( "sudo /usr/local/bin/recover-110-control-path-and-harbor-local.sh --apply-all" in block ) assert ".gitea/workflows/harbor-110-local-repair.yaml" in block assert "systemctl restart docker" not in block assert "\nreboot" not in block assert "sleep \"${LOGIN_SLEEP_SECONDS}\"" in block assert "${HARBOR_PASSWORD}" in block assert "--password " not in block def test_host_pressure_gate_ignores_stale_docker_stats_for_cpu_attribution() -> None: text = WAIT_HOST_PRESSURE.read_text(encoding="utf-8") assert "MAX_DOCKER_METRICS_AGE_SECONDS" in text assert 'mtime="$(stat -c %Y "$DOCKER_METRICS_FILE"' in text assert 'if [ "$age" -gt "$MAX_DOCKER_METRICS_AGE_SECONDS" ]; then' in text assert "docker_container_cpu_cores" in text def test_harbor_110_local_repair_workflow_is_dispatch_only_and_bounded() -> None: text = HARBOR_110_REPAIR_WORKFLOW.read_text(encoding="utf-8") assert "workflow_dispatch:" in text assert "schedule:" in text assert 'cron: "17 * * * *"' in text assert "push:" not in text assert "pull_request:" not in text assert "pull_request_target:" not in text assert "runs-on: awoooi-non110-host" in text assert "runs-on: awoooi-host" not in text assert "guard-gitea-runner-pressure.py --root ." in text assert "recover-110-control-path-and-harbor-local.sh" in text assert "--check" in text assert "--apply-all" in text assert "sudo -n env" in text assert "AWOOOI_110_SSH_TARGET" in text assert "BatchMode=yes" in text assert "-n" in text assert "-T" in text assert "StdinNull=yes" in text assert "RequestTTY=no" in text assert "PreferredAuthentications=publickey" in text assert "PasswordAuthentication=no" in text assert "KbdInteractiveAuthentication=no" in text assert "GSSAPIAuthentication=no" in text assert "NumberOfPasswordPrompts=0" in text assert "ConnectionAttempts=1" in text assert 'SSH_PROBE_ATTEMPTS="${AWOOOI_110_SSH_PROBE_ATTEMPTS:-6}"' in text assert ( 'SSH_PROBE_SLEEP_SECONDS="${AWOOOI_110_SSH_PROBE_SLEEP_SECONDS:-10}"' in text ) assert "else\n rc=$?" in text assert "harbor_110_remote_ssh_probe_attempt=" in text assert 'AWOOOI_110_SSH_CONNECT_TIMEOUT_SECONDS: "3"' in text assert 'AWOOOI_110_SSH_COMMAND_TIMEOUT_SECONDS: "12"' in text assert 'ConnectTimeout="${AWOOOI_110_SSH_CONNECT_TIMEOUT_SECONDS}"' in text assert ( 'timeout --foreground --kill-after=3s "${AWOOOI_110_SSH_COMMAND_TIMEOUT_SECONDS}"' in text ) assert "timeout --foreground --kill-after=3s 20 ssh -vvv -4" in text assert 'timeout --foreground --kill-after=5s 240 "${ssh_base[@]}"' in text assert "operation_boundary_remote_ssh_bounded=true" in text assert "harbor_110_remote_control_channel_unavailable" in text assert "harbor_110_remote_ssh_server_accepts_key_then_session_timeout" in text assert "classification=server_accepts_key_then_timeout" in text assert "harbor_110_remote_ssh_diag_raw_log_printed=false" in text assert "harbor_110_remote_repair_check_start=1" in text assert 'if [ "${check_rc}" -ne 0 ]; then' in text assert "BLOCKED harbor_110_remote_repair_check_unavailable" in text assert "harbor_110_remote_repair_apply_all_start=1" in text assert "harbor_110_remote_repair_apply_all_rc=" in text assert "BLOCKED harbor_110_remote_repair_apply_all_failed" in text assert "GITHUB_EVENT_NAME" in text assert "harbor_110_remote_repair_skipped=already_ready" in text assert "192.168.0.110" in text assert "http://127.0.0.1:5000/v2/" in text assert "https://registry.wooo.work/v2/" in text assert "HARBOR_PASSWORD" not in text assert "secrets." not in text assert "systemctl restart docker" not in text assert "\nreboot" not in text assert "kubectl drain" not in text def test_non110_cd_lane_keeps_pressure_guard_fail_hard_with_bounded_load_threshold() -> None: text = _workflow_text() assert 'HOST_WEB_BUILD_PRESSURE_WARN_ONLY: "0"' in text assert 'HOST_WEB_BUILD_PRESSURE_MAX_LOAD5_PER_CORE: "1.05"' in text assert "awoooi-non110-host" in text assert 'HOST_WEB_BUILD_PRESSURE_WARN_ONLY: "1"' not in text assert "warn-only" in text def test_non110_runner_keepalive_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "ops/runner/check-awoooi-non110-runner-readiness.sh)", "ops/runner/install-awoooi-non110-runner-user-service.sh)", "ops/runner/register-awoooi-110-controlled-cd-lane-drain.sh)", "ops/runner/test_check_awoooi_non110_runner_readiness.py)", "ops/runner/test_install_awoooi_non110_runner_user_service.py)", "ops/runner/test_register_awoooi_110_controlled_cd_lane_drain.py)", "../../ops/runner/check-awoooi-non110-runner-readiness.sh", "../../ops/runner/install-awoooi-non110-runner-user-service.sh", "../../ops/runner/register-awoooi-110-controlled-cd-lane-drain.sh", "../../ops/runner/test_check_awoooi_non110_runner_readiness.py", "../../ops/runner/test_install_awoooi_non110_runner_user_service.py", "../../ops/runner/test_register_awoooi_110_controlled_cd_lane_drain.py", ] for source in expected_sources: assert source in text def test_harbor_watchdog_exposes_controlled_check_and_one_shot_repair() -> None: text = (ROOT / "scripts/reboot-recovery/harbor-watchdog.sh").read_text( encoding="utf-8" ) assert "--check" in text assert "--repair-once" in text assert "--apply-once" in text assert "check_only=true" in text assert "docker_compose_action_performed=false" in text assert "container_remove_performed=false" in text assert "AWOOI_ALLOW_NON_110_HARBOR_REPAIR" not in text assert "AWOOOI_ALLOW_NON_110_HARBOR_REPAIR" in text assert "EXPECTED_HOST_IP" in text assert "192.168.0.110" in text assert "REFUSE harbor repair" in text assert "require_expected_host_for_apply || return 1" in text assert "while true" in text def test_deploy_to_110_syncs_local_control_path_recovery_helpers() -> None: text = (ROOT / "scripts/reboot-recovery/deploy-to-110.sh").read_text( encoding="utf-8" ) assert "repair-110-ssh-publickey-auth-local.sh" in text assert "recover-110-control-path-and-harbor-local.sh" in text assert "check-awoooi-110-controlled-cd-lane-readiness.sh" in text assert "/usr/local/bin/repair-110-ssh-publickey-auth-local.sh" in text assert "/usr/local/bin/recover-110-control-path-and-harbor-local.sh" in text assert "/usr/local/bin/check-awoooi-110-controlled-cd-lane-readiness.sh" in text def test_reboot_recovery_188_helpers_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "scripts/reboot-recovery/deploy-to-188.sh)", "scripts/reboot-recovery/awoooi-startup.sh)", "scripts/reboot-recovery/awoooi-startup.service)", "scripts/reboot-recovery/tests/test_188_host_hygiene_checklist.py)", "../../scripts/reboot-recovery/deploy-to-188.sh", "../../scripts/reboot-recovery/awoooi-startup.sh", "../../scripts/reboot-recovery/tests/test_188_host_hygiene_checklist.py", ] for source in expected_sources: assert source in text def test_onboarding_warning_step_template_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() assert "onboarding warning-step workflow is" in text assert ".gitea/workflows/awoooi-onboarding-warning-step.yaml)" in text assert ( "docs/operations/templates/awoooi-gitea-onboarding-warning-step.workflow.yaml)" in text ) def test_credential_escrow_intake_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() assert ( "docs/operations/awoooi-credential-escrow-evidence-controlled-closeout-receipt.snapshot.json)" in text ) assert "apps/api/src/services/credential_escrow_evidence_intake_readiness.py)" in text assert "src/services/credential_escrow_evidence_intake_readiness.py" in text assert "tests/test_credential_escrow_evidence_intake_readiness_api.py" in text def test_p0_onboarding_readiness_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "awoooi_gitea_onboarding_warning_step_dashboard.py", "awoooi_gitea_onboarding_warning_step_owner_package.py", "awoooi_gitea_onboarding_warning_step_owner_response_preflight.py", "awoooi_gitea_onboarding_warning_step_template_copy_apply_gate.py", "awoooi_gitea_onboarding_warning_step_template_copy_execution_plan.py", "awoooi_gitea_onboarding_warning_step_template_copy_receipt.py", "awoooi_gitea_onboarding_warning_step_runtime_enablement_gate.py", "awoooi_new_product_onboarding_page_model.py", "awoooi_onboarding_reminder_contract.py", "awoooi_onboarding_source_contracts.py", "awoooi_product_onboarding_guard.py", ] for source in expected_sources: assert f"apps/api/src/services/{source})" in text assert f"src/services/{source}" in text assert ".gitea/workflows/awoooi-onboarding-warning-step.yaml)" in text assert "docs/operations/templates/awoooi-gitea-onboarding-warning-step.workflow.yaml)" in text assert ( "docs/operations/awoooi-gitea-onboarding-warning-step-template-copy-receipt.snapshot.json)" in text ) assert "tests/test_p0_cicd_baseline_source_readiness_api.py" in text def test_priority_work_order_readback_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "docs/operations/awoooi-priority-work-order-readback.snapshot.json)", "apps/api/src/services/awoooi_priority_work_order_readback.py)", "apps/api/tests/test_awoooi_priority_work_order_readback_api.py)", "src/services/awoooi_priority_work_order_readback.py", "tests/test_awoooi_priority_work_order_readback_api.py", ] for source in expected_sources: assert source in text def test_report_truth_actionability_review_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_report_truth_actionability_review.py)", "apps/api/tests/test_ai_agent_report_truth_actionability_review.py)", "apps/api/tests/test_ai_agent_report_truth_actionability_review_api.py)", ] for source in expected_sources: assert source in text def test_recovery_apply_receipts_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() assert ( "docs/operations/awooop-conversation-event-hot-path-index-apply-receipt-*.snapshot.json)" in text ) def test_iwooos_security_operation_api_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/api/v1/iwooos.py)", "apps/api/src/services/iwooos_security_operating_system.py)", "apps/api/tests/test_iwooos_security_operating_system.py)", "apps/api/tests/test_iwooos_wazuh_prod_manifest.py)", "src/api/v1/iwooos.py", "src/services/iwooos_security_operating_system.py", "tests/test_iwooos_security_operating_system.py", ] for source in expected_sources: assert source in text def test_navigation_coverage_guard_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() assert "scripts/dev/awoooi-navigation-coverage-guard.py)" in text def test_ai_autonomous_runtime_control_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_autonomous_runtime_control.py)", "apps/api/tests/test_ai_agent_autonomous_runtime_control.py)", "src/services/ai_agent_autonomous_runtime_control.py", "tests/test_ai_agent_autonomous_runtime_control.py", ] for source in expected_sources: assert source in text def test_ai_log_intelligence_readback_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "docs/operations/ai-agent-log-intelligence-runtime-sample-readback.snapshot.json)", "apps/api/src/services/ai_agent_log_intelligence_integration_readback.py)", "apps/api/tests/test_ai_agent_log_intelligence_integration_readback_api.py)", "src/services/ai_agent_log_intelligence_integration_readback.py", "tests/test_ai_agent_log_intelligence_integration_readback_api.py", ] for source in expected_sources: assert source in text def test_ai_log_feedback_receipt_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_log_feedback_receipt_dry_run.py)", "apps/api/tests/test_ai_agent_log_feedback_receipt_dry_run_api.py)", "src/services/ai_agent_log_feedback_receipt_dry_run.py", "tests/test_ai_agent_log_feedback_receipt_dry_run_api.py", ] for source in expected_sources: assert source in text def test_ai_log_post_write_verifier_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_log_post_write_verifier_dry_run.py)", "apps/api/tests/test_ai_agent_log_post_write_verifier_dry_run_api.py)", "src/services/ai_agent_log_post_write_verifier_dry_run.py", "tests/test_ai_agent_log_post_write_verifier_dry_run_api.py", ] for source in expected_sources: assert source in text def test_ai_log_controlled_writeback_plan_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py)", "apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py)", "src/services/ai_agent_log_controlled_writeback_plan_readback.py", "tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py", ] for source in expected_sources: assert source in text def test_ai_log_controlled_writeback_executor_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py)", "apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py)", "src/services/ai_agent_log_controlled_writeback_executor_readback.py", "tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py", ] for source in expected_sources: assert source in text def test_ai_log_controlled_writeback_dispatch_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_log_controlled_writeback_dispatch.py)", "apps/api/tests/test_ai_agent_log_controlled_writeback_dispatch_api.py)", "apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type.sql)", "apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql)", "src/services/ai_agent_log_controlled_writeback_dispatch.py", "tests/test_ai_agent_log_controlled_writeback_dispatch_api.py", ] for source in expected_sources: assert source in text def test_ai_log_controlled_writeback_consumer_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py)", "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_apply.py)", "apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py)", "apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py)", "src/services/ai_agent_log_controlled_writeback_consumer_readback.py", "src/services/ai_agent_log_controlled_writeback_consumer_apply.py", "tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py", "tests/test_ai_agent_log_controlled_writeback_consumer_apply_api.py", ] for source in expected_sources: assert source in text def test_awooop_ansible_check_mode_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "apps/api/src/services/awooop_ansible_audit_service.py)", "apps/api/src/services/awooop_ansible_check_mode_service.py)", "apps/api/migrations/adr090e_ansible_learning_writeback_operation_type.sql)", "apps/api/migrations/adr090e_ansible_learning_writeback_operation_type_down.sql)", "src/services/awooop_ansible_audit_service.py", "apps/api/tests/test_awooop_truth_chain_service.py)", "src/services/awooop_ansible_check_mode_service.py", "tests/test_awooop_truth_chain_service.py", ] for source in expected_sources: assert source in text def test_dr_escrow_checklist_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "scripts/reboot-recovery/dr-escrow-evidence-checklist.py)", "scripts/reboot-recovery/post-reboot-owner-response-preflight.py)", "scripts/reboot-recovery/tests/test_dr_escrow_evidence_checklist.py)", "../../scripts/reboot-recovery/dr-escrow-evidence-checklist.py", "../../scripts/reboot-recovery/post-reboot-owner-response-preflight.py", "../../scripts/reboot-recovery/tests/test_dr_escrow_evidence_checklist.py", ] for source in expected_sources: assert source in text def test_gitea_private_inventory_scorecard_stays_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "docs/operations/awoooi-gitea-private-inventory-p0-scorecard.snapshot.json)", "docs/operations/awoooi-gitea-private-inventory-controlled-closeout-receipt.snapshot.json)", "apps/api/src/services/gitea_authenticated_inventory_payload_validation.py)", "apps/api/src/services/gitea_owner_coverage_attestation_validation.py)", "apps/api/src/services/gitea_private_inventory_closeout_validation.py)", "apps/api/src/services/gitea_private_inventory_p0_scorecard.py)", "apps/api/src/services/gitea_repo_bundle_backup_readback.py)", "apps/api/src/services/gitea_workflow_runner_owner_attestation_request.py)", "apps/api/tests/test_gitea_private_inventory_p0_scorecard_api.py)", "apps/api/tests/test_gitea_repo_bundle_backup_readback_api.py)", "apps/api/tests/test_gitea_workflow_runner_owner_attestation_request_api.py)", "docs/operations/awoooi-gitea-authenticated-inventory-payload-validation.snapshot.json)", "docs/security/GITEA-REPO-INVENTORY-SNAPSHOT.md)", "docs/security/gitea-repo-inventory.snapshot.json)", "apps/api/src/services/gitea_authenticated_inventory_payload_validation.py)", "scripts/security/gitea-private-inventory-p0-scorecard.py)", "scripts/security/gitea-authenticated-inventory-payload-validator.py)", "scripts/security/tests/test_gitea_private_inventory_p0_scorecard.py)", "src/services/gitea_authenticated_inventory_payload_validation.py", "src/services/gitea_owner_coverage_attestation_validation.py", "src/services/gitea_private_inventory_closeout_validation.py", "src/services/gitea_private_inventory_p0_scorecard.py", "src/services/gitea_repo_bundle_backup_readback.py", "src/services/gitea_workflow_runner_owner_attestation_request.py", "tests/test_gitea_private_inventory_p0_scorecard_api.py", "tests/test_gitea_repo_bundle_backup_readback_api.py", "tests/test_gitea_workflow_runner_owner_attestation_request_api.py", "scripts/security/tests/test_gitea_authenticated_inventory_payload_validator.py)", "../../scripts/security/gitea-private-inventory-p0-scorecard.py", "../../scripts/security/gitea-authenticated-inventory-payload-validator.py", "../../scripts/security/tests/test_gitea_private_inventory_p0_scorecard.py", "../../scripts/security/tests/test_gitea_authenticated_inventory_payload_validator.py", ] for source in expected_sources: assert source in text def test_reboot_auto_recovery_slo_sources_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "docs/operations/awoooi-reboot-auto-recovery-slo-scorecard.snapshot.json)", "apps/api/src/services/reboot_auto_recovery_slo_scorecard.py)", "apps/api/src/services/reboot_auto_recovery_drill_preflight.py)", "apps/api/src/services/stockplatform_public_api_runtime_readback.py)", "apps/api/src/services/stockplatform_public_api_controlled_recovery_preflight.py)", "apps/api/src/services/harbor_registry_controlled_recovery_preflight.py)", "apps/api/src/services/harbor_registry_controlled_recovery_receipt.py)", "apps/api/tests/test_reboot_auto_recovery_slo_scorecard_api.py)", "apps/api/tests/test_stockplatform_public_api_runtime_readback.py)", "apps/api/tests/test_stockplatform_public_api_controlled_recovery_preflight.py)", "apps/api/tests/test_harbor_registry_controlled_recovery_preflight.py)", "apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py)", "src/services/reboot_auto_recovery_slo_scorecard.py", "src/services/reboot_auto_recovery_drill_preflight.py", "src/services/stockplatform_public_api_runtime_readback.py", "src/services/stockplatform_public_api_controlled_recovery_preflight.py", "src/services/harbor_registry_controlled_recovery_preflight.py", "src/services/harbor_registry_controlled_recovery_receipt.py", "tests/test_reboot_auto_recovery_slo_scorecard_api.py", "tests/test_stockplatform_public_api_runtime_readback.py", "tests/test_stockplatform_public_api_controlled_recovery_preflight.py", "tests/test_harbor_registry_controlled_recovery_preflight.py", "tests/test_harbor_registry_controlled_recovery_receipt.py", "scripts/reboot-recovery/awoooi-reboot-auto-recovery-slo.service)", "scripts/reboot-recovery/awoooi-reboot-auto-recovery-slo.timer)", "scripts/reboot-recovery/install-reboot-auto-recovery-slo-110.sh)", "scripts/reboot-recovery/reboot-auto-recovery-host-probe.sh)", "scripts/reboot-recovery/reboot-auto-recovery-slo-exporter.sh)", "scripts/reboot-recovery/reboot-auto-recovery-slo-scorecard.py)", "scripts/reboot-recovery/collect-windows99-vmware-verify.sh)", "scripts/reboot-recovery/windows99-vmware-autostart.ps1)", "scripts/reboot-recovery/windows99-management-channel-probe.py)", "scripts/reboot-recovery/full-stack-cold-start-check.sh)", "scripts/reboot-recovery/full-stack-recovery-scorecard.sh)", "scripts/reboot-recovery/harbor-watchdog.sh)", "scripts/reboot-recovery/awoooi-startup-110.sh)", "scripts/reboot-recovery/diagnose-110-ssh-publickey-auth.sh)", "scripts/reboot-recovery/repair-110-ssh-publickey-auth-local.sh)", "scripts/reboot-recovery/verify-cold-start-monitor-deploy.sh)", "scripts/reboot-recovery/tests/test_cold_start_monitor_bounded_probes.py)", "scripts/reboot-recovery/tests/test_reboot_auto_recovery_slo_installer.py)", "scripts/reboot-recovery/tests/test_reboot_auto_recovery_slo_scorecard.py)", "scripts/reboot-recovery/tests/test_windows99_vmware_verify_collector.py)", "scripts/reboot-recovery/tests/test_harbor_watchdog_contract.py)", "../../scripts/reboot-recovery/reboot-auto-recovery-slo-scorecard.py", "../../scripts/reboot-recovery/collect-windows99-vmware-verify.sh", "../../scripts/reboot-recovery/windows99-management-channel-probe.py", "../../scripts/reboot-recovery/tests/test_cold_start_monitor_bounded_probes.py", "../../scripts/reboot-recovery/tests/test_reboot_auto_recovery_slo_installer.py", "../../scripts/reboot-recovery/tests/test_reboot_auto_recovery_slo_scorecard.py", "../../scripts/reboot-recovery/tests/test_windows99_vmware_verify_collector.py", ] for source in expected_sources: assert source in text def test_post_start_recovery_verifiers_stay_on_controlled_runtime_profile() -> None: text = _workflow_text() expected_sources = [ "docs/runbooks/REBOOT-POST-START-QUICK-CHECK.md)", "docs/runbooks/REBOOT-RECOVERY-SOP.md)", "docs/runbooks/FULL-STACK-COLD-START-SOP.md)", "docs/runbooks/HOST-RUNAWAY-PROCESS-AIOPS-PLAYBOOK.md)", "docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json)", "docs/operations/post-reboot-runtime-recovery-readback-2026-07-01.snapshot.json)", "docs/workplans/2026-06-04-reboot-cold-start-backup-recovery-workplan.md)", "docs/workplans/2026-07-02-commander-inserted-requirements-priority-ledger.md)", "apps/api/src/api/v1/platform/operator_runs.py)", "ops/monitoring/alerts-unified.yml)", "ops/monitoring/alerts.yml)", "ops/alertmanager/alertmanager.yml)", "ops/reboot-recovery/full-stack-cold-start-baseline.yml)", "ops/runner/awoooi-cd-lane-drain.service)", "ops/runner/README.md)", "ops/runner/check-awoooi-110-controlled-cd-lane-readiness.sh)", "ops/runner/register-awoooi-110-controlled-cd-lane-drain.sh)", "ops/runner/test_check_awoooi_110_controlled_cd_lane_readiness.py)", "ops/runner/test_register_awoooi_110_controlled_cd_lane_drain.py)", "scripts/backup/gitea-repo-bundle-backup.sh)", "scripts/ops/backup-health-textfile-exporter.py)", "scripts/ops/docker-disk-pressure-retention-cleanup.py)", "scripts/ops/gitea-queue-hook-backlog-playbook.py)", "scripts/ops/host-runaway-process-exporter.py)", "scripts/ops/host-sustained-load-evidence.py)", "scripts/ops/deploy-alertmanager-config.sh)", "scripts/ops/tests/test_gitea_queue_hook_backlog_playbook.py)", "src/api/v1/platform/operator_runs.py", "scripts/ops/tests/test_alertmanager_webhook_config.py)", "scripts/reboot-recovery/deploy-to-110.sh)", "scripts/reboot-recovery/enforce-110-runner-failclosed.sh)", "scripts/reboot-recovery/recover-110-control-path-and-harbor-local.sh)", "scripts/reboot-recovery/apply-credential-escrow-closeout-receipt-to-110.sh)", "scripts/reboot-recovery/post-start-quick-check.sh)", "scripts/reboot-recovery/reboot-recovery-readiness-audit.sh)", "scripts/reboot-recovery/188-host-hygiene-maintenance-checklist.sh)", "scripts/reboot-recovery/full-stack-cold-start-check.sh)", "scripts/reboot-recovery/cold-start-textfile-exporter.sh)", "scripts/reboot-recovery/install-cold-start-monitor-110.sh)", "scripts/reboot-recovery/momo-drive-token-source-recovery-preflight.sh)", "scripts/reboot-recovery/momo-source-arrival-gate.py)", "scripts/reboot-recovery/full-stack-recovery-scorecard.sh)", "scripts/reboot-recovery/awoooi-startup-110.sh)", "scripts/reboot-recovery/harbor-watchdog.sh)", "scripts/reboot-recovery/windows99-vmware-autostart.ps1)", "scripts/reboot-recovery/windows99-management-channel-probe.py)", "scripts/reboot-recovery/apply-credential-escrow-closeout-receipt-to-110.sh)", "scripts/reboot-recovery/verify-cold-start-monitor-deploy.sh)", "scripts/reboot-recovery/tests/test_188_host_hygiene_checklist.py)", "scripts/reboot-recovery/tests/test_post_start_quick_check_contract.py)", "scripts/reboot-recovery/tests/test_cold_start_monitor_bounded_probes.py)", "scripts/reboot-recovery/tests/test_reboot_p0_operational_contract.py)", "scripts/reboot-recovery/tests/test_harbor_watchdog_contract.py)", "scripts/reboot-recovery/tests/test_recover_110_control_path_and_harbor_local.py)", "scripts/reboot-recovery/tests/test_momo_source_arrival_gate.py)", "../../scripts/ops/backup-health-textfile-exporter.py", "../../scripts/ops/docker-disk-pressure-retention-cleanup.py", "../../scripts/ops/gitea-queue-hook-backlog-playbook.py", "../../scripts/ops/host-runaway-process-exporter.py", "../../scripts/ops/host-sustained-load-controller.py", "../../scripts/ops/host-sustained-load-evidence.py", "../../scripts/ops/deploy-alertmanager-config.sh", "../../scripts/backup/gitea-repo-bundle-backup.sh", "../../ops/monitoring/alerts-unified.yml", "../../ops/monitoring/alerts.yml", "../../ops/alertmanager/alertmanager.yml", "../../ops/reboot-recovery/full-stack-cold-start-baseline.yml", "../../ops/runner/check-awoooi-110-controlled-cd-lane-readiness.sh", "../../ops/runner/register-awoooi-110-controlled-cd-lane-drain.sh", "../../scripts/reboot-recovery/deploy-to-110.sh", "../../scripts/reboot-recovery/enforce-110-runner-failclosed.sh", "../../scripts/reboot-recovery/recover-110-control-path-and-harbor-local.sh", "../../scripts/reboot-recovery/apply-credential-escrow-closeout-receipt-to-110.sh", "../../scripts/reboot-recovery/post-start-quick-check.sh", "../../scripts/reboot-recovery/188-host-hygiene-maintenance-checklist.sh", "../../scripts/reboot-recovery/full-stack-cold-start-check.sh", "../../scripts/reboot-recovery/cold-start-textfile-exporter.sh", "../../scripts/reboot-recovery/install-cold-start-monitor-110.sh", "../../scripts/reboot-recovery/momo-drive-token-source-recovery-preflight.sh", "../../scripts/reboot-recovery/momo-source-arrival-gate.py", "../../scripts/reboot-recovery/full-stack-recovery-scorecard.sh", "../../scripts/reboot-recovery/harbor-watchdog.sh", "../../scripts/reboot-recovery/apply-credential-escrow-closeout-receipt-to-110.sh", "../../scripts/reboot-recovery/awoooi-startup-110.sh", "../../scripts/reboot-recovery/diagnose-110-ssh-publickey-auth.sh", "../../scripts/reboot-recovery/repair-110-ssh-publickey-auth-local.sh", "../../scripts/reboot-recovery/verify-cold-start-monitor-deploy.sh", "../../scripts/reboot-recovery/tests/test_188_host_hygiene_checklist.py", "../../scripts/ops/tests/test_docker_disk_pressure_retention_cleanup.py", "../../scripts/ops/tests/test_host_runaway_process_exporter.py", "../../scripts/ops/tests/test_alertmanager_webhook_config.py", "../../scripts/reboot-recovery/tests/test_post_start_quick_check_contract.py", "../../scripts/reboot-recovery/tests/test_cold_start_monitor_bounded_probes.py", "../../scripts/reboot-recovery/tests/test_reboot_p0_operational_contract.py", "../../scripts/reboot-recovery/tests/test_harbor_watchdog_contract.py", "../../scripts/reboot-recovery/tests/test_recover_110_control_path_and_harbor_local.py", "../../scripts/reboot-recovery/tests/test_momo_source_arrival_gate.py", "../../ops/runner/test_check_awoooi_110_controlled_cd_lane_readiness.py", "../../ops/runner/test_register_awoooi_110_controlled_cd_lane_drain.py", ] for source in expected_sources: assert source in text def test_controlled_runtime_skips_b5_before_docker_socket_use() -> None: text = _workflow_text() b5_start = text.index("- name: Integration Tests (B5") docker_socket = text.index("-v /var/run/docker.sock:/var/run/docker.sock", b5_start) persisted_profile_write = text.index('> .awoooi-cd-test-profile') persisted_profile_read = text.index( 'tr -d \'\\r\\n\' < .awoooi-cd-test-profile', b5_start, ) controlled_gate = text.index( 'if [ "${AWOOOI_CD_TEST_PROFILE:-full}" = "controlled-runtime" ]; then', b5_start, ) exit_zero = text.index("exit 0", controlled_gate) assert persisted_profile_write < b5_start assert b5_start < persisted_profile_read < controlled_gate assert controlled_gate < exit_zero < docker_socket def test_metadata_only_marker_skips_deploy_jobs_after_tests() -> None: text = _workflow_text() tests_header = text.split("tests:", 1)[1].split("steps:", 1)[0] build_header = text.split("build-and-deploy:", 1)[1].split("steps:", 1)[0] post_deploy_header = text.split("post-deploy-checks:", 1)[1].split("steps:", 1)[0] metadata_gate = "contains(github.event.head_commit.message, '[metadata-only]')" assert metadata_gate not in tests_header assert metadata_gate in build_header assert metadata_gate in post_deploy_header assert "metadata-only controlled-runtime fixes already run the" in text def test_b5_full_profile_fails_fast_when_docker_socket_or_db_network_is_unready() -> None: text = _workflow_text() b5_block = text.split("- name: Integration Tests (B5", 1)[1] b5_block = b5_block.split("- name: Clean Test Workspace Artifacts", 1)[0] assert "set -euo pipefail" in b5_block assert "docker info >/dev/null 2>&1" in b5_block assert "BLOCKER b5_docker_socket_unavailable" in b5_block assert "B5_DB_READY=0" in b5_block assert "BLOCKER b5_pg_test_container_not_ready" in b5_block assert "--user 0:0" in b5_block assert b5_block.index("docker info >/dev/null 2>&1") < b5_block.index( "docker run -d --name pg-test-b5" ) assert b5_block.index("BLOCKER b5_pg_test_container_not_ready") < b5_block.index( "/opt/api-venv/bin/pytest tests/integration/test_b5_core_flows.py" ) def test_controlled_runtime_pytest_paths_exist() -> None: text = _workflow_text() block = text.split("PYTHONFAULTHANDLER=1 python3.11 -m pytest", 1)[1] block = block.split("-v --tb=short", 1)[0] path_tokens = sorted(set(re.findall( r"((?:\.\./\.\./)?(?:tests|ops|scripts)/[A-Za-z0-9_./-]+\.py)", block, ))) missing: list[str] = [] for token in path_tokens: if token.startswith("tests/"): path = ROOT / "apps/api" / token else: path = ROOT / token.removeprefix("../../") if not path.exists(): missing.append(token) assert missing == []