host
8fe4ad2f38
New module cloud/routing_dryrun.py: dry-run a merged policy document or an env mapping (file first, JSON overrides whole tier entries, CLOUD_ROUTING_MODE overrides global mode only, never explicit tier modes), reporting per-tier effective model/mode/mode-source, dropped tiers with reasons, absent and unknown tiers, and would_activate/would_shadow/would_off. Dropped or unknown tiers block (CLI exit 1); warnings alone (no version, fallback_models parsed but not applied, failed file/JSON sources that the loader silently ignores) do not block but are always printed. Model names are NOT validated against a catalog (deliberately not whitelisted) and sample counts are NOT checked here (that is the shadow report's job); the CLI never writes env, files, or worker state. Tests: test_routing_dryrun.py (8 tests).
170 lines
7.1 KiB
Python
170 lines
7.1 KiB
Python
"""策略干跑校验 v0d 单测：有效档、丢档、未知档、环境覆盖、CLI。
|
|
|
|
运行（在仓库根目录下）：
|
|
python3 -m unittest discover -s cloud/tests -p "test_routing_dryrun.py" -v
|
|
|
|
说明：纯函数部分用 plain dict，不碰库、不联网、不跑模型；
|
|
CLI 部分用子进程跑 `python3 -m cloud.routing_dryrun`，只读
|
|
传入的策略，不改任何文件或环境。关键断言：干跑本身不
|
|
激活任何东西（只报告 would_activate）；丢档会阻塞退出码；
|
|
文件/JSON 合并是整档替换、不是按字段合并；环境模式只
|
|
覆盖全局、不覆盖显式档模式。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.routing_dryrun import dry_run_from_env
|
|
from cloud.routing_dryrun import dry_run_policy_document
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.routing_dryrun import dry_run_from_env
|
|
from cloud.routing_dryrun import dry_run_policy_document
|
|
|
|
|
|
class DryRunDocumentTestCase(unittest.TestCase):
|
|
def test_plan_active_only(self):
|
|
report = dry_run_policy_document({
|
|
"version": "v1", "mode": "shadow",
|
|
"tiers": {
|
|
"plan": {"model": "gpt-6.1-sol", "mode": "active"},
|
|
"goal": "gpt-6.1-sol",
|
|
"default": "gpt-6.1-sol",
|
|
},
|
|
})
|
|
self.assertTrue(report["valid"])
|
|
self.assertFalse(report["blocked"])
|
|
self.assertEqual(report["tiers"]["plan"]["mode"], "active")
|
|
self.assertEqual(report["tiers"]["plan"]["mode_source"],
|
|
"tier_mode")
|
|
self.assertEqual(report["tiers"]["goal"]["mode"], "shadow")
|
|
self.assertEqual(report["tiers"]["goal"]["mode_source"],
|
|
"global_mode")
|
|
self.assertEqual(report["would_activate"], ["plan"])
|
|
|
|
def test_dropped_tier_blocks(self):
|
|
report = dry_run_policy_document({
|
|
"mode": "active",
|
|
"tiers": {
|
|
"plan": {"model": "gpt-6.1-sol", "mode": "activ"},
|
|
"default": "gpt-6.1-sol",
|
|
},
|
|
})
|
|
self.assertTrue(report["valid"])
|
|
self.assertTrue(report["blocked"])
|
|
self.assertEqual(report["dropped_tiers"], ["plan"])
|
|
self.assertEqual(report["tiers"]["plan"]["status"], "dropped")
|
|
self.assertEqual(report["tiers"]["default"]["status"], "active")
|
|
|
|
def test_unknown_tier_and_absent_tier(self):
|
|
report = dry_run_policy_document({
|
|
"mode": "shadow",
|
|
"tiers": {"plan": "gpt-6.1-sol", "weird": "gpt-6.1-sol"},
|
|
})
|
|
self.assertEqual(report["unknown_tiers"], ["weird"])
|
|
self.assertTrue(report["blocked"])
|
|
self.assertEqual(report["tiers"]["goal"]["status"], "absent")
|
|
self.assertIsNone(report["tiers"]["goal"]["model"])
|
|
|
|
def test_invalid_global_mode_and_no_version_warning(self):
|
|
report = dry_run_policy_document({
|
|
"mode": "bogus", "tiers": {"plan": "gpt-6.1-sol"},
|
|
})
|
|
self.assertFalse(report["valid"])
|
|
self.assertTrue(report["blocked"])
|
|
valid = dry_run_policy_document({
|
|
"mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"},
|
|
"fallback_models": ["gpt-5.6-terra"],
|
|
})
|
|
self.assertTrue(any("no policy version" in w
|
|
for w in valid["warnings"]))
|
|
self.assertTrue(any("fallback_models" in w
|
|
for w in valid["warnings"]))
|
|
|
|
def test_env_mode_overrides_global_not_tier(self):
|
|
report = dry_run_policy_document({
|
|
"mode": "shadow",
|
|
"tiers": {
|
|
"plan": {"model": "gpt-6.1-sol", "mode": "active"},
|
|
"default": "gpt-6.1-sol",
|
|
},
|
|
}, env_mode_override="active")
|
|
self.assertEqual(report["global_mode"], "active")
|
|
self.assertEqual(report["global_mode_source"], "env_override")
|
|
self.assertEqual(report["tiers"]["plan"]["mode"], "active")
|
|
self.assertEqual(report["tiers"]["default"]["mode"], "active")
|
|
bad = dry_run_policy_document({
|
|
"mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"},
|
|
}, env_mode_override="bogus")
|
|
self.assertEqual(bad["global_mode"], "shadow")
|
|
self.assertTrue(any("invalid and ignored" in w
|
|
for w in bad["warnings"]))
|
|
|
|
|
|
class DryRunEnvTestCase(unittest.TestCase):
|
|
def test_file_json_merge_replaces_whole_tier(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
policy_path = os.path.join(tmpdir, "policy.json")
|
|
with open(policy_path, "w", encoding="utf-8") as fh:
|
|
json.dump({
|
|
"mode": "shadow",
|
|
"tiers": {
|
|
"plan": {"model": "gpt-6.1-sol",
|
|
"mode": "active"},
|
|
"default": "gpt-6.1-sol",
|
|
},
|
|
}, fh)
|
|
# JSON 覆盖 plan 为字符串：整档替换，档模式丢失，
|
|
# plan 回到全局 shadow，而不是字段合并保留 active。
|
|
report = dry_run_from_env({
|
|
"CLOUD_ROUTING_POLICY_FILE": policy_path,
|
|
"CLOUD_ROUTING_POLICY_JSON": json.dumps({
|
|
"tiers": {"plan": "gpt-6.1-sol"},
|
|
}),
|
|
})
|
|
self.assertEqual(report["tiers"]["plan"]["mode"], "shadow")
|
|
self.assertEqual(report["tiers"]["plan"]["mode_source"],
|
|
"global_mode")
|
|
|
|
def test_unreadable_file_is_warning_not_silent(self):
|
|
report = dry_run_from_env({
|
|
"CLOUD_ROUTING_POLICY_FILE": "/nonexistent/policy.json",
|
|
"CLOUD_ROUTING_POLICY_JSON": json.dumps({
|
|
"mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"},
|
|
}),
|
|
})
|
|
self.assertTrue(report["valid"])
|
|
self.assertTrue(any("could not be read" in w
|
|
for w in report["warnings"]))
|
|
|
|
def test_cli_exit_codes(self):
|
|
repo_root = str(Path(__file__).resolve().parents[2])
|
|
good = subprocess.run(
|
|
[sys.executable, "-m", "cloud.routing_dryrun",
|
|
"--policy-json", json.dumps({
|
|
"mode": "shadow", "tiers": {"plan": "gpt-6.1-sol"},
|
|
})],
|
|
cwd=repo_root, capture_output=True, text=True, timeout=30)
|
|
self.assertEqual(good.returncode, 0, good.stderr)
|
|
report = json.loads(good.stdout)
|
|
self.assertTrue(report["valid"])
|
|
bad = subprocess.run(
|
|
[sys.executable, "-m", "cloud.routing_dryrun",
|
|
"--policy-json", json.dumps({
|
|
"mode": "active",
|
|
"tiers": {
|
|
"plan": {"model": "gpt-6.1-sol", "mode": "activ"},
|
|
"default": "gpt-6.1-sol",
|
|
},
|
|
})],
|
|
cwd=repo_root, capture_output=True, text=True, timeout=30)
|
|
self.assertEqual(bad.returncode, 1, bad.stdout)
|
|
self.assertEqual(json.loads(bad.stdout)["dropped_tiers"],
|
|
["plan"])
|