test: resident website loop end-to-end with real fetch
This commit is contained in:
host committed
1 parent
0722ca3fd8
commit
561b3891fb
1 file changed
+364
@@ -0,0 +1,364 @@
|
||||
"""常驻监听网站闭环端到端测试。
|
||||
|
||||
上一轮只验到注入假抓取的单测：真实抓取、真实 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):
|
||||
socket.getaddrinfo = self._orig_getaddrinfo
|
||||
if self._old_no_proxy is None:
|
||||
os.environ.pop("no_proxy", None)
|
||||
else:
|
||||
os.environ["no_proxy"] = self._old_no_proxy
|
||||
if self._old_NO_PROXY is None:
|
||||
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
|
||||
self.content_thread.join(timeout=5)
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.thread.join(timeout=5)
|
||||
try:
|
||||
self.auth.conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.tmp.cleanup()
|
||||
|
||||
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()
|
||||
Reference in new issue
Block a user