host
b391fe9320
- DNS 级 SSRF pin：解析后 pin IP 建连，防 DNS 重绑定 - 批量总预算与每日上限/抖动冷却/保留期 - PG schema 与迁移清单 - 双连接死锁隔离 - 改目标重置基线 - 显式代理配置与结构化失败原因 - /metrics resident 口径
417 lines
18 KiB
Python
417 lines
18 KiB
Python
"""常驻监听网站闭环端到端测试。
|
||
|
||
上一轮只验到注入假抓取的单测：真实抓取、真实 run_loop 轮次
|
||
边界、经服务端 API 的全链路（创建 -> 基线 -> 变化建任务 ->
|
||
暂停 -> 恢复 -> 失败退避 -> 结束）都没有跑过。本文件补这一环，
|
||
不改任何生产默认值、不改生产校验口径：
|
||
|
||
- 间隔仍是生产下限 60 秒；到期不是改间隔，而是给驱动传一个
|
||
更晚的 now（模拟时间流逝），与上一轮单测同款时间注入。
|
||
- 目标 URL 仍走生产校验：主机名 website-loop.test 不是 IP 字
|
||
面、不是 localhost，所以校验通过；抓取本身走真实
|
||
default_fetch（DNS 门、pin 连接、超时、1MB 截断），不注入
|
||
假 fetcher。注意：生产默认 DNS 门会把解析到 127.0.0.1 的
|
||
目标拦住，所以本文件用两段测试接缝：worker 的
|
||
resident_resolver 把该主机名解析成公网测试 IP（门正常放
|
||
行），测试进程内再在 socket 层把该 IP:内容端口重定向回
|
||
本地服务（传输层终点是测试替身，门与 pin 全程真实跑）；
|
||
DNS 门本身由 test_resident_p0_fixes.py 的默认解析器用例
|
||
覆盖，不在本文件重复证明。
|
||
- 内容服务是线程内真实 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=""))
|
||
# 两段测试接缝（见文件头说明）：resolver 返回公网测试
|
||
# IP，socket 层把该 IP:内容端口重定向回本地服务。
|
||
try:
|
||
from cloud.resident import _default_resolver
|
||
except ImportError:
|
||
_default_resolver = None
|
||
|
||
def _loop_resolver(host, port, _default=_default_resolver):
|
||
if host == HOST:
|
||
return ["93.184.216.34"]
|
||
if _default is not None:
|
||
return _default(host, port)
|
||
return [socket.getaddrinfo(host, port)[0][4][0]]
|
||
|
||
self.worker.resident_resolver = _loop_resolver
|
||
self._orig_create_connection = socket.create_connection
|
||
|
||
def _patched_create_connection(address, *args, **kwargs):
|
||
if address[0] == "93.184.216.34" \
|
||
and address[1] == self.content_port:
|
||
address = ("127.0.0.1", self.content_port)
|
||
return self._orig_create_connection(address, *args,
|
||
**kwargs)
|
||
|
||
socket.create_connection = _patched_create_connection
|
||
|
||
def tearDown(self):
|
||
# setUp 中途失败时属性可能未建：getattr 防御，避免
|
||
# tearDown 的 AttributeError 掩盖原始失败。
|
||
if getattr(self, "_orig_getaddrinfo", None) is not None:
|
||
socket.getaddrinfo = self._orig_getaddrinfo
|
||
if getattr(self, "_orig_create_connection", None) is not None:
|
||
socket.create_connection = self._orig_create_connection
|
||
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. 恢复：用旧游标比对，报出暂停期间的变化并建任务，
|
||
# 不是从头建基线。轮询时刻取 t0+540：距上次建任务
|
||
# （t0+240）已过 300 秒抖动冷却，否则新冷却语义会
|
||
# 正确地记 changed_cooldown 而不是 changed（冷却期
|
||
# 内的变化不丢，冷却过后复查仍可报，见 P0 批单测）。
|
||
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 + 540)
|
||
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()
|