"""常驻监听网站闭环端到端测试。

上一轮只验到注入假抓取的单测：真实抓取、真实 run_loop 轮次
边界、经服务端 API 的全链路（创建 -> 基线 -> 变化建任务 ->
暂停 -> 恢复 -> 失败退避 -> 结束）都没有跑过。本文件补这一环，
不改任何生产默认值、不改生产校验口径：

- 间隔仍是生产下限 60 秒；到期不是改间隔，而是给驱动传一个
  更晚的 now（模拟时间流逝），与上一轮单测同款时间注入。
- 目标 URL 仍走生产校验：主机名 website-loop.test 不是 IP 字
  面、不是 localhost，所以校验通过；测试进程内只把这个主机
  名的 DNS 解析指向本地测试 HTTP 服务（相当于 /etc/hosts 的
  一条测试条目），抓取本身走真实 default_fetch（urllib、超时、
  1MB 截断），不注入假 fetcher、不绕过生产代码。
- 内容服务是线程内真实 http.server，可变内容、也可停服；
  控制面是真实 ThreadingHTTPServer + 真实 SQLite 临时库；
  变化建出的任务、轮询、审计都直接读库/接口核对。

不联网、不跑真实模型、不碰生产。
"""
import http.client
import http.server
import json
import os
import socket
import sqlite3
import sys
import tempfile
import threading
import time
import unittest
from pathlib import Path
from types import SimpleNamespace

try:
    from cloud.auth import AuthStore
    from cloud.resident import _content_hash
    from cloud.server import create_server
    from cloud.worker import Worker
except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.auth import AuthStore
    from cloud.resident import _content_hash
    from cloud.server import create_server
    from cloud.worker import Worker

HOST = "website-loop.test"


class WebsiteLoopTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        self.auth = AuthStore(self.db_path)
        self.owner_id, self.owner_key = self.auth.create_user("owner")
        self.server = create_server(self.db_path, "127.0.0.1", 0)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, daemon=True)
        self.thread.start()
        status, self.project = self._req(
            "POST", "/projects", {"name": "demo"}, self.owner_key)
        self.assertEqual(status, 201, self.project)
        # 可变内容服务：state 字典跨重启保留，停服/重启可复现。
        self.state = {"content": "version one"}
        state = self.state

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_GET(self):
                body = state["content"].encode("utf-8")
                self.send_response(200)
                self.send_header("Content-Type", "text/plain")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, *args):
                pass

        self._handler = Handler
        self.content_server = http.server.ThreadingHTTPServer(
            ("127.0.0.1", 0), Handler)
        self.content_port = self.content_server.server_address[1]
        self.content_thread = threading.Thread(
            target=self.content_server.serve_forever, daemon=True)
        self.content_thread.start()
        # 测试专用 DNS：只把 website-loop.test 指到本地服务，
        # 其它主机一律走原始解析；生产校验与抓取代码一字不改。
        self._orig_getaddrinfo = socket.getaddrinfo

        def patched(host, port, family=0, type=0, proto=0, flags=0):
            if host == HOST:
                return [(socket.AF_INET, socket.SOCK_STREAM, 6, "",
                         ("127.0.0.1", port))]
            return self._orig_getaddrinfo(
                host, port, family, type, proto, flags)

        socket.getaddrinfo = patched
        # 测试环境自带 egress 代理：urllib 默认会把这个测试主机名
        # 也送代理并被断开。只给本测试主机名加 no_proxy（测试
        # 进程内生效），生产抓取代码与代理行为一字不改。
        self._old_no_proxy = os.environ.get("no_proxy")
        self._old_NO_PROXY = os.environ.get("NO_PROXY")
        os.environ["no_proxy"] = (
            (self._old_no_proxy + ",") if self._old_no_proxy
            else "") + HOST
        os.environ["NO_PROXY"] = (
            (self._old_NO_PROXY + ",") if self._old_NO_PROXY
            else "") + HOST
        self.worker = Worker(
            self.db_path, "/fake/host.py",
            env={"XCM_API_KEY": "test-placeholder"},
            runner=lambda cmd: SimpleNamespace(
                returncode=0, stdout="{}", stderr=""))

    def tearDown(self):
        # setUp 中途失败时属性可能未建：getattr 防御，避免
        # tearDown 的 AttributeError 掩盖原始失败。
        if getattr(self, "_orig_getaddrinfo", None) is not None:
            socket.getaddrinfo = self._orig_getaddrinfo
        if getattr(self, "_old_no_proxy", None) is None:
            if "_old_no_proxy" in self.__dict__:
                os.environ.pop("no_proxy", None)
        else:
            os.environ["no_proxy"] = self._old_no_proxy
        if getattr(self, "_old_NO_PROXY", None) is None:
            if "_old_NO_PROXY" in self.__dict__:
                os.environ.pop("NO_PROXY", None)
        else:
            os.environ["NO_PROXY"] = self._old_NO_PROXY
        try:
            self.content_server.shutdown()
            self.content_server.server_close()
        except Exception:
            pass
        try:
            self.content_thread.join(timeout=5)
        except Exception:
            pass
        try:
            self.server.shutdown()
            self.server.server_close()
        except Exception:
            pass
        try:
            self.thread.join(timeout=5)
        except Exception:
            pass
        try:
            self.auth.conn.close()
        except Exception:
            pass
        try:
            self.tmp.cleanup()
        except Exception:
            pass

    def _req(self, method, path, body=None, key=None):
        conn = http.client.HTTPConnection("127.0.0.1", self.port,
                                          timeout=10)
        hdrs = {}
        payload = None
        if body is not None:
            payload = json.dumps(body).encode("utf-8")
            hdrs["Content-Type"] = "application/json"
        if key is not None:
            hdrs["X-API-Key"] = key
        try:
            conn.request(method, path, body=payload, headers=hdrs)
            resp = conn.getresponse()
            data = resp.read()
        finally:
            conn.close()
        try:
            parsed = json.loads(data.decode("utf-8")) if data else None
        except ValueError:
            parsed = None
        return resp.status, parsed

    def _url(self, suffix=""):
        return "/projects/%s/resident-watches%s" % (
            self.project["id"], suffix)

    def _create_watch(self):
        body = {"name": "site1",
                "target_url": "http://%s:%d/data" % (
                    HOST, self.content_port),
                "prompt": "变化时汇报",
                "interval_seconds": 60}
        status, watch = self._req("POST", self._url(), body,
                                  self.owner_key)
        self.assertEqual(status, 201, watch)
        return watch

    def _get_watch(self, watch_id):
        status, body = self._req(
            "GET", self._url("/" + watch_id), None, self.owner_key)
        self.assertEqual(status, 200, body)
        return body

    def _tasks(self):
        conn = sqlite3.connect(self.db_path)
        try:
            return conn.execute(
                "SELECT id, prompt, status FROM tasks"
                " ORDER BY created_at, id").fetchall()
        finally:
            conn.close()

    def _audit(self):
        conn = sqlite3.connect(self.db_path)
        try:
            return [r[0] for r in conn.execute(
                "SELECT action FROM audit_log WHERE project_id = ?",
                (self.project["id"],)).fetchall()]
        finally:
            conn.close()

    def test_full_website_loop_end_to_end(self):
        # 1. 创建：经服务端 API，running、游标为空、下次轮询在位。
        watch = self._create_watch()
        wid = watch["id"]
        self.assertEqual(watch["status"], "running")
        self.assertEqual(watch["interval_seconds"], 60)
        self.assertEqual(watch["fetch_timeout_seconds"], 10)
        self.assertIsNone(watch["last_content_hash"])
        self.assertIsNotNone(watch["next_poll_at"])
        self.assertIn("resident_watch_created", self._audit())
        t0 = time.time()

        # 2. 基线：真实抓取 version one，只建游标，不建任务。
        polls = self.worker.process_due_resident_watches(now=t0 + 120)
        self.assertEqual(len(polls), 1, polls)
        self.assertEqual(polls[0]["status"], "baseline")
        self.assertEqual(
            polls[0]["observed_hash"], _content_hash("version one"))
        self.assertEqual(self._tasks(), [])
        body = self._get_watch(wid)
        self.assertEqual(body["last_content_hash"],
                         _content_hash("version one"))
        self.assertEqual(len(body["polls"]), 1)
        self.assertEqual(body["polls"][0]["status"], "baseline")

        # 3. 变化：内容改成 version two，到期轮询建出 queued 任务。
        self.state["content"] = "version two"
        polls = self.worker.process_due_resident_watches(now=t0 + 240)
        self.assertEqual(len(polls), 1, polls)
        self.assertEqual(polls[0]["status"], "changed")
        self.assertIsNotNone(polls[0]["task_id"])
        tasks = self._tasks()
        self.assertEqual(len(tasks), 1)
        self.assertEqual(tasks[0][0], polls[0]["task_id"])
        self.assertEqual(tasks[0][2], "queued")
        self.assertIn("变化时汇报", tasks[0][1])
        self.assertIn("version two", tasks[0][1])
        self.assertIn("resident_watch_change_task_created",
                      self._audit())
        body = self._get_watch(wid)
        self.assertEqual(body["last_content_hash"],
                         _content_hash("version two"))
        self.assertIn("version two", body["last_change_text"])

        # 4. 暂停：暂停期间再改内容，驱动不轮询、游标不动。
        status, body = self._req(
            "POST", self._url("/%s/pause" % wid), {}, self.owner_key)
        self.assertEqual(status, 200, body)
        self.assertEqual(body["status"], "paused")
        self.assertIsNone(body["next_poll_at"])
        self.state["content"] = "version three"
        polls = self.worker.process_due_resident_watches(now=t0 + 400)
        self.assertEqual(polls, [])
        body = self._get_watch(wid)
        self.assertEqual(len(body["polls"]), 2)
        self.assertEqual(body["last_content_hash"],
                         _content_hash("version two"))

        # 5. 恢复：用旧游标比对，报出暂停期间的变化并建任务，
        #    不是从头建基线。
        status, body = self._req(
            "POST", self._url("/%s/resume" % wid), {}, self.owner_key)
        self.assertEqual(status, 200, body)
        self.assertEqual(body["status"], "running")
        polls = self.worker.process_due_resident_watches(now=t0 + 400)
        self.assertEqual(len(polls), 1, polls)
        self.assertEqual(polls[0]["status"], "changed")
        tasks = self._tasks()
        self.assertEqual(len(tasks), 2)
        self.assertIn("version three", tasks[1][1])
        body = self._get_watch(wid)
        self.assertEqual(body["last_content_hash"],
                         _content_hash("version three"))

        # 6. 失败：停服后轮询记 failed、退避、不结束、游标不动；
        #    恢复服务后内容未变，轮询记 unchanged、失败清零。
        self.content_server.shutdown()
        self.content_server.server_close()
        self.content_thread.join(timeout=5)
        polls = self.worker.process_due_resident_watches(now=t0 + 600)
        self.assertEqual(len(polls), 1, polls)
        self.assertEqual(polls[0]["status"], "failed")
        body = self._get_watch(wid)
        self.assertEqual(body["status"], "running")
        self.assertEqual(body["consecutive_failures"], 1)
        self.assertEqual(body["last_content_hash"],
                         _content_hash("version three"))
        self.assertGreater(body["next_poll_at"], t0 + 600 + 60)
        self.content_server = http.server.ThreadingHTTPServer(
            ("127.0.0.1", self.content_port), self._handler)
        self.content_thread = threading.Thread(
            target=self.content_server.serve_forever, daemon=True)
        self.content_thread.start()
        polls = self.worker.process_due_resident_watches(now=t0 + 800)
        self.assertEqual(len(polls), 1, polls)
        self.assertEqual(polls[0]["status"], "unchanged")
        body = self._get_watch(wid)
        self.assertEqual(body["consecutive_failures"], 0)
        self.assertEqual(len(self._tasks()), 2)

        # 7. 结束：终态，驱动不再轮询，改/停/恢复一律 409。
        status, body = self._req(
            "POST", self._url("/%s/end" % wid), {}, self.owner_key)
        self.assertEqual(status, 200, body)
        self.assertEqual(body["status"], "ended")
        self.assertIsNone(body["next_poll_at"])
        self.assertEqual(body["last_content_hash"],
                         _content_hash("version three"))
        polls = self.worker.process_due_resident_watches(now=t0 + 1000)
        self.assertEqual(polls, [])
        for action in ("pause", "resume", "end"):
            status, body = self._req(
                "POST", self._url("/%s/%s" % (wid, action)), {},
                self.owner_key)
            self.assertEqual(status, 409, action)
        status, _ = self._req(
            "PATCH", self._url("/" + wid), {"interval_seconds": 120},
            self.owner_key)
        self.assertEqual(status, 409)
        body = self._get_watch(wid)
        self.assertEqual(body["status"], "ended")
        self.assertEqual(len(body["polls"]), 5)

    def test_run_loop_boundary_calls_resident_driver(self):
        # 8. 驱动接线：run_loop 轮次边界确实调到常驻驱动，
        #    不只是测试直接调 process_due。run_all 等无关边界
        #    在本测试里打桩，只留 resident 驱动走真实抓取。
        watch = self._create_watch()
        wid = watch["id"]
        worker = self.worker
        calls = []
        original = Worker.process_due_resident_watches

        def recording():
            # 实例属性上的函数不会自动绑定 self，必须闭包捕获
            # worker，不能写成 def recording(self)。
            calls.append(time.time())
            return original(worker)

        worker.process_due_resident_watches = recording
        worker.run_all = lambda: []
        worker.cleanup_stale_containers = lambda: None
        worker._write_heartbeat = lambda: None
        worker.register = lambda: None
        worker._process_push_due = lambda: None
        worker.process_auto_memories = lambda: None
        worker._reap_kill_requests = lambda *a, **k: None
        worker.auto_resume_interrupted = lambda: None
        worker._heartbeat_running_tasks = lambda: None
        worker._registry_heartbeat = lambda: None
        worker._code_changed = lambda: False
        worker._code_signature_valid = lambda: True
        worker.run_loop(interval=0.05, max_seconds=0.4)
        self.assertGreaterEqual(len(calls), 1)
        body = self._get_watch(wid)
        self.assertEqual(len(body["polls"]), 1)
        self.assertEqual(body["polls"][0]["status"], "baseline")
        self.assertEqual(body["last_content_hash"],
                         _content_hash("version one"))


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