"""CI 等待（scripts/ci_wait.py）：临时 Git 仓库 + 进程内 Gitea 合并状态桩 + 注入的时钟与 sleep。

全部离线运行：不联网、不读取真实 ~/.secrets（HOME 指向临时目录），token 为合成值。
"""
import contextlib
import http.server
import io
import json
import os
import socket
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest import mock

SOURCE = Path(__file__).resolve().parents[1]
SCRIPT = SOURCE / "scripts" / "ci_wait.py"
sys.path.insert(0, str(SOURCE / "scripts"))

import ci_wait  # noqa: E402

TOKEN = "synthetic-gitea-token-not-real"


def status(context, state, url=None):
    return {"context": context, "status": state, "target_url": url or f"https://ci.invalid/{context}"}


class FakeGitea:
    """Gitea 合并状态桩（127.0.0.1）：校验 token；每次 GET 依次返回脚本中的下一份 statuses（末项重复）。"""

    def __init__(self, token=TOKEN):
        self.token = token
        self.script = []
        self.requests = []
        stub = self

        class Handler(http.server.BaseHTTPRequestHandler):
            def log_message(self, *args):
                pass

            def _send(self, code, obj):
                body = json.dumps(obj).encode()
                self.send_response(code)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def do_GET(self):
                if self.headers.get("Authorization") != f"token {stub.token}":
                    return self._send(401, {"message": "unauthorized"})
                parts = self.path.split("?", 1)[0].split("/")
                if parts[1:4] != ["api", "v1", "repos"] or len(parts) != 9 or parts[6] != "commits" \
                        or parts[8] != "status":
                    return self._send(404, {"message": "not found"})
                stub.requests.append("/".join(parts[4:6]) + "@" + parts[7])
                statuses = stub.script.pop(0) if len(stub.script) > 1 else (stub.script or [[]])[0]
                return self._send(200, {"sha": parts[7], "statuses": statuses})

        self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.url = f"http://127.0.0.1:{self.server.server_address[1]}/api/v1"
        threading.Thread(target=self.server.serve_forever, daemon=True).start()

    def close(self):
        self.server.shutdown()
        self.server.server_close()


def closed_port_url():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    return f"http://127.0.0.1:{port}/api/v1"


class FakeClock:
    def __init__(self):
        self.now = 0.0
        self.sleeps = []

    def clock(self):
        return self.now

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.now += seconds


class CiWaitTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix="ci-wait-")
        self.addCleanup(self.tmp.cleanup)
        root = Path(self.tmp.name)
        self.repo = root / "repo"
        env = {k: v for k, v in os.environ.items()
               if not k.startswith("GIT_") and k not in {"GITEA_API_URL", "GITEA_REPO", "GITEA_TOKEN"}}
        env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_TERMINAL_PROMPT": "0",
                    "GIT_AUTHOR_NAME": "CI Wait Test", "GIT_AUTHOR_EMAIL": "ci-wait@example.invalid",
                    "GIT_COMMITTER_NAME": "CI Wait Test", "GIT_COMMITTER_EMAIL": "ci-wait@example.invalid",
                    "HOME": str(root / "no-home"),
                    "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost",
                    "GITEA_TOKEN": TOKEN})
        self.gitea = FakeGitea()
        self.addCleanup(self.gitea.close)
        env["GITEA_API_URL"] = self.gitea.url
        self.env = env
        env_patch = mock.patch.dict(os.environ, env, clear=True)
        env_patch.start()
        self.addCleanup(env_patch.stop)
        self.git(root, "init", "--initial-branch=main", str(self.repo))
        (self.repo / "README").write_text("fixture\n")
        self.git(self.repo, "add", "README")
        self.git(self.repo, "commit", "-m", "Create fixture")
        self.git(self.repo, "remote", "add", "origin", "ssh://git@git-ssh.example.invalid:31411/fixture/repo.git")
        self.sha = self.git(self.repo, "rev-parse", "HEAD")
        cwd = os.getcwd()
        os.chdir(self.repo)
        self.addCleanup(os.chdir, cwd)

    def git(self, cwd, *args):
        return subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, check=True).stdout.strip()

    def run_wait(self, *args, env=None):
        clock = FakeClock()
        out = io.StringIO()
        with mock.patch.dict(os.environ, env or {}), contextlib.redirect_stdout(out), \
                contextlib.redirect_stderr(out):
            code = ci_wait.main(list(args), sleep=clock.sleep, clock=clock.clock)
        text = out.getvalue()
        self.assertNotIn(TOKEN, text)
        return code, text, clock

    def test_success_requires_two_consecutive_all_green_polls(self):
        self.gitea.script = [[status("ci/test", "pending")], [status("ci/test", "pending")],
                             [status("ci/test", "success")], [status("ci/test", "success")]]
        code, out, clock = self.run_wait("--timeout", "300")
        self.assertEqual(code, 0, out)
        self.assertIn(f"[verdict] CI_SUCCESS sha={self.sha}", out)
        self.assertEqual(len(self.gitea.requests), 4, out)
        self.assertEqual(self.gitea.requests[0], f"fixture/repo@{self.sha}")
        self.assertEqual(clock.sleeps, [15.0, 15.0, 15.0], out)
        # 状态行只在变化时打印：pending 与 success 各一行。
        lines = [line for line in out.splitlines() if line.startswith("[ci-wait] sha=")]
        self.assertEqual(len(lines), 2, out)

    def test_late_registering_job_does_not_succeed_early(self):
        self.gitea.script = [[status("ci/test", "success")],
                             [status("ci/test", "success"), status("ci/build", "pending")],
                             [status("ci/test", "success"), status("ci/build", "success")],
                             [status("ci/test", "success"), status("ci/build", "success")]]
        code, out, _ = self.run_wait("--timeout", "300")
        self.assertEqual(code, 0, out)
        self.assertEqual(len(self.gitea.requests), 4, out)
        self.assertIn("ci/build=pending", out)

    def test_failure_exits_fast_with_target_url(self):
        url = "https://ci.invalid/runs/42"
        self.gitea.script = [[status("ci/test", "pending"), status("ci/build", "success")],
                             [status("ci/test", "failure", url), status("ci/build", "success")]]
        code, out, _ = self.run_wait("--sha", self.sha, "--timeout", "300")
        self.assertEqual(code, 1, out)
        self.assertEqual(len(self.gitea.requests), 2, out)
        self.assertIn(f"[ci-wait] FAIL ci/test {url}", out)
        self.assertNotIn("FAIL ci/build", out)
        self.assertIn(f"[verdict] CI_FAILED sha={self.sha}", out)

    def test_missing_required_context_waits_until_timeout(self):
        self.gitea.script = [[status("ci/test", "success")]]
        code, out, clock = self.run_wait("--contexts", "ci/test,e2e", "--interval", "5", "--timeout", "30")
        self.assertEqual(code, 2, out)
        self.assertEqual(clock.now, 30.0)
        self.assertIn("last observed: ci/test=success", out)
        self.assertIn("missing contexts: e2e", out)
        self.assertIn(f"[verdict] CI_TIMEOUT sha={self.sha}", out)

    def test_skipped_and_warning_contexts_count_as_done_not_failed(self):
        self.gitea.script = [[status("ci/test", "pending"), status("ci/test-pg", "skipped")],
                             [status("ci/test", "success"), status("ci/test-pg", "skipped"),
                              status("ci/svc-realcombo", "warning")]]
        code, out, _ = self.run_wait("--contexts", "test-pg", "--timeout", "300")
        self.assertEqual(code, 0, out)
        self.assertEqual(len(self.gitea.requests), 3, out)
        self.assertNotIn("FAIL", out)
        self.assertIn(f"[verdict] CI_SUCCESS sha={self.sha}", out)

    def test_skipped_does_not_mask_a_pending_context(self):
        self.gitea.script = [[status("ci/test", "pending"), status("ci/test-pg", "skipped")]]
        code, out, _ = self.run_wait("--interval", "5", "--timeout", "30")
        self.assertEqual(code, 2, out)
        self.assertIn(f"[verdict] CI_TIMEOUT sha={self.sha}", out)

    def test_context_names_match_as_substrings(self):
        self.gitea.script = [[status("CI / test (pull_request)", "success")]]
        code, out, _ = self.run_wait("--contexts", "test", "--timeout", "60")
        self.assertEqual(code, 0, out)

    def test_unreachable_api_exits_6_after_retrying_until_deadline(self):
        code, out, clock = self.run_wait("--timeout", "60", env={"GITEA_API_URL": closed_port_url()})
        self.assertEqual(code, 6, out)
        self.assertEqual(clock.now, 60.0)
        self.assertIn("[verdict] REMOTE_UNVERIFIED", out)

    def test_unauthorized_or_missing_token_exits_6_immediately(self):
        code, out, clock = self.run_wait("--timeout", "60", env={"GITEA_TOKEN": "wrong-token"})
        self.assertEqual(code, 6, out)
        self.assertEqual(clock.sleeps, [], out)
        self.assertIn("HTTP 401", out)
        with mock.patch.dict(os.environ):
            del os.environ["GITEA_TOKEN"]
            code, out, _ = self.run_wait("--timeout", "60")
        self.assertEqual(code, 6, out)
        self.assertIn("[verdict] REMOTE_UNVERIFIED", out)

    def test_sha_must_resolve_to_full_sha(self):
        code, out, _ = self.run_wait("--sha", "not-a-ref")
        self.assertEqual(code, 2, out)
        self.assertIn("full 40-hex SHA", out)
        self.assertEqual(self.gitea.requests, [])

    def test_cli_entry_point_reports_remote_unverified(self):
        env = self.env | {"GITEA_API_URL": closed_port_url()}
        result = subprocess.run([sys.executable, str(SCRIPT), "--interval", "5", "--timeout", "0.2"],
                                cwd=self.repo, env=env, capture_output=True, text=True, timeout=60)
        out = result.stdout + result.stderr
        self.assertEqual(result.returncode, 6, out)
        self.assertIn("[verdict] REMOTE_UNVERIFIED", out)
        self.assertNotIn(TOKEN, out)


if __name__ == "__main__":
    unittest.main()