879 lines
36 KiB
Python
879 lines
36 KiB
Python
"""
|
||
优先级同步测试套件 — 分组内竞争逻辑
|
||
|
||
核心规则（测试覆盖）：
|
||
- 竞争分组键 = imported_target_group_id or group_id（老数据 fallback）
|
||
- 只有同一竞争分组内账号数 > 1 时才更新 priority / 发通知
|
||
- 不同分组各 1 个账号：不调用 update_account，不发通知
|
||
- 同一目标分组多账号：组内按倍率升序独立排序，priority 从 1 开始，每档间隔 10
|
||
- 两个目标分组各有多账号：彼此独立，每组内 priority 都从 1 开始，每档间隔 10
|
||
- 老数据 imported_target_group_id=NULL：fallback group_id，不报错
|
||
"""
|
||
import json
|
||
from datetime import datetime, timezone
|
||
import pytest
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import sessionmaker
|
||
from sqlalchemy.pool import StaticPool
|
||
|
||
from app.database import Base
|
||
from app.models.upstream import Upstream
|
||
from app.models.website import Website, WebsiteSyncLog
|
||
from app.models.upstream_key import UpstreamGeneratedKey
|
||
from app.models.snapshot import UpstreamRateSnapshot
|
||
from app.services.website_sync import (
|
||
build_rate_priority_map,
|
||
sync_account_priorities_for_upstream,
|
||
)
|
||
from app.services.website_client import Sub2ApiWebsiteClient
|
||
|
||
|
||
@pytest.fixture()
|
||
def db_session():
|
||
engine = create_engine(
|
||
"sqlite://",
|
||
connect_args={"check_same_thread": False},
|
||
poolclass=StaticPool,
|
||
)
|
||
Base.metadata.create_all(bind=engine)
|
||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||
db = TestingSessionLocal()
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
Base.metadata.drop_all(bind=engine)
|
||
|
||
|
||
# ── 辅助 ─────────────────────────────────────────────────────────────────────
|
||
|
||
def _make_snapshot(db, upstream_id: int, groups: dict):
|
||
"""插入快照 groups = {group_id: rate}。"""
|
||
snap = {
|
||
"groups": {
|
||
gid: {"group_name": gid, "rate": rate}
|
||
for gid, rate in groups.items()
|
||
}
|
||
}
|
||
db.add(UpstreamRateSnapshot(
|
||
upstream_id=upstream_id,
|
||
snapshot_json=json.dumps(snap),
|
||
captured_at=datetime.now(timezone.utc),
|
||
))
|
||
db.commit()
|
||
|
||
|
||
def _make_key(db, upstream_id, group_id, key_name, key_value,
|
||
website_id, account_id, imported_target_group_id=None):
|
||
k = UpstreamGeneratedKey(
|
||
upstream_id=upstream_id,
|
||
group_id=group_id,
|
||
group_name=group_id,
|
||
key_name=key_name,
|
||
key_value=key_value,
|
||
imported_website_id=website_id,
|
||
imported_account_id=account_id,
|
||
imported_target_group_id=imported_target_group_id,
|
||
)
|
||
db.add(k)
|
||
db.commit()
|
||
return k
|
||
|
||
|
||
class MockClient:
|
||
"""可注入的 update_account 记录器。"""
|
||
_calls: list # 由子类/工厂绑定
|
||
|
||
def __init__(self, **kwargs):
|
||
pass # calls 由工厂方法注入
|
||
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
|
||
def update_account(self, account_id, data):
|
||
type(self)._shared_calls.append((account_id, data))
|
||
|
||
|
||
def make_mock_client(calls: list):
|
||
"""返回一个与 calls 列表绑定的 MockClient 类。"""
|
||
class _Client(MockClient):
|
||
_shared_calls = calls
|
||
return _Client
|
||
|
||
|
||
# ── 用例 ─────────────────────────────────────────────────────────────────────
|
||
|
||
def test_no_update_when_different_groups_single_account_each(db_session, monkeypatch):
|
||
"""不同分组各 1 个账号 → 无竞争 → 不调用 update_account，不发通知。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
u2 = Upstream(name="U2", base_url="http://u2")
|
||
db_session.add_all([w, u1, u2])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1); db_session.refresh(u2)
|
||
|
||
_make_snapshot(db_session, u1.id, {"G1": 1.0})
|
||
_make_snapshot(db_session, u2.id, {"G2": 2.0})
|
||
|
||
# 不同分组，各 1 个账号，imported_target_group_id=None (fallback group_id)
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w.id, "A1")
|
||
_make_key(db_session, u2.id, "G2", "K2", "V2", w.id, "A2")
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
results = sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
# 无竞争分组 → 直接返回空列表，不调用 update_account
|
||
assert update_calls == [], f"不应有更新调用，实际：{update_calls}"
|
||
assert results == []
|
||
|
||
# 不写日志
|
||
logs = db_session.query(WebsiteSyncLog).filter(WebsiteSyncLog.website_id == w.id).all()
|
||
assert logs == []
|
||
|
||
|
||
def test_same_target_group_two_accounts_updated(db_session, monkeypatch):
|
||
"""同一目标分组 2 个账号 → 有竞争 → 按倍率更新 priority。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
u2 = Upstream(name="U2", base_url="http://u2")
|
||
db_session.add_all([w, u1, u2])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1); db_session.refresh(u2)
|
||
|
||
_make_snapshot(db_session, u1.id, {"G1": 1.0})
|
||
_make_snapshot(db_session, u2.id, {"G2": 2.0})
|
||
|
||
# 两个账号都属于目标分组 "TG1"
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u2.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
results = sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
assert len(update_calls) == 2
|
||
priority_map = {aid: data["priority"] for aid, data in update_calls}
|
||
# G1 rate=1.0 → priority=1（低倍率优先）；G2 rate=2.0 → priority=11
|
||
assert priority_map["A1"] == 1
|
||
assert priority_map["A2"] == 11
|
||
|
||
# 写了日志
|
||
log = db_session.query(WebsiteSyncLog).filter(WebsiteSyncLog.website_id == w.id).first()
|
||
assert log is not None
|
||
assert log.algorithm == "priority_sync"
|
||
|
||
|
||
def test_two_target_groups_independent_priority(db_session, monkeypatch):
|
||
"""两个目标分组各有多账号 → 每组内独立从 1 开始排序，不互相影响。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u1])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1)
|
||
|
||
_make_snapshot(db_session, u1.id, {
|
||
"G1": 1.0, # → TG1 中排 priority=1
|
||
"G2": 2.0, # → TG1 中排 priority=11
|
||
"G3": 0.5, # → TG2 中排 priority=1
|
||
"G4": 3.0, # → TG2 中排 priority=11
|
||
})
|
||
|
||
# TG1: G1(1.0), G2(2.0)
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
# TG2: G3(0.5), G4(3.0)
|
||
_make_key(db_session, u1.id, "G3", "K3", "V3", w.id, "A3", imported_target_group_id="TG2")
|
||
_make_key(db_session, u1.id, "G4", "K4", "V4", w.id, "A4", imported_target_group_id="TG2")
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
priority_map = {aid: data["priority"] for aid, data in update_calls}
|
||
assert len(update_calls) == 4
|
||
|
||
# TG1 内部：G1(1.0)→p1, G2(2.0)→p11
|
||
assert priority_map["A1"] == 1
|
||
assert priority_map["A2"] == 11
|
||
# TG2 内部：G3(0.5)→p1, G4(3.0)→p11（独立从 1 开始）
|
||
assert priority_map["A3"] == 1
|
||
assert priority_map["A4"] == 11
|
||
|
||
|
||
def test_old_data_null_target_group_fallback(db_session, monkeypatch):
|
||
"""老数据 imported_target_group_id=NULL → fallback group_id，不报错。
|
||
|
||
两个账号同 group_id（极端边界），视为同一竞争分组 → 有竞争 → 更新。
|
||
"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
u2 = Upstream(name="U2", base_url="http://u2")
|
||
db_session.add_all([w, u1, u2])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1); db_session.refresh(u2)
|
||
|
||
_make_snapshot(db_session, u1.id, {"GSHARED": 1.0})
|
||
_make_snapshot(db_session, u2.id, {"GSHARED": 2.0})
|
||
|
||
# 老数据：imported_target_group_id=None，两账号 group_id 相同
|
||
_make_key(db_session, u1.id, "GSHARED", "K1", "V1", w.id, "A1", imported_target_group_id=None)
|
||
_make_key(db_session, u2.id, "GSHARED", "K2", "V2", w.id, "A2", imported_target_group_id=None)
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
# 不应抛异常
|
||
results = sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
# 同 group_id="GSHARED" → 竞争分组，两账号都更新
|
||
assert len(update_calls) == 2
|
||
|
||
|
||
def test_null_target_group_backfills_from_remote_account_group(db_session, monkeypatch):
|
||
"""老导入数据 target_group_id=NULL，但远端账号有单一 group_ids → 回填后参与同组重排。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
u2 = Upstream(name="U2", base_url="http://u2")
|
||
u3 = Upstream(name="U3", base_url="http://u3")
|
||
db_session.add_all([w, u1, u2, u3])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1); db_session.refresh(u2); db_session.refresh(u3)
|
||
|
||
_make_snapshot(db_session, u1.id, {"G2": 0.06})
|
||
_make_snapshot(db_session, u2.id, {"G19": 0.07})
|
||
_make_snapshot(db_session, u3.id, {"G21": 0.05})
|
||
|
||
legacy = _make_key(db_session, u1.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id=None)
|
||
_make_key(db_session, u2.id, "G19", "K19", "V19", w.id, "A19", imported_target_group_id="TG1")
|
||
_make_key(db_session, u3.id, "G21", "K21", "V21", w.id, "A21", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
|
||
class AccountListMockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def list_accounts(self):
|
||
return [
|
||
{"id": "A2", "group_ids": ["TG1"]},
|
||
{"id": "A19", "group_ids": ["TG1"]},
|
||
{"id": "A21", "group_ids": ["TG1"]},
|
||
]
|
||
def update_account(self, account_id, data):
|
||
update_calls.append((account_id, data))
|
||
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", AccountListMockClient)
|
||
|
||
sync_account_priorities_for_upstream(db_session, u1.id, website_id=w.id)
|
||
|
||
updated = {account_id: payload["priority"] for account_id, payload in update_calls}
|
||
assert updated == {"A21": 1, "A2": 11, "A19": 21}
|
||
db_session.refresh(legacy)
|
||
assert legacy.imported_target_group_id == "TG1"
|
||
|
||
|
||
def test_single_account_in_mixed_website(db_session, monkeypatch):
|
||
"""同一网站：一个目标分组有 2 账号（竞争），另一个目标分组只有 1 账号（不参与）。
|
||
|
||
只有竞争分组的 2 个账号被更新，单账号分组不调用 update_account。
|
||
"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u1])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1)
|
||
|
||
_make_snapshot(db_session, u1.id, {"G1": 1.0, "G2": 2.0, "G3": 3.0})
|
||
|
||
# TG1: G1 + G2 → 竞争
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
# TG2: 只有 G3 → 单账号，跳过
|
||
_make_key(db_session, u1.id, "G3", "K3", "V3", w.id, "A3", imported_target_group_id="TG2")
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
updated_ids = {c[0] for c in update_calls}
|
||
assert "A3" not in updated_ids, "单账号分组不应被更新"
|
||
assert updated_ids == {"A1", "A2"}
|
||
|
||
|
||
def test_missing_rate_skips_entire_competitive_group(db_session, monkeypatch):
|
||
"""竞争分组内所有账号均无快照倍率 → 有效账号 < 2 → 整组跳过，不调用 update_account。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u1])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1)
|
||
|
||
# 故意不插快照 → raw_rate_map 为空
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
results = sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
assert update_calls == [], "无快照倍率时不应调用 update_account"
|
||
assert results == []
|
||
|
||
|
||
def test_partial_missing_rate_sufficient_accounts_still_updates(db_session, monkeypatch):
|
||
"""竞争分组内 3 账号，1 个无快照倍率，剩余 2 个有倍率 → 仍有竞争，2 个有倍率的账号正常更新。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u1])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1)
|
||
|
||
# G1、G2 有快照，G3 没有快照
|
||
_make_snapshot(db_session, u1.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G3", "K3", "V3", w.id, "A3", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
updated = {c[0]: c[1]["priority"] for c in update_calls}
|
||
# A3 无快照 → 不参与排序，不被更新
|
||
assert "A3" not in updated
|
||
# A1(G1, rate=1.0) → priority=1；A2(G2, rate=2.0) → priority=11
|
||
assert updated["A1"] == 1
|
||
assert updated["A2"] == 11
|
||
|
||
|
||
def test_priority_sync_log_structure(db_session, monkeypatch):
|
||
"""日志写入格式验证（需要竞争分组才会写日志）。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u1])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1)
|
||
|
||
_make_snapshot(db_session, u1.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
# 同一目标分组 2 账号 → 有日志
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
class LoggingMockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def update_account(self, account_id, data): pass
|
||
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", LoggingMockClient)
|
||
|
||
sync_account_priorities_for_upstream(db_session, u1.id)
|
||
|
||
log = db_session.query(WebsiteSyncLog).filter(WebsiteSyncLog.website_id == w.id).first()
|
||
assert log is not None
|
||
assert log.algorithm == "priority_sync"
|
||
|
||
data = json.loads(log.source_rates_json)
|
||
# 第一项是 priority_map 元数据
|
||
assert data[0]["_meta"] == "priority_map"
|
||
# 后续项是账号结果
|
||
account_ids = {item["account_id"] for item in data[1:] if "account_id" in item}
|
||
assert {"A1", "A2"} == account_ids
|
||
|
||
|
||
# ── 保留：build_rate_priority_map 单元测试（供初始导入使用）────────────────
|
||
|
||
def test_priority_sync_cross_upstream_group(db_session):
|
||
"""build_rate_priority_map：相同 group_id 不同上游不互相覆盖。"""
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
u2 = Upstream(name="U2", base_url="http://u2")
|
||
db_session.add_all([u1, u2])
|
||
db_session.commit()
|
||
db_session.refresh(u1); db_session.refresh(u2)
|
||
|
||
_make_snapshot(db_session, u1.id, {"VIP": 1.0})
|
||
_make_snapshot(db_session, u2.id, {"VIP": 2.0})
|
||
|
||
priority_map = build_rate_priority_map(db_session, {u1.id, u2.id})
|
||
|
||
assert priority_map[f"{u1.id}:VIP"] == 1
|
||
assert priority_map[f"{u2.id}:VIP"] == 11
|
||
assert len(priority_map) == 2
|
||
|
||
|
||
def test_import_auto_priority_by_rate(db_session, monkeypatch):
|
||
"""初始导入时 auto_priority_by_rate=True 按全局倍率分配 priority。"""
|
||
from app.routers.websites import import_upstream_keys_as_accounts
|
||
from app.schemas.website import ImportAccountsRequest
|
||
|
||
w = Website(name="W1", base_url="http://w1", enabled=True, auth_config_json="{}",
|
||
groups_endpoint="/groups", group_update_endpoint="/groups/{id}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
u2 = Upstream(name="U2", base_url="http://u2")
|
||
db_session.add_all([w, u1, u2])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u1); db_session.refresh(u2)
|
||
|
||
_make_snapshot(db_session, u1.id, {"G1": 2.0})
|
||
_make_snapshot(db_session, u2.id, {"G2": 1.0})
|
||
|
||
k1 = UpstreamGeneratedKey(upstream_id=u1.id, group_id="G1", group_name="G1",
|
||
key_name="K1", key_value="V1")
|
||
k2 = UpstreamGeneratedKey(upstream_id=u2.id, group_id="G2", group_name="G2",
|
||
key_name="K2", key_value="V2")
|
||
db_session.add_all([k1, k2])
|
||
db_session.commit()
|
||
|
||
created_accounts = []
|
||
|
||
class MockImportClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def create_account(self, body):
|
||
created_accounts.append(body)
|
||
return {"id": f"remote-{len(created_accounts)}", "name": body["name"]}
|
||
def extract_id(self, data): return data["id"]
|
||
def account_exists(self, aid): return False
|
||
|
||
monkeypatch.setattr("app.routers.websites._client", lambda website: MockImportClient())
|
||
monkeypatch.setattr("app.services.website_client.Sub2ApiWebsiteClient", MockImportClient)
|
||
|
||
req = ImportAccountsRequest(
|
||
upstream_key_ids=[k1.id, k2.id],
|
||
target_group_map={},
|
||
auto_priority_by_rate=True,
|
||
priority=10,
|
||
account_name_prefix="test",
|
||
default_platform="openai",
|
||
)
|
||
|
||
import_upstream_keys_as_accounts(w.id, req, db_session)
|
||
|
||
assert len(created_accounts) == 2
|
||
# G2 rate=1.0 → priority 1；G1 rate=2.0 → priority 11
|
||
p1 = next(a["priority"] for a in created_accounts if "G1" in a["name"])
|
||
p2 = next(a["priority"] for a in created_accounts if "G2" in a["name"])
|
||
assert p2 == 1
|
||
assert p1 == 11
|
||
|
||
|
||
def test_reorder_priority_endpoint_scopes_to_current_website(db_session, monkeypatch):
|
||
"""手动重排只影响当前网站，不误改同一上游导入到其他网站的账号。"""
|
||
from app.routers.websites import reorder_account_priorities
|
||
from app.schemas.website import ReorderPriorityRequest
|
||
|
||
w1 = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
w2 = Website(name="W2", base_url="http://w2", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u1 = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w1, w2, u1])
|
||
db_session.commit()
|
||
db_session.refresh(w1); db_session.refresh(w2); db_session.refresh(u1)
|
||
|
||
_make_snapshot(db_session, u1.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
_make_key(db_session, u1.id, "G1", "K1", "V1", w1.id, "W1A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G2", "K2", "V2", w1.id, "W1A2", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G1", "K3", "V3", w2.id, "W2A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u1.id, "G2", "K4", "V4", w2.id, "W2A2", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
monkeypatch.setattr(
|
||
"app.services.website_sync.Sub2ApiWebsiteClient",
|
||
make_mock_client(update_calls),
|
||
)
|
||
|
||
response = reorder_account_priorities(
|
||
w1.id,
|
||
ReorderPriorityRequest(upstream_id=u1.id),
|
||
db_session,
|
||
)
|
||
|
||
updated = {account_id: payload["priority"] for account_id, payload in update_calls}
|
||
assert response.success is True
|
||
assert updated == {"W1A1": 1, "W1A2": 11}
|
||
|
||
|
||
def test_priority_sync_notification_grouped_by_target_group():
|
||
"""测试 DingTalk Markdown 通知按目标分组正确聚合。"""
|
||
from app.utils.dingtalk import format_dingtalk_priority_changed
|
||
|
||
updates = [
|
||
{
|
||
"target_group_id": "TG1",
|
||
"target_group_name": "PRO智能分组",
|
||
"account_id": "A123",
|
||
"account_name": "SmartUp-PRO-xxx",
|
||
"source_group_id": "sg1",
|
||
"source_group_name": "28",
|
||
"source_rate": 0.8,
|
||
"old_priority": 11,
|
||
"new_priority": 1,
|
||
"status": "success",
|
||
"message": "success",
|
||
},
|
||
{
|
||
"target_group_id": "TG1",
|
||
"target_group_name": "PRO智能分组",
|
||
"account_id": "A456",
|
||
"account_name": "SmartUp-Claude-xxx",
|
||
"source_group_id": "sg2",
|
||
"source_group_name": "claude max渠道",
|
||
"source_rate": 1.2,
|
||
"old_priority": 21,
|
||
"new_priority": 11,
|
||
"status": "failed",
|
||
"message": "xxx",
|
||
},
|
||
{
|
||
"target_group_id": "TG2",
|
||
"target_group_name": "Basic智能分组",
|
||
"account_id": "A789",
|
||
"account_name": "SmartUp-Basic-xxx",
|
||
"source_group_id": "sg3",
|
||
"source_group_name": "basic渠道",
|
||
"source_rate": 2.0,
|
||
"old_priority": "未知",
|
||
"new_priority": 31,
|
||
"status": "success",
|
||
"message": "success",
|
||
}
|
||
]
|
||
|
||
msg = format_dingtalk_priority_changed("SmartSite", "Upstream1", "2026-07-01 09:00:00", updates)
|
||
text = msg["markdown"]["text"]
|
||
|
||
# 验证分组标题
|
||
assert "目标分组：PRO智能分组" in text
|
||
assert "目标分组：Basic智能分组" in text
|
||
|
||
# 验证账号明细
|
||
assert "- ✅ SmartUp-PRO-xxx（账号ID A123）：old=11 → new=1，来源分组=28，倍率=0.8" in text
|
||
assert "- ❌ SmartUp-Claude-xxx（账号ID A456）：old=21 → new=11，来源分组=claude max渠道，倍率=1.2，原因=xxx" in text
|
||
assert "- ✅ SmartUp-Basic-xxx（账号ID A789）：old=未知 → new=31，来源分组=basic渠道，倍率=2.0" in text
|
||
|
||
|
||
def test_priority_sync_fetches_remote_accounts_and_backfills_info(db_session, monkeypatch):
|
||
"""验证优先级同步能获取并回填远端账号的名称和旧优先级。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u)
|
||
|
||
_make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
_make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
|
||
class CustomMockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def list_accounts(self):
|
||
return [
|
||
{"id": "A1", "name": "SmartUp-A1-Name", "priority": 15},
|
||
{"id": "A2", "name": "SmartUp-A2-Name", "priority": 25},
|
||
]
|
||
def get_groups(self, *args, **kwargs):
|
||
return [
|
||
{"id": "TG1", "name": "TG1-Readable-Group-Name"}
|
||
]
|
||
def update_account(self, account_id, data):
|
||
update_calls.append((account_id, data))
|
||
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", CustomMockClient)
|
||
|
||
results = sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id)
|
||
assert len(results) == 2
|
||
|
||
# 校验 site_results 中的丰富信息是否填充
|
||
res_a1 = next(r for r in results if r["account_id"] == "A1")
|
||
assert res_a1["account_name"] == "SmartUp-A1-Name"
|
||
assert res_a1["old_priority"] == 15
|
||
assert res_a1["new_priority"] == 1
|
||
assert res_a1["source_rate"] == 1.0
|
||
assert res_a1["target_group_id"] == "TG1"
|
||
assert res_a1["target_group_name"] == "TG1-Readable-Group-Name"
|
||
|
||
res_a2 = next(r for r in results if r["account_id"] == "A2")
|
||
assert res_a2["account_name"] == "SmartUp-A2-Name"
|
||
assert res_a2["old_priority"] == 25
|
||
assert res_a2["new_priority"] == 11
|
||
assert res_a2["source_rate"] == 2.0
|
||
|
||
# 验证本地 DB 中行的 imported_target_group_name 被正确更新写入
|
||
row1 = db_session.query(UpstreamGeneratedKey).filter_by(imported_account_id="A1").first()
|
||
assert row1.imported_target_group_name == "TG1-Readable-Group-Name"
|
||
|
||
|
||
def test_priority_sync_graceful_on_remote_list_failure(db_session, monkeypatch):
|
||
"""验证 list_accounts 抛异常时，优先级同步依然成功更新，且各项降级为账号 ID 和未知。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u)
|
||
|
||
_make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
_make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
_make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
class CustomMockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def list_accounts(self):
|
||
raise Exception("Remote server error 500")
|
||
def update_account(self, account_id, data):
|
||
pass
|
||
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", CustomMockClient)
|
||
|
||
results = sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id)
|
||
assert len(results) == 2
|
||
|
||
# 校验降级逻辑
|
||
res_a1 = next(r for r in results if r["account_id"] == "A1")
|
||
assert res_a1["account_name"] == "A1"
|
||
assert res_a1["old_priority"] is None
|
||
assert res_a1["new_priority"] == 1
|
||
assert res_a1["status"] == "success"
|
||
|
||
|
||
def test_priority_sync_stale_account_cleanup_on_list_success(db_session, monkeypatch):
|
||
"""验证 priority sync 时，若远端账号列表中缺失某个导入账号，能自愈清理本地标记并不影响其他竞争账号。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u)
|
||
|
||
_make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0, "G3": 3.0})
|
||
|
||
# A1(K1), A2(K2) 与 A3(K3) 竞争
|
||
k1 = _make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
k2 = _make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
k3 = _make_key(db_session, u.id, "G3", "K3", "V3", w.id, "A3", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
|
||
class MockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def list_accounts(self):
|
||
# 只有 A1 和 A2，缺失 A3
|
||
return [
|
||
{"id": "A1", "name": "SmartUp-A1"},
|
||
{"id": "A2", "name": "SmartUp-A2"},
|
||
]
|
||
def get_groups(self, *a, **kw):
|
||
return [{"id": "TG1", "name": "TG1-Group"}]
|
||
def update_account(self, account_id, data):
|
||
update_calls.append((account_id, data))
|
||
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", MockClient)
|
||
|
||
results = sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id)
|
||
|
||
# A1 和 A2 正常竞争更新，A3 被自愈清除不调用 update_account
|
||
assert len(update_calls) == 2
|
||
priority_map = {aid: data["priority"] for aid, data in update_calls}
|
||
assert priority_map["A1"] == 1
|
||
assert priority_map["A2"] == 11
|
||
|
||
res_a3 = next(r for r in results if r["account_id"] == "A3")
|
||
assert res_a3["status"] == "stale"
|
||
assert "目标账号已不存在，已清除导入标记" in res_a3["message"]
|
||
|
||
db_session.refresh(k3)
|
||
assert k3.imported_account_id is None
|
||
assert k3.imported_website_id is None
|
||
assert k3.status == "created"
|
||
|
||
|
||
def test_priority_sync_stale_account_single_remaining_skips_update(db_session, monkeypatch):
|
||
"""验证 priority sync 时，若清除失效账号后某分组只剩一个有效账号，该分组跳过更新不调用 update_account。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u)
|
||
|
||
_make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
# A1 与 A2 竞争
|
||
k1 = _make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
k2 = _make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
|
||
class MockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def list_accounts(self):
|
||
# 只有 A1，缺失 A2
|
||
return [{"id": "A1", "name": "SmartUp-A1"}]
|
||
def get_groups(self, *a, **kw):
|
||
return [{"id": "TG1", "name": "TG1-Group"}]
|
||
def update_account(self, account_id, data):
|
||
update_calls.append((account_id, data))
|
||
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", MockClient)
|
||
|
||
results = sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id)
|
||
|
||
# 竞争分组只剩 A1 一个有效账号 -> 跳过不更新，update_account 不应被调用
|
||
assert update_calls == []
|
||
res_a2 = next(r for r in results if r["account_id"] == "A2")
|
||
assert res_a2["status"] == "stale"
|
||
db_session.refresh(k2)
|
||
assert k2.imported_account_id is None
|
||
|
||
|
||
def test_priority_sync_no_cleanup_on_list_failure(db_session, monkeypatch):
|
||
"""验证 list_accounts 失败时，不清理本地导入标记，仍尝试调用 update_account。"""
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u)
|
||
|
||
_make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
k1 = _make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
k2 = _make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
update_calls = []
|
||
|
||
class MockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def list_accounts(self):
|
||
raise Exception("Remote list error")
|
||
def update_account(self, account_id, data):
|
||
update_calls.append((account_id, data))
|
||
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", MockClient)
|
||
|
||
sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id)
|
||
|
||
# list_accounts 失败，不做清理，继续正常对 A1 和 A2 尝试更新
|
||
assert len(update_calls) == 2
|
||
db_session.refresh(k1)
|
||
db_session.refresh(k2)
|
||
assert k1.imported_account_id == "A1"
|
||
assert k2.imported_account_id == "A2"
|
||
|
||
|
||
def test_update_account_404_error_custom_msg_and_fallback_cleanup(db_session, monkeypatch):
|
||
"""验证 update_account 抛出 404 错误时转换为友好文案并触发自愈清除。"""
|
||
from app.services.website_client import WebsiteError
|
||
w = Website(name="W1", base_url="http://w1", enabled=True,
|
||
auth_config_json="{}", timeout_seconds=30)
|
||
u = Upstream(name="U1", base_url="http://u1")
|
||
db_session.add_all([w, u])
|
||
db_session.commit()
|
||
db_session.refresh(w); db_session.refresh(u)
|
||
|
||
_make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0})
|
||
|
||
k1 = _make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1")
|
||
k2 = _make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1")
|
||
|
||
import httpx
|
||
|
||
class MockClient:
|
||
def __init__(self, **kwargs): pass
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): pass
|
||
def list_accounts(self):
|
||
# 模拟 list 失败，以便依赖 update_account 兜底清理
|
||
raise Exception("list failed")
|
||
def update_account(self, account_id, data):
|
||
# A2 触发 404 错误
|
||
if account_id == "A2":
|
||
req = httpx.Request("PUT", f"http://w1/accounts/{account_id}")
|
||
resp = httpx.Response(404, request=req)
|
||
raise WebsiteError("目标网站接口不存在") from httpx.HTTPStatusError("404", request=req, response=resp)
|
||
|
||
def get_groups(self, *a, **kw):
|
||
return []
|
||
|
||
# 验证单独调用 update_account 404 时的定制错误文案
|
||
client = Sub2ApiWebsiteClient(base_url="http://w1", api_prefix="", auth_type="api_key", auth_config={})
|
||
# Mock _request in client
|
||
def mock_request(method, path, body=None):
|
||
req = httpx.Request(method, path)
|
||
resp = httpx.Response(404, request=req)
|
||
raise WebsiteError("目标网站接口不存在") from httpx.HTTPStatusError("404", request=req, response=resp)
|
||
monkeypatch.setattr(client, "_request", mock_request)
|
||
|
||
with pytest.raises(WebsiteError) as exc_info:
|
||
client.update_account("A2", {"priority": 1})
|
||
assert "目标账号 A2 不存在或已被删除" in str(exc_info.value)
|
||
|
||
# 验证同步流程中的 404 兜底清理
|
||
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", MockClient)
|
||
results = sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id)
|
||
|
||
res_a2 = next(r for r in results if r["account_id"] == "A2")
|
||
assert res_a2["status"] == "stale"
|
||
assert "目标账号已不存在，已清除导入标记" in res_a2["message"]
|
||
|
||
db_session.refresh(k2)
|
||
assert k2.imported_account_id is None
|
||
assert k2.status == "created"
|