import json
import os
import subprocess
from pathlib import Path

import yaml

REPO_ROOT = Path(__file__).resolve().parents[1]
PLAN = REPO_ROOT / "ops" / "plan_gcp_iap_operator_access.py"
WORKFLOW = REPO_ROOT / ".github" / "workflows" / "gcp-iap-operator.yml"
OPERATOR = REPO_ROOT / "scripts" / "gcp_iap_operator.sh"


def run_plan(*args: str) -> str:
    return subprocess.check_output(["python3", str(PLAN), *args], cwd=REPO_ROOT, text=True)


def test_plan_is_emit_only_and_pins_exact_github_identity() -> None:
    plan = json.loads(run_plan())

    assert plan["mode"] == "dry_run_emit_only"
    assert plan["executes_commands"] is False
    condition = plan["wif"]["provider_condition"]
    assert "assertion.repository == 'living-ip/teleo-infrastructure'" in condition
    assert "assertion.ref == 'refs/heads/main'" in condition
    assert (
        "assertion.workflow_ref == "
        "'living-ip/teleo-infrastructure/.github/workflows/gcp-iap-operator.yml@refs/heads/main'"
    ) in condition
    assert "assertion.event_name == 'workflow_dispatch'" in condition
    assert plan["wif"]["long_lived_google_credentials"] is False


def test_plan_uses_exact_iap_firewall_and_least_privilege_split() -> None:
    plan = json.loads(run_plan())
    commands = "\n".join(plan["bootstrap_commands"])

    assert plan["target"]["internal_ip"] == "10.60.0.3"
    assert plan["iap"]["condition"] == 'destination.ip == "10.60.0.3" && destination.port == 22'
    assert plan["iap"]["firewall"] == {
        "protocol_ports": "tcp:22",
        "rule": "teleo-prod-allow-ssh-iap",
        "source_range": "35.235.240.0/20",
        "target_tag": "teleo-prod-ssh",
    }
    assert plan["gcloud_ssh_custom_role"]["permissions"] == [
        "compute.instances.get",
        "compute.instances.list",
        "compute.projects.get",
    ]
    assert plan["identities"]["status"]["os_login_role"] == "roles/compute.osLogin"
    assert plan["identities"]["status"]["sudo"] is False
    assert plan["identities"]["clone_operator"]["os_login_role"] == "roles/compute.osAdminLogin"
    assert plan["identities"]["clone_operator"]["operations"] == [
        "direct-claim-replay",
        "cleanup-clone",
    ]
    assert "roles/iam.serviceAccountUser" in commands
    assert "--ssh-key-expire-after=5m" in commands
    assert "--ssh-key-expiration" not in commands
    assert "sha256sum -c dispatcher.sha256" in commands
    assert "teleo-gcp-iap-operator-v1" in commands
    for role in plan["explicitly_absent"]["roles"]:
        assert f"--role {role}" not in commands
    assert "compute.instances.setMetadata" not in plan["gcloud_ssh_custom_role"]["permissions"]


def test_public_rule_is_disabled_only_after_workflow_status_verification() -> None:
    plan = json.loads(run_plan())
    verification = plan["post_bootstrap_verification"]
    joined = "\n".join(verification)

    dispatch_index = next(i for i, command in enumerate(verification) if "gh workflow run" in command)
    disable_index = next(
        i
        for i, command in enumerate(verification)
        if "teleo-prod-allow-ssh-current-ip" in command and "--disabled" in command
    )
    assert dispatch_index < disable_index
    assert "operation=status" in joined
    assert "target_db=teleo_clone_status" in joined
    assert "dispatcher_sha256" in joined
    assert 'Path("scripts/gcp_iap_operator.sh")' in joined
    assert plan["public_ingress_cutover"]["automatic_fallback_to_public_ssh"] is False
    assert "DATA_READ for iap.googleapis.com" in plan["audit_logging"]["guidance"]
    assert plan["revocation_commands"]


def test_shell_plan_is_clearly_dry_run() -> None:
    shell = run_plan("--format", "shell")

    assert shell.startswith("# DRY RUN:")
    assert "# bootstrap_commands" in shell
    assert "# post_bootstrap_verification" in shell
    assert "# revocation_commands" in shell
    assert "subprocess" not in PLAN.read_text(encoding="utf-8")


def load_workflow() -> dict[str, object]:
    return yaml.load(WORKFLOW.read_text(encoding="utf-8"), Loader=yaml.BaseLoader)


def test_workflow_accepts_only_fixed_operation_and_bounded_ids() -> None:
    workflow = load_workflow()
    inputs = workflow["on"]["workflow_dispatch"]["inputs"]

    assert set(inputs) == {"operation", "request_id", "target_db"}
    assert inputs["operation"]["options"] == ["status", "direct-claim-replay", "cleanup-clone"]
    assert inputs["target_db"]["default"] == "teleo_clone_status"
    source = WORKFLOW.read_text(encoding="utf-8")
    assert '[[ "${GITHUB_REF}" == "refs/heads/main" ]]' in source
    assert '[[ "${GITHUB_WORKFLOW_REF}" == "${EXPECTED_WORKFLOW_REF}" ]]' in source
    assert "request_id_re='^iap-[a-z0-9]{12,32}$'" in source
    assert "target_db_re='^teleo_clone_[a-z0-9][a-z0-9_]{0,50}$'" in source
    for forbidden_input in ("command:", "path:", "host:", "project:", "zone:", "ip:", "ref:", "service_account:"):
        assert forbidden_input not in inputs


def test_workflow_uses_wif_official_actions_and_sanitized_short_retention() -> None:
    workflow = load_workflow()
    assert workflow["permissions"] == {"contents": "read", "id-token": "write"}
    steps = workflow["jobs"]["operate"]["steps"]
    actions = [step["uses"] for step in steps if "uses" in step]

    assert actions == [
        "actions/checkout@v4",
        "google-github-actions/auth@v3",
        "google-github-actions/setup-gcloud@v3",
        "actions/upload-artifact@v4",
    ]
    auth = next(step for step in steps if step.get("uses") == "google-github-actions/auth@v3")
    assert auth["with"]["create_credentials_file"] == "true"
    assert auth["with"]["cleanup_credentials"] == "true"
    upload = next(step for step in steps if step.get("uses") == "actions/upload-artifact@v4")
    assert upload["with"]["path"] == "${{ env.RESULT_DIR }}/result.json"
    assert upload["with"]["retention-days"] == "7"
    validate_index = next(i for i, step in enumerate(steps) if step.get("id") == "validate")
    auth_index = next(i for i, step in enumerate(steps) if step.get("uses") == "google-github-actions/auth@v3")
    assert validate_index < auth_index


def test_workflow_initializes_runner_paths_at_step_runtime() -> None:
    workflow = load_workflow()
    assert "BUNDLE_DIR" not in workflow["env"]
    assert "RESULT_DIR" not in workflow["env"]
    steps = workflow["jobs"]["operate"]["steps"]
    paths_index = next(i for i, step in enumerate(steps) if step.get("id") == "paths")
    validate_index = next(i for i, step in enumerate(steps) if step.get("id") == "validate")
    paths_source = steps[paths_index]["run"]
    assert paths_index < validate_index
    assert '"${RUNNER_TEMP}/gcp-iap-operator-bundle" >> "${GITHUB_ENV}"' in paths_source
    assert '"${RUNNER_TEMP}/gcp-iap-operator-result" >> "${GITHUB_ENV}"' in paths_source
    assert "runner.temp" not in WORKFLOW.read_text(encoding="utf-8")


def test_workflow_retains_sanitized_failure_receipt_before_upload() -> None:
    workflow = load_workflow()
    steps = workflow["jobs"]["operate"]["steps"]
    receipt_index = next(i for i, step in enumerate(steps) if step.get("name") == "Ensure sanitized result receipt")
    upload_index = next(i for i, step in enumerate(steps) if step.get("uses") == "actions/upload-artifact@v4")
    receipt = steps[receipt_index]
    source = receipt["run"]
    assert receipt_index < upload_index
    assert receipt["if"] == "always()"
    assert receipt["env"]["AUTH_OUTCOME"] == "${{ steps.auth.outcome }}"
    assert receipt["env"]["OPERATE_OUTCOME"] == "${{ steps.operate.outcome }}"
    assert "workflow_stopped_before_valid_operator_result" in source
    assert '"credential_values_logged": False' in source
    assert '"raw_stdout_retained": False' in source
    assert '"raw_stderr_retained": False' in source


def test_workflow_builds_exact_main_bundle_from_tracked_files_and_hashes() -> None:
    source = WORKFLOW.read_text(encoding="utf-8")

    expected_files = [
        "scripts/gcp_iap_operator.sh",
        "scripts/run_gcp_generated_db_direct_claim_suite.py",
        "scripts/run_leo_clone_bound_handler_checkpoint.py",
        "scripts/working_leo_open_ended_benchmark.py",
        "hermes-agent/leoclean-bin/cloudsql_memory_tool.py",
        "ops/postgres_parity_manifest.sql",
    ]
    for relative in expected_files:
        assert f'"{relative}"' in source
    assert 'f"docs/reports/leo-working-state-20260709/{target_db}-canonical-parity-receipt.json"' in source
    assert '["git", "ls-files", "--error-unmatch", "--", relative]' in source
    assert 'if head != os.environ["GITHUB_SHA"]' in source
    assert "hashlib.sha256(data).hexdigest()" in source
    assert '"schema": "livingip.gcpIapOperatorBundle.v1"' in source
    assert 'bundle_path = bundle_dir / f"{request_id}.tar.gz"' in source


def test_operator_allowlists_commands_and_has_no_live_mutation_surface() -> None:
    source = OPERATOR.read_text(encoding="utf-8")

    assert "status | direct-claim-replay | cleanup-clone" in source
    assert "scripts/run_gcp_generated_db_direct_claim_suite.py" in source
    assert "six_replies_returned" in source
    assert "posted_to_telegram" in source
    assert "live_service_unchanged" in source
    assert "live_profile_unchanged" in source
    assert "%s-canonical-parity-receipt.json" in source
    assert "/opt/teleo-eval" not in source
    assert "systemctl restart" not in source
    assert "systemctl stop" not in source
    assert "systemctl start" not in source
    assert "curl " not in source
    assert "--profile" not in source
    assert "eval " not in source


def test_operator_cleanup_is_bound_to_marker_clone_prefix_and_exact_run_dir() -> None:
    source = OPERATOR.read_text(encoding="utf-8")

    assert "TARGET_DB_RE='^teleo_clone_" in source
    assert 'expected="$REMOTE_RUN_ROOT/$request_id"' in source
    assert '[[ "$resolved" == "$expected" ]]' in source
    assert ".run-owner" in source
    assert "run ownership marker does not match request_id and target_db" in source
    assert 'drop database :"target_db";' in source
    assert '[[ "$remaining" == "0" ]]' in source
    assert 'rm -rf --one-file-system -- "$run_dir"' in source


def test_operator_uses_five_minute_key_private_modes_and_cleanup_trap() -> None:
    source = OPERATOR.read_text(encoding="utf-8")

    assert '--ssh-key-file="$ssh_key"' in source
    assert "--ssh-key-expire-after=5m" in source
    assert "--ssh-key-expiration" not in source
    assert "chmod 0700" in source
    assert "chmod 0600" in source
    assert "trap cleanup_runner_temp EXIT" in source
    assert "--tunnel-through-iap" in source
    assert "gcloud compute scp" in source
    assert "--verbosity=debug" not in source


def test_dispatcher_validates_exact_members_hashes_context_and_self_hash() -> None:
    source = OPERATOR.read_text(encoding="utf-8")

    assert 'expected_members = sorted([*expected, "bundle-manifest.json"])' in source
    assert "bundle member allowlist mismatch" in source
    assert "bundle manifest file allowlist mismatch" in source
    assert 'receipt.get("sha256") != hashlib.sha256(data).hexdigest()' in source
    assert '"workflow_ref": expected_workflow_ref' in source
    assert "installed dispatcher does not match the reviewed main bundle" in source
    assert 'dispatcher_sha != files["scripts/gcp_iap_operator.sh"]["sha256"]' in source


def test_runner_fake_gcloud_smoke_retains_only_sanitized_result(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    runner_temp = tmp_path / "runner"
    result_dir = runner_temp / "result"
    invocation = tmp_path / "gcloud-argv.txt"
    bin_dir.mkdir()
    runner_temp.mkdir()
    fake_gcloud = bin_dir / "gcloud"
    fake_gcloud.write_text(
        "#!/usr/bin/env bash\n"
        'printf \'%s\\n\' "$@" > "${FAKE_GCLOUD_ARGV}"\n'
        "printf '%s\\n' "
        '\'{"operation":"status","request_id":"iap-abcdefghijkl",\''
        '\'"target_db":"teleo_clone_status","status":"pass",\''
        '\'"instance":"teleo-prod-1","expected_internal_ip":"10.60.0.3",\''
        '\'"active_state":"active","sub_state":"running","nrestarts":0,\''
        '\'"dispatcher_version":"teleo-gcp-iap-operator-v1",\''
        '\'"dispatcher_sha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}\'\n',
        encoding="utf-8",
    )
    fake_gcloud.chmod(0o755)
    env = {
        **os.environ,
        "PATH": f"{bin_dir}:{os.environ['PATH']}",
        "RUNNER_TEMP": str(runner_temp),
        "RESULT_DIR": str(result_dir),
        "FAKE_GCLOUD_ARGV": str(invocation),
    }

    completed = subprocess.run(
        ["bash", str(OPERATOR), "status", "iap-abcdefghijkl", "teleo_clone_status"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
        check=False,
    )

    assert completed.returncode == 0, completed.stderr
    result = json.loads((result_dir / "result.json").read_text(encoding="utf-8"))
    assert result["status"] == "pass"
    assert result["credential_values_logged"] is False
    assert result["raw_stdout_retained"] is False
    assert result["raw_stderr_retained"] is False
    assert result["remote_result"]["active_state"] == "active"
    argv = invocation.read_text(encoding="utf-8")
    assert "compute\nssh\nteleo-prod-1\n" in argv
    assert "--tunnel-through-iap" in argv
    assert "--ssh-key-expire-after=5m" in argv
    assert not list(runner_temp.glob("teleo-iap-operator.*"))


def test_runner_fails_closed_when_remote_output_is_not_valid_json(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    runner_temp = tmp_path / "runner"
    result_dir = runner_temp / "result"
    bin_dir.mkdir()
    runner_temp.mkdir()
    fake_gcloud = bin_dir / "gcloud"
    fake_gcloud.write_text("#!/usr/bin/env bash\nprintf 'not-json\\n'\n", encoding="utf-8")
    fake_gcloud.chmod(0o755)
    env = {
        **os.environ,
        "PATH": f"{bin_dir}:{os.environ['PATH']}",
        "RUNNER_TEMP": str(runner_temp),
        "RESULT_DIR": str(result_dir),
    }

    completed = subprocess.run(
        ["bash", str(OPERATOR), "status", "iap-abcdefghijkl", "teleo_clone_status"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
        check=False,
    )

    assert completed.returncode == 90
    result = json.loads((result_dir / "result.json").read_text(encoding="utf-8"))
    assert result["status"] == "fail"
    assert result["remote_result"] == {}
    assert not list(runner_temp.glob("teleo-iap-operator.*"))


def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    runner_temp = tmp_path / "runner"
    result_dir = runner_temp / "result"
    bundle_dir = runner_temp / "gcp-iap-operator-bundle"
    invocation = tmp_path / "gcloud-argv.txt"
    bin_dir.mkdir()
    runner_temp.mkdir()
    bundle_dir.mkdir(mode=0o700)
    bundle = bundle_dir / "iap-abcdefghijkl.tar.gz"
    bundle.write_bytes(b"reviewed-bundle")
    bundle.chmod(0o600)
    fake_gcloud = bin_dir / "gcloud"
    fake_gcloud.write_text(
        "#!/usr/bin/env bash\n"
        'printf \'%s \' "$@" >> "${FAKE_GCLOUD_ARGV}"\n'
        "printf '\\n' >> \"${FAKE_GCLOUD_ARGV}\"\n"
        'if [[ "$2" == "ssh" ]]; then\n'
        "  printf '%s\\n' "
        '\'{"operation":"direct-claim-replay","request_id":"iap-abcdefghijkl",\''
        '\'"target_db":"teleo_clone_test","status":"pass",\''
        '\'"bundle_commit":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",\''
        '\'"bundle_sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}\'\n'
        "fi\n",
        encoding="utf-8",
    )
    fake_gcloud.chmod(0o755)
    env = {
        **os.environ,
        "PATH": f"{bin_dir}:{os.environ['PATH']}",
        "RUNNER_TEMP": str(runner_temp),
        "RESULT_DIR": str(result_dir),
        "FAKE_GCLOUD_ARGV": str(invocation),
    }

    completed = subprocess.run(
        ["bash", str(OPERATOR), "direct-claim-replay", "iap-abcdefghijkl", "teleo_clone_test"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
        check=False,
    )

    assert completed.returncode == 0, completed.stderr
    calls = invocation.read_text(encoding="utf-8").splitlines()
    assert len(calls) == 2
    assert "compute scp" in calls[0]
    assert str(bundle) in calls[0]
    assert "teleo-prod-1:.teleo-iap-upload-iap-abcdefghijkl.tar.gz" in calls[0]
    assert "--tunnel-through-iap" in calls[0]
    assert "--ssh-key-expire-after=5m" in calls[0]
    assert "compute ssh" in calls[1]
    assert "--tunnel-through-iap" in calls[1]


def test_clone_runner_rejects_bundle_symlink_before_gcloud(tmp_path: Path) -> None:
    bin_dir = tmp_path / "bin"
    runner_temp = tmp_path / "runner"
    result_dir = runner_temp / "result"
    bundle_dir = runner_temp / "gcp-iap-operator-bundle"
    outside = tmp_path / "outside.tar.gz"
    marker = tmp_path / "gcloud-called"
    bin_dir.mkdir()
    runner_temp.mkdir()
    bundle_dir.mkdir(mode=0o700)
    outside.write_bytes(b"outside")
    (bundle_dir / "iap-abcdefghijkl.tar.gz").symlink_to(outside)
    fake_gcloud = bin_dir / "gcloud"
    fake_gcloud.write_text(
        '#!/usr/bin/env bash\ntouch "${FAKE_GCLOUD_MARKER}"\n',
        encoding="utf-8",
    )
    fake_gcloud.chmod(0o755)
    env = {
        **os.environ,
        "PATH": f"{bin_dir}:{os.environ['PATH']}",
        "RUNNER_TEMP": str(runner_temp),
        "RESULT_DIR": str(result_dir),
        "FAKE_GCLOUD_MARKER": str(marker),
    }

    completed = subprocess.run(
        ["bash", str(OPERATOR), "cleanup-clone", "iap-abcdefghijkl", "teleo_clone_test"],
        cwd=REPO_ROOT,
        env=env,
        text=True,
        capture_output=True,
        check=False,
    )

    assert completed.returncode == 2
    assert "fixed workflow bundle is absent or unsafe" in completed.stderr
    assert not marker.exists()


def test_google_auth_credential_file_pattern_is_ignored_exactly() -> None:
    ignore_lines = (REPO_ROOT / ".gitignore").read_text(encoding="utf-8").splitlines()

    assert "gha-creds-*.json" in ignore_lines