d58964c105
礼品卡 face_min/max + discount_at_*；e2e E0–E3（可选 --call）； chain-policy 接受 sms 单点；本机 codex health 路径约定；碎银折扣列。
200 lines
7.3 KiB
Python
200 lines
7.3 KiB
Python
"""chain_v2 breaks · sms single-source · path_id · pool burn."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
from joy.dashboard import manhuang as mh
|
||
|
||
|
||
def test_compute_path_id_stable():
|
||
np = {
|
||
"primary": {
|
||
"name": "clash-meta-9090",
|
||
"mode": "rule",
|
||
"selected": "香港300",
|
||
"mixed_port": 7893,
|
||
},
|
||
"n2_n3_egress": {"via_proxy": {"ip": "1.2.3.4"}},
|
||
}
|
||
a = mh._compute_path_id(np)
|
||
b = mh._compute_path_id(np)
|
||
assert a and a.startswith("path_")
|
||
assert a == b
|
||
assert mh._compute_path_id({"missing": True}) is None
|
||
|
||
|
||
def test_sms_source_meta_reads_repo_catalog():
|
||
sms = mh._sms_source_meta()
|
||
assert isinstance(sms, dict)
|
||
assert "active_n" in sms
|
||
# 仓库 channels.json：5sim active，SMS-Activate shutdown → 通常单点
|
||
if sms.get("active_n") == 1:
|
||
assert sms.get("single_source") is True
|
||
assert "5sim" in str(sms.get("primary") or "")
|
||
|
||
|
||
def test_build_chain_status_sms_single_and_breaks(tmp_path=None):
|
||
import tempfile
|
||
td = Path(tmp_path) if tmp_path else Path(tempfile.mkdtemp())
|
||
old = mh._POOL_QUOTA_HIST
|
||
mh._POOL_QUOTA_HIST = td / "quota.jsonl"
|
||
try:
|
||
hist = mh._POOL_QUOTA_HIST
|
||
hist.parent.mkdir(parents=True, exist_ok=True)
|
||
# 两笔历史制造 burn（_record 会再写一行）
|
||
hist.write_text(
|
||
json.dumps({"epoch": 1000.0, "quota": 20, "total": 30}) + "\n"
|
||
+ json.dumps({"epoch": 4600.0, "quota": 5, "total": 30}) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
pmeta = mh._record_pool_quota(5, 30, ts="2026-07-13T00:00:00Z")
|
||
assert pmeta.get("samples", 0) >= 2
|
||
|
||
sms = {
|
||
"active_n": 1,
|
||
"watch_n": 2,
|
||
"active_ids": ["5sim-guest"],
|
||
"single_source": True,
|
||
"primary": "5sim-guest",
|
||
}
|
||
summary = {
|
||
"codex_total": 10,
|
||
"codex_quota": 5,
|
||
"codex_exhausted": 3,
|
||
"codex_dead": 2,
|
||
"live_probes_ok": 3,
|
||
"live_probes_fail": 0,
|
||
"free_endpoints": 2,
|
||
"tor_socks": True,
|
||
"tor_active": 1,
|
||
"guishi_channels": 4,
|
||
"proxies_alive": 0,
|
||
"miling_need_register": 0,
|
||
}
|
||
net_path = {
|
||
"ok": True,
|
||
"primary": {
|
||
"name": "clash-meta-9090",
|
||
"ok": True,
|
||
"mode": "rule",
|
||
"selected": "node-a",
|
||
"mixed_port": 7893,
|
||
},
|
||
"n2_n3_egress": {"via_proxy": {"ip": "9.9.9.9"}, "same_as_direct": False},
|
||
"n4_n5_isolation": {"risk": "ok"},
|
||
"system_proxy": {"enabled": True},
|
||
"headline": "ok",
|
||
}
|
||
chain = mh._build_chain_status(
|
||
summary=summary,
|
||
miling={"summary": {}},
|
||
blackmarket={"skus": [], "demand": [], "channels": []},
|
||
pool={},
|
||
net_path=net_path,
|
||
relays={"counts": {"n": 5, "ok": 5, "down": 0}},
|
||
pool_meta=pmeta,
|
||
sms_meta=sms,
|
||
)
|
||
assert chain.get("schema") == "chain_v2"
|
||
assert chain.get("path_id")
|
||
raw = next(r for r in chain["rings"] if r["id"] == "raw")
|
||
assert raw["level"] == "warn"
|
||
assert raw.get("reason") == "sms_single_source"
|
||
reasons = {b.get("reason") for b in (chain.get("breaks") or [])}
|
||
assert "sms_single_source" in reasons
|
||
assert chain.get("top_break")
|
||
finally:
|
||
mh._POOL_QUOTA_HIST = old
|
||
|
||
|
||
def test_face_discount_range_and_fixed():
|
||
import importlib.util
|
||
from pathlib import Path
|
||
p = Path(__file__).resolve().parents[1] / "scripts/manhuang/probe-ai-account-quotes.py"
|
||
spec = importlib.util.spec_from_file_location("ai_quotes2", p)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
assert spec and spec.loader
|
||
spec.loader.exec_module(mod)
|
||
r = mod.face_discount_fields("Apple iTunes Gift Card 1-100 USD (USA)", 1.81)
|
||
assert r["face_kind"] == "range"
|
||
assert r["discount"] is None
|
||
assert r["discount_note"] == "multi_denom_listing_min_price"
|
||
assert abs(r["discount_at_max"] - 0.0181) < 1e-6
|
||
f = mod.face_discount_fields("Apple Gift Card 100 USD", 85.0)
|
||
assert f["face_kind"] == "fixed"
|
||
assert abs(f["discount"] - 0.85) < 1e-6
|
||
|
||
|
||
def test_ggsel_price_wme_parser():
|
||
"""内嵌 price_wme JSON 可被抽成礼品卡市价。"""
|
||
import importlib.util
|
||
from pathlib import Path
|
||
p = Path(__file__).resolve().parents[1] / "scripts/manhuang/probe-ai-account-quotes.py"
|
||
spec = importlib.util.spec_from_file_location("ai_quotes", p)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
assert spec and spec.loader
|
||
spec.loader.exec_module(mod)
|
||
html = (
|
||
'{"id_goods":111,"url":"apple-itunes-gift-card-1-100-usd-usa-111",'
|
||
'"is_active":true,"name":"Apple iTunes Gift Card 1-100 USD (USA)",'
|
||
'"price_wme":"1.81","cnt_sell":10}'
|
||
'{"id_goods":222,"url":"turkey-itunes-222",'
|
||
'"name":"iTunes Gift Card 10-5000 TL Turkey AUTO","price_wme":"0.22"}'
|
||
'{"id_goods":333,"url":"apple-id-usa-333",'
|
||
'"name":"[US] USA Apple ID Solo Access","price_wme":"1.48"}'
|
||
)
|
||
us = mod.parse_ggsel_catalog(
|
||
html, r"(gift\s*card|iTunes).{0,40}(USA|US\b|USD)|Apple iTunes Gift Card.{0,20}USD"
|
||
)
|
||
assert any(abs(q["price"] - 1.81) < 0.01 for q in us), us
|
||
assert not any("Apple ID" in (q.get("title") or "") for q in us)
|
||
tr = mod.parse_ggsel_catalog(
|
||
html, r"(gift\s*card|iTunes).{0,40}(Turkey|TRY|TL\b)|Turkey.{0,20}(gift|iTunes)"
|
||
)
|
||
assert any(abs(q["price"] - 0.22) < 0.01 for q in tr), tr
|
||
|
||
|
||
def test_pay_msrp_only_reason():
|
||
pay = mh._pay_rail_meta(
|
||
quotes_skus=[
|
||
{"id": "giftcard_us", "tier": "pay", "best": {"kind": "msrp", "price": 100, "channel": "官方", "currency": "USD"}},
|
||
{"id": "virtual_card", "tier": "pay", "best": {"kind": "msrp", "price": 5, "channel": "official-msrp", "currency": "USD"}},
|
||
{"id": "sms_us", "tier": "pay", "best": {"kind": "market", "price": 0.1, "channel": "5sim guest", "currency": "USD"}},
|
||
]
|
||
)
|
||
assert pay.get("market_n") == 0
|
||
assert pay.get("msrp_n") == 2
|
||
assert pay.get("reason") == "pay_msrp_only"
|
||
ids = {r["id"] for r in pay.get("rails") or []}
|
||
assert "sms_us" not in ids
|
||
|
||
|
||
def test_pool_empty_is_crisis():
|
||
chain = mh._build_chain_status(
|
||
summary={
|
||
"codex_total": 12,
|
||
"codex_quota": 0,
|
||
"live_probes_ok": 1,
|
||
"live_probes_fail": 0,
|
||
"free_endpoints": 0,
|
||
"tor_socks": True,
|
||
"tor_active": 1,
|
||
},
|
||
miling={"summary": {}},
|
||
blackmarket={"skus": [], "demand": [], "channels": []},
|
||
pool={},
|
||
net_path={
|
||
"ok": True,
|
||
"primary": {"ok": True, "mode": "rule", "selected": "n", "name": "c", "mixed_port": 1},
|
||
"n4_n5_isolation": {"risk": "ok"},
|
||
"system_proxy": {"enabled": True},
|
||
},
|
||
sms_meta={"active_n": 2, "single_source": False, "watch_n": 0, "active_ids": ["a", "b"]},
|
||
pool_meta={},
|
||
)
|
||
assert chain["mode"] == "crisis_pool"
|
||
assert chain["crisis"] is True
|
||
pool = next(r for r in chain["rings"] if r["id"] == "pool")
|
||
assert pool["level"] == "err"
|
||
assert pool.get("reason") == "pool_empty"
|
