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_api_has_capacity_for_the_production_concurrency_verifier() -> None:
    container = _deployment_container(
        "k8s/awoooi-prod/06-deployment-api.yaml",
        "api",
    )

    assert container["resources"] == {
        "requests": {"cpu": "500m", "memory": "512Mi"},
        "limits": {"cpu": "2", "memory": "1Gi"},
    }


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