599 lines
25 KiB
Python
599 lines
25 KiB
Python
"""审计批 5（I-9、I-14、N-1～N-12、I-12、事务分离）测试。
|
||
|
||
只用注入的假 runner 与 init_db 的本地库，不跑真实模型回合、
|
||
不联网。按任务书：新测试 + 相关旧套件确认无回归；本文件
|
||
只覆盖本批的新行为与新契约，旧行为由旧套件锁定。
|
||
"""
|
||
|
||
import http.client
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest import mock
|
||
|
||
try:
|
||
from cloud.agent_def import (
|
||
missing_required_tools,
|
||
parse_agent_tools_column,
|
||
)
|
||
from cloud.audit_loop import Finding
|
||
from cloud.auth import AuthStore
|
||
from cloud.models import parse_routing_policy
|
||
from cloud.routing_dryrun import dry_run_policy_document
|
||
from cloud.server import CloudControlPlaneHandler, create_server, init_db
|
||
from cloud.worker import Worker
|
||
import cloud.worker as worker_module
|
||
import cloud.server as server_module
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.agent_def import (
|
||
missing_required_tools,
|
||
parse_agent_tools_column,
|
||
)
|
||
from cloud.audit_loop import Finding
|
||
from cloud.auth import AuthStore
|
||
from cloud.models import parse_routing_policy
|
||
from cloud.routing_dryrun import dry_run_policy_document
|
||
from cloud.server import CloudControlPlaneHandler, create_server, init_db
|
||
from cloud.worker import Worker
|
||
import cloud.worker as worker_module
|
||
import cloud.server as server_module
|
||
|
||
|
||
def _completed_runner(cmd):
|
||
name = cmd[cmd.index("--name") + 1] if "dispatch" in cmd else "t"
|
||
return SimpleNamespace(
|
||
returncode=0,
|
||
stdout=json.dumps({"task": name, "status": "completed",
|
||
"text": "done"}),
|
||
stderr="")
|
||
|
||
|
||
def _insert_task(db_path, task_id, status="queued", audit_loop=0,
|
||
goal_mode=0, goal=None, verify_command=None,
|
||
goal_max_rounds=None, model=None, packet=None):
|
||
now = time.time()
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO tasks(id, project_id, prompt, status,"
|
||
" idempotency_key, depends_on, priority, created_by,"
|
||
" created_at, updated_at, last_heartbeat, audit_loop,"
|
||
" goal_mode, goal, verify_command, goal_max_rounds, model,"
|
||
" audit_loop_packet)"
|
||
" VALUES(?, 'p1', 'do work', ?, NULL, '[]', 0, 'tester',"
|
||
" ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?)",
|
||
(task_id, status, now, now, audit_loop, goal_mode, goal,
|
||
verify_command, goal_max_rounds, model, packet))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _set_status(db_path, task_id, status):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
|
||
(status, task_id))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _row(db_path, task_id):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
return conn.execute(
|
||
"SELECT status, audit_loop_packet, goal, verify_command,"
|
||
" goal_max_rounds, model FROM tasks WHERE id = ?",
|
||
(task_id,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _state_row(db_path, task_id):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
return conn.execute(
|
||
"SELECT round, last_decision FROM task_audit_loop_state"
|
||
" WHERE task_id = ?", (task_id,)).fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _seed_state(db_path, task_id, round_no, last_decision):
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
conn.execute(
|
||
"CREATE TABLE IF NOT EXISTS task_audit_loop_state("
|
||
"task_id TEXT PRIMARY KEY, round INTEGER,"
|
||
" passed_gates TEXT, criteria TEXT, last_decision TEXT,"
|
||
" updated_at DOUBLE PRECISION)")
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO task_audit_loop_state"
|
||
"(task_id, round, passed_gates, criteria, last_decision,"
|
||
" updated_at) VALUES(?, ?, ?, ?, ?, ?)",
|
||
(task_id, round_no, '["execution"]', '{}', last_decision,
|
||
time.time()))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
class _FailingExecuteConn:
|
||
"""注入失败的连接包装：指定 SQL 片段一律抛错，其余透传。"""
|
||
|
||
def __init__(self, real, needle):
|
||
self._real = real
|
||
self._needle = needle
|
||
|
||
def execute(self, sql, params=()):
|
||
if self._needle in sql:
|
||
raise RuntimeError("injected store failure")
|
||
return self._real.execute(sql, params)
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(self._real, name)
|
||
|
||
|
||
class _WorkerCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
init_db(self.db_path)
|
||
self.audits = []
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def _worker(self, **kwargs):
|
||
kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
|
||
kwargs.setdefault("max_retries", 0)
|
||
kwargs.setdefault("retry_backoff_seconds", 0)
|
||
kwargs.setdefault("runner", _completed_runner)
|
||
worker = Worker(self.db_path, "/fake/host.py", **kwargs)
|
||
worker._audit = lambda p, a, d="": self.audits.append((p, a, d))
|
||
return worker
|
||
|
||
def _actions(self):
|
||
return [a for _, a, _ in self.audits]
|
||
|
||
|
||
class FingerprintTestCase(unittest.TestCase):
|
||
def test_fallback_models_change_fingerprint(self):
|
||
# I-14①：指纹宣称含 fallback_models，变化必须敏感。
|
||
base = {"mode": "shadow", "tiers": {"default": "m1"},
|
||
"model_capabilities": {"m1": {"tool_calling": True}}}
|
||
a = parse_routing_policy(base)
|
||
b = parse_routing_policy(
|
||
dict(base, fallback_models=["m2"]))
|
||
self.assertNotEqual(a.content_fingerprint(),
|
||
b.content_fingerprint())
|
||
c = parse_routing_policy(dict(base, fallback_models=["m2"]))
|
||
self.assertEqual(b.content_fingerprint(),
|
||
c.content_fingerprint())
|
||
|
||
|
||
class WireTxTestCase(_WorkerCase):
|
||
def _patched_worker(self, needle):
|
||
real_connect = worker_module._db_connect
|
||
|
||
def _connect(path):
|
||
return _FailingExecuteConn(real_connect(path), needle)
|
||
|
||
patcher = mock.patch.object(worker_module, "_db_connect",
|
||
_connect)
|
||
patcher.start()
|
||
self.addCleanup(patcher.stop)
|
||
return self._worker()
|
||
|
||
def test_pushback_store_failure_rolls_back_state(self):
|
||
# 事务分离：推回 UPDATE 失败时状态保存一并回滚，任务不
|
||
# 以“已推回状态 + failed 任务行”的半截态落地。
|
||
_insert_task(self.db_path, "TP", status="failed", audit_loop=1)
|
||
worker = self._patched_worker("SET status = 'queued'")
|
||
task = worker.get_task("TP")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "failed", {"stdout": ""}, "boom")
|
||
self.assertEqual(decision, "error")
|
||
self.assertEqual(final, "failed")
|
||
row = _row(self.db_path, "TP")
|
||
self.assertEqual(row[0], "failed")
|
||
self.assertIsNone(row[1])
|
||
self.assertIsNone(_state_row(self.db_path, "TP"))
|
||
self.assertNotIn("worker_audit_loop_pushback", self._actions())
|
||
|
||
def test_escalate_store_failure_claims_nothing(self):
|
||
# I-14②（改动后新行为）：升级 UPDATE 失败时接线返回
|
||
# error、任务保持原终局、状态行不动，且不记“已升级”
|
||
# 审计——升级审计只在提交后记，绝不谎称已升级。
|
||
_insert_task(self.db_path, "TE", status="failed", audit_loop=1)
|
||
_seed_state(self.db_path, "TE", 3, "pushback")
|
||
worker = self._patched_worker("SET status = 'failed'")
|
||
task = worker.get_task("TE")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "failed", {"stdout": ""}, "boom")
|
||
self.assertEqual(decision, "error")
|
||
self.assertEqual(final, "failed")
|
||
self.assertEqual(_row(self.db_path, "TE")[0], "failed")
|
||
state = _state_row(self.db_path, "TE")
|
||
self.assertEqual(state[0], 3)
|
||
self.assertEqual(state[1], "pushback")
|
||
self.assertNotIn("worker_audit_loop_escalated", self._actions())
|
||
|
||
def test_escalate_cas_miss_audits_skipped(self):
|
||
# N-2：升级 CAS 未命中（状态已被改动）时记
|
||
# escalate_skipped 审计、不谎称已升级，任务与状态原样不动。
|
||
# 审计批 6（I2）：接线 error 时返回的不再是输入终局，
|
||
# 而是回滚后读回的行实际状态（queued）；run_one 终局块
|
||
# 据此抑制终态审计与通知。本用例断言意图随之改写：
|
||
# 旧断言 final == "failed"（输入终局）已是 I2 要消除的
|
||
# 矛盾，新断言 final == 行状态 "queued"。
|
||
_insert_task(self.db_path, "TC", status="failed", audit_loop=1)
|
||
_seed_state(self.db_path, "TC", 3, "pushback")
|
||
_set_status(self.db_path, "TC", "queued")
|
||
worker = self._worker()
|
||
task = worker.get_task("TC")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "failed", {"stdout": ""}, "boom")
|
||
self.assertEqual(decision, "error")
|
||
self.assertEqual(final, "queued")
|
||
self.assertEqual(_row(self.db_path, "TC")[0], "queued")
|
||
self.assertEqual(_state_row(self.db_path, "TC")[0], 3)
|
||
self.assertIn("worker_audit_loop_escalate_skipped",
|
||
self._actions())
|
||
self.assertNotIn("worker_audit_loop_escalated", self._actions())
|
||
|
||
|
||
class CoordClearTestCase(_WorkerCase):
|
||
def test_clear_packets_failure_is_best_effort(self):
|
||
# I-14③：清包失败只记警告、不炸调用方、不改已定结局。
|
||
_insert_task(self.db_path, "PP", status="failed", audit_loop=1,
|
||
packet='{"round": 1}')
|
||
_insert_task(self.db_path, "CC", status="failed", audit_loop=0,
|
||
packet='{"round": 1}')
|
||
worker = self._worker()
|
||
worker._coord_clear_packets(
|
||
{"id": "PP", "project_id": "p1"}, ["CC"])
|
||
self.assertIsNone(_row(self.db_path, "PP")[1])
|
||
self.assertIsNone(_row(self.db_path, "CC")[1])
|
||
|
||
def _boom(path):
|
||
raise RuntimeError("injected connect failure")
|
||
|
||
with mock.patch.object(worker_module, "_db_connect", _boom):
|
||
# 不抛错即通过：清失败绝不改变已定的父任务结局。
|
||
worker._coord_clear_packets(
|
||
{"id": "PP", "project_id": "p1"}, ["CC"])
|
||
|
||
|
||
class ClassifyAndSplitTestCase(_WorkerCase):
|
||
def test_merge_failed_is_gate_rejected(self):
|
||
# N-3：merge-failed 归 gate_rejected，不再是
|
||
# excluded/non_terminal；容忍声明仍优先。
|
||
worker = self._worker()
|
||
kind, reason = worker._coord_classify_child(
|
||
None, {"id": "C1", "status": "merge-failed"}, set())
|
||
self.assertEqual((kind, reason), ("gate_rejected", "merge_failed"))
|
||
kind, _ = worker._coord_classify_child(
|
||
{"on_failure": "tolerate"},
|
||
{"id": "C2", "status": "merge-failed"}, set())
|
||
self.assertEqual(kind, "excluded")
|
||
kind, _ = worker._coord_classify_child(
|
||
None, {"id": "C3", "status": "merged"}, set())
|
||
self.assertEqual(kind, "completed")
|
||
|
||
def test_wire_splits_subjective_before_evaluate(self):
|
||
# N-4：单任务接线先拆分再评估——确定性照常推回，主观
|
||
# 只记审计，不吞推回。
|
||
_insert_task(self.db_path, "TS", status="failed", audit_loop=1)
|
||
worker = self._worker()
|
||
det = Finding("execution_failed", "high", "boom", "re-run",
|
||
"task completes", kind="deterministic",
|
||
gate="execution")
|
||
subj = Finding("design_note", "low", "looks odd", "review",
|
||
"human ok", kind="subjective", gate=None)
|
||
worker._audit_loop_gate_results = (
|
||
lambda task, final, result, gates_checked=None: (
|
||
{"execution": False}, [det, subj]))
|
||
task = worker.get_task("TS")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "failed", {"stdout": ""}, "boom")
|
||
self.assertEqual(decision, "pushback")
|
||
self.assertEqual(final, "queued")
|
||
needs = [d for _, a, d in self.audits
|
||
if a == "worker_audit_loop_needs_review"]
|
||
self.assertTrue(any("design_note" in d for d in needs))
|
||
|
||
|
||
class TerminalSqlTestCase(unittest.TestCase):
|
||
def test_terminal_sql_matches_old_literal_set(self):
|
||
# N-6：生成的终态串集合与旧手写字面量逐字等价。
|
||
from cloud.worker import _terminal_status_sql
|
||
sql = _terminal_status_sql()
|
||
inner = sql.strip()[1:-1]
|
||
got = {part.strip().strip("'") for part in inner.split(",")}
|
||
self.assertEqual(
|
||
got,
|
||
{"completed", "failed", "interrupted", "merged",
|
||
"merge-failed"})
|
||
|
||
|
||
class GoalWireTestCase(_WorkerCase):
|
||
def test_goal_pushback_leaves_goal_columns_and_bounds_rounds(self):
|
||
# N-12 定性：推回不改 goal 列；包贴进内存 goal 与 prompt；
|
||
# 审计轮次 3 轮后必升级（无无界循环）。
|
||
_insert_task(self.db_path, "TG", status="failed", audit_loop=1,
|
||
goal_mode=1, goal="reach the goal",
|
||
verify_command="true", goal_max_rounds=3)
|
||
worker = self._worker()
|
||
task = worker.get_task("TG")
|
||
final, _, decision = worker._audit_loop_wire(
|
||
task, "failed", {"goal_met": False, "goal_error": "nope"},
|
||
"goal unmet")
|
||
self.assertEqual(decision, "pushback")
|
||
self.assertEqual(final, "queued")
|
||
row = _row(self.db_path, "TG")
|
||
self.assertEqual(row[2], "reach the goal")
|
||
self.assertEqual(row[3], "true")
|
||
self.assertEqual(row[4], 3)
|
||
packet = row[1]
|
||
self.assertIsNotNone(packet)
|
||
injected = worker._apply_audit_loop_packet({
|
||
"id": "TG", "project_id": "p1", "prompt": "do work",
|
||
"goal": "reach the goal", "goal_mode": 1,
|
||
"audit_loop_packet": packet})
|
||
self.assertTrue(injected["prompt"].startswith("[审计推回包"))
|
||
self.assertTrue(injected["goal"].startswith("[审计推回包"))
|
||
# 库里的原 prompt/goal 不动。
|
||
self.assertEqual(_row(self.db_path, "TG")[2], "reach the goal")
|
||
# 继续推回直至升级：轮次有界。
|
||
decisions = []
|
||
for _ in range(3):
|
||
_set_status(self.db_path, "TG", "failed")
|
||
task = worker.get_task("TG")
|
||
_, _, decision = worker._audit_loop_wire(
|
||
task, "failed",
|
||
{"goal_met": False, "goal_error": "nope"}, "goal unmet")
|
||
decisions.append(decision)
|
||
if decision == "escalate":
|
||
break
|
||
self.assertEqual(decisions[-1], "escalate")
|
||
self.assertEqual(_row(self.db_path, "TG")[0], "failed")
|
||
self.assertIsNone(_row(self.db_path, "TG")[1])
|
||
|
||
|
||
class ToolApprovalEffectiveTestCase(_WorkerCase):
|
||
def _insert_approval_task(self, task_id, approvals):
|
||
_insert_task(self.db_path, task_id, status="queued")
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute(
|
||
"UPDATE tasks SET agent_tools = ?,"
|
||
" agent_tool_approvals = ? WHERE id = ?",
|
||
(json.dumps(["websearch"]), json.dumps(approvals),
|
||
task_id))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
def test_ask_is_audited_as_decline_under_host_deny(self):
|
||
# I-12：派发时真的调 effective，ask 在行里折算成
|
||
# decline，ask_honoured 恒为 False。
|
||
self._insert_approval_task("TA", {"websearch": "ask"})
|
||
worker = self._worker()
|
||
out = worker.run_one()
|
||
self.assertEqual(out["status"], "completed")
|
||
details = [d for _, a, d in self.audits
|
||
if a == "worker_tool_approvals_declared"]
|
||
self.assertTrue(details)
|
||
self.assertIn('"websearch"', details[0])
|
||
self.assertIn('"declared": "ask"', details[0])
|
||
self.assertIn('"effective": "decline"', details[0])
|
||
self.assertIn('"ask_honoured": false', details[0])
|
||
|
||
def test_auto_accepts_only_when_host_config_is_auto(self):
|
||
# host approval 取自传给 host 的 --config 文件：auto 时
|
||
# auto 声明 effective 为 accept；默认 deny 时为 decline。
|
||
self._insert_approval_task("TB", {"websearch": "auto"})
|
||
cfg = os.path.join(self.tmp.name, "host-config.json")
|
||
with open(cfg, "w", encoding="utf-8") as fh:
|
||
json.dump({"approval": "auto"}, fh)
|
||
worker = self._worker(config_path=cfg)
|
||
out = worker.run_one()
|
||
self.assertEqual(out["status"], "completed")
|
||
details = [d for _, a, d in self.audits
|
||
if a == "worker_tool_approvals_declared"]
|
||
self.assertIn('"effective": "accept"', details[0])
|
||
|
||
|
||
class GatewayExtractTestCase(_WorkerCase):
|
||
def _policy(self, capable):
|
||
return {"mode": "active",
|
||
"tiers": {"default": {
|
||
"model": "m1",
|
||
"requires": {"tool_calling": True}}},
|
||
"model_capabilities": {"m1": {"tool_calling": capable}}}
|
||
|
||
def test_blocked_and_pass_through(self):
|
||
# I-9①：抽取后行为逐字不变——被拦时早返回 dict +
|
||
# 门审计 + 标记点；满足时返回 None 继续派发。
|
||
_insert_task(self.db_path, "TG1", status="queued")
|
||
worker = self._worker(routing_policy=self._policy(False))
|
||
task = {"id": "TG1", "project_id": "p1", "model": "m1"}
|
||
blocked = worker._gateway_capability_check(
|
||
task, "sub2api", False, time.time())
|
||
self.assertIsNotNone(blocked)
|
||
self.assertEqual(blocked["status"], "failed")
|
||
self.assertIn("worker_capability_gate_rejected", self._actions())
|
||
self.assertIn("capability", task.get("_gates_checked", []))
|
||
|
||
self.audits.clear()
|
||
_insert_task(self.db_path, "TG2", status="queued")
|
||
worker2 = self._worker(routing_policy=self._policy(True))
|
||
task2 = {"id": "TG2", "project_id": "p1", "model": "m1"}
|
||
self.assertIsNone(worker2._gateway_capability_check(
|
||
task2, "sub2api", False, time.time()))
|
||
self.assertNotIn("worker_capability_gate_rejected",
|
||
self._actions())
|
||
|
||
|
||
class PersistExtractTestCase(_WorkerCase):
|
||
def test_writeback_only_for_applied_policy(self):
|
||
# I-9②：只持久化 source=policy 且 applied 的决策。
|
||
_insert_task(self.db_path, "TW", status="queued", model=None)
|
||
worker = self._worker()
|
||
task = {"id": "TW", "project_id": "p1"}
|
||
worker._persist_policy_model(
|
||
task, SimpleNamespace(chosen_model="m9", source="policy",
|
||
applied=True, tier="default"))
|
||
self.assertEqual(_row(self.db_path, "TW")[5], "m9")
|
||
self.assertIn("worker_model_persisted", self._actions())
|
||
self.audits.clear()
|
||
_insert_task(self.db_path, "TW2", status="queued", model=None)
|
||
worker._persist_policy_model(
|
||
{"id": "TW2", "project_id": "p1"},
|
||
SimpleNamespace(chosen_model="m9", source="shadow",
|
||
applied=False, tier="default"))
|
||
self.assertIsNone(_row(self.db_path, "TW2")[5])
|
||
self.assertNotIn("worker_model_persisted", self._actions())
|
||
|
||
|
||
class ServerHelperTestCase(unittest.TestCase):
|
||
def test_agent_definition_and_audit_loop_helpers(self):
|
||
# I-9③：两个小函数与原地行为一致（错误文案逐字）。
|
||
resolve = CloudControlPlaneHandler._resolve_create_agent_definition
|
||
_, err = resolve({"agent": {"a": 1}, "agent_name": "x"})
|
||
self.assertEqual(err, "agent and agent_name cannot both be used")
|
||
_, err = resolve({"agent": "not-a-dict"})
|
||
self.assertEqual(err, "agent must be an object")
|
||
validate = CloudControlPlaneHandler._validate_create_audit_loop
|
||
self.assertEqual(validate({}), (0, None))
|
||
self.assertEqual(validate({"audit_loop": True}), (1, None))
|
||
self.assertEqual(validate({"audit_loop": 1}), (1, None))
|
||
_, err = validate({"audit_loop": 2})
|
||
self.assertEqual(err, "audit_loop must be true or 1")
|
||
|
||
|
||
class ParseToleranceTestCase(unittest.TestCase):
|
||
def test_non_string_items_survive_and_are_tolerated(self):
|
||
# N-9（文档选项）：解析层只保 list 形状，下游容忍非
|
||
# 字符串项——不过滤、不改旧脏行序列化。
|
||
self.assertEqual(parse_agent_tools_column('["a", 1]'),
|
||
["a", 1])
|
||
self.assertEqual(missing_required_tools(["a", 1], ""),
|
||
["a", 1])
|
||
|
||
|
||
class DryRunEmptyTestCase(unittest.TestCase):
|
||
def test_early_returns_share_shape(self):
|
||
# N-10：两处早期返回经 _empty_report 归一，键集合一致。
|
||
a = dry_run_policy_document("not-a-dict")
|
||
b = dry_run_policy_document({"mode": "bogus"})
|
||
self.assertFalse(a["valid"])
|
||
self.assertFalse(b["valid"])
|
||
self.assertEqual(set(a.keys()), set(b.keys()))
|
||
|
||
|
||
class ShadowEndpointTestCase(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)
|
||
|
||
def tearDown(self):
|
||
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)
|
||
headers = {"Content-Type": "application/json"}
|
||
if key:
|
||
headers["X-API-Key"] = key
|
||
conn.request(method, path,
|
||
json.dumps(body) if body is not None else None,
|
||
headers)
|
||
resp = conn.getresponse()
|
||
raw = resp.read()
|
||
conn.close()
|
||
try:
|
||
data = json.loads(raw)
|
||
except ValueError:
|
||
data = None
|
||
return resp.status, data
|
||
|
||
def _shadow(self):
|
||
status, report = self._req(
|
||
"GET", "/routing-shadow?project_id=" + self.project["id"],
|
||
key=self.owner_key)
|
||
self.assertEqual(status, 200, report)
|
||
return report
|
||
|
||
def test_cost_query_failure_flagged_not_500(self):
|
||
# N-1 + I-14④：成本查询失败时报告仍 200、成本按空算、
|
||
# cost_unavailable 标明，不抛到 handler 层。
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute("DROP TABLE IF EXISTS cost_records")
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
class _BoomCostStore:
|
||
def __init__(self, *args, **kwargs):
|
||
raise RuntimeError("injected cost store failure")
|
||
|
||
with mock.patch.object(server_module, "CostStore",
|
||
_BoomCostStore):
|
||
report = self._shadow()
|
||
self.assertTrue(report["cost_unavailable"])
|
||
|
||
def test_sample_window_and_normal_cost_flag(self):
|
||
# N-5：截断样本（最早 10000 行）附带最早/最晚时间窗；
|
||
# 正常路径 cost_unavailable 为 False。
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
for task_id, created in (("t1", 10.0), ("t2", 20.0)):
|
||
conn.execute(
|
||
"INSERT INTO audit_log(user_id, project_id, action,"
|
||
" detail, created_at) VALUES(?, ?, ?, ?, ?)",
|
||
("system", self.project["id"], "worker_routing",
|
||
"%s tier=plan source=shadow applied=False"
|
||
" suggested=mx" % task_id, created))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
report = self._shadow()
|
||
self.assertFalse(report["cost_unavailable"])
|
||
self.assertEqual(report["sample_window"],
|
||
{"earliest_created_at": 10.0,
|
||
"latest_created_at": 20.0})
|
||
self.assertFalse(report["truncated"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|