import json 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.upstream_key import UpstreamGeneratedKey from app.models.website import Website, WebsiteGroupBinding from app.routers.websites import organize_website_groups @pytest.fixture() def db_session(): engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) # Ensure all models are created from app.models import admin_user, upstream, snapshot, webhook_config, notification_log, custom_page, website, revoked_token, upstream_key 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 test_organize_groups_full_scenarios(db_session, monkeypatch): """测试一键整理分组的各种核心场景: 1. 已有 Key 未导入时,创建目标账号并绑定目标分组 2. 已导入且目标账号存在时,不重复创建(若未绑定目标分组则补齐绑定) 3. 已导入但目标账号不存在时,清理旧标记并重建 4. 绑定关系存在但没有对应 Key 时,返回 missing_key 5. 多个目标分组、多上游分组绑定时,正确分类 """ # 1. 建立测试 Website 和 Upstream w = Website( name="W1", site_type="sub2api", 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) # 2. 建立绑定关系 # 绑定 1:目标分组 TG1 ↔ 上游分组 G1(有 Key 未导入),G2(没有 Key — 对应 missing_key) b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([ {"upstream_id": u1.id, "group_id": "G1"}, {"upstream_id": u1.id, "group_id": "G2"}, ]), enabled=True, platform="openai", ) # 绑定 2:目标分组 TG2 ↔ 上游分组 G3(已导入且账号仍存在),G4(已导入但账号已删除) b2 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG2", target_group_name="TG2-Group", source_groups_json=json.dumps([ {"upstream_id": u1.id, "group_id": "G3"}, {"upstream_id": u1.id, "group_id": "G4"}, ]), enabled=True, platform="openai", ) db_session.add_all([b1, b2]) db_session.commit() # 3. 建立上游 Key 记录 # G1: 未导入 Key k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-Name", key_name="Key-G1", key_value="sk-g1-secret", status="created", ) # G3: 已导入且账号存在 k3 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G3", group_name="G3-Name", key_name="Key-G3", key_value="sk-g3-secret", status="imported", imported_website_id=w.id, imported_account_id="ACC-G3", imported_target_group_id="TG2", ) # G4: 已导入但账号已删除 k4 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G4", group_name="G4-Name", key_name="Key-G4", key_value="sk-g4-secret", status="imported", imported_website_id=w.id, imported_account_id="ACC-G4-DELETED", imported_target_group_id="TG2", ) db_session.add_all([k1, k3, k4]) db_session.commit() # 4. Mock Sub2ApiWebsiteClient 交互 created_accounts = [] updated_accounts = [] class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [ {"id": "TG1", "name": "TG1-Group", "platform": "openai"}, {"id": "TG2", "name": "TG2-Group", "platform": "openai"}, ] def list_accounts(self): # ACC-G3 仍存在,ACC-G4-DELETED 不在此列表中代表已被删除 # 另外,我们假设 ACC-G3 目前绑定的 group_ids 是 [999],即不包含 TG2 (或整型数值) return [ { "id": "ACC-G3", "name": "SmartUp-G3-ACC", "group_ids": [999], "platform": "openai", } ] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): # 模拟创建账号 acc_id = f"NEW-{body['name']}" new_acc = {"id": acc_id, "name": body["name"], "group_ids": body["group_ids"]} created_accounts.append(new_acc) return new_acc def update_account(self, account_id, body): # 模拟更新账号的分组绑定 updated_accounts.append((account_id, body)) return {"id": account_id, "group_ids": body.get("group_ids")} monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) # 5. 执行一键整理 response = organize_website_groups(wid=w.id, db=db_session) # 6. 断言结果 assert response.success is True # 整理完成:已创建 2 / 已存在已迁移 1 / 缺少 Key 1(G4 无 link 且不复用 legacy 字段,直接新建) assert "已创建 2" in response.message assert "已存在已迁移 1" in response.message assert "缺少 Key 1" in response.message items = response.items assert len(items) == 4 # 按 target_group_id 分类校验 # TG1 分组下的 G1 (已创建) item_g1 = next(item for item in items if item.target_group_id == "TG1" and item.source_group_id == "G1") assert item_g1.status == "created" assert item_g1.key_name == "Key-G1" assert item_g1.account_id.startswith("NEW-U1-") # TG1 分组下的 G2 (缺少 Key) item_g2 = next(item for item in items if item.target_group_id == "TG1" and item.source_group_id == "G2") assert item_g2.status == "missing_key" assert item_g2.message == "请先生成上游 Key" # TG2 分组下的 G3 (已存在并补齐) item_g3 = next(item for item in items if item.target_group_id == "TG2" and item.source_group_id == "G3") assert item_g3.status == "exists" assert "已迁移到目标分组" in item_g3.message # 校验是否触发了 update_account 补齐了 TG2 assert len(updated_accounts) == 1 assert updated_accounts[0][0] == "ACC-G3" # TG2 (或者其数值) 应该在更新后的 group_ids 中 assert "TG2" in updated_accounts[0][1]["group_ids"] or 999 in updated_accounts[0][1]["group_ids"] # TG2 分组下的 G4(无 link 且不复用 legacy,直接新建) item_g4 = next(item for item in items if item.target_group_id == "TG2" and item.source_group_id == "G4") assert item_g4.status == "created" assert item_g4.account_id.startswith("NEW-U1-") # 7. 检查数据库中 Key 记录的状态变化 db_session.refresh(k1) db_session.refresh(k3) db_session.refresh(k4) assert k1.imported_website_id == w.id assert k1.imported_account_id is not None assert k1.imported_target_group_id == "TG1" assert k1.status == "imported" assert k3.imported_target_group_id == "TG2" assert k4.imported_website_id == w.id assert k4.imported_account_id is not None assert k4.imported_target_group_id == "TG2" assert k4.status == "imported" def test_organize_groups_list_accounts_none_conservatively_skips(db_session, monkeypatch): """测试一键整理分组时,若远端账号列表拉取失败返回 None: - 已导入的账号不得被清除或重建,而应该保守跳过(状态记为 failed,提示无法校验)。 - 未导入的账号仍能正常导入创建。 """ w = Website( name="W1", site_type="sub2api", 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) # 绑定关系:目标分组 TG1 ↔ 上游分组 G1(未导入),G3(已导入) b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([ {"upstream_id": u1.id, "group_id": "G1", "group_name": "G1-SourceGroup"}, {"upstream_id": u1.id, "group_id": "G3", "group_name": "G3-SourceGroup"}, ]), enabled=True, platform="openai", ) db_session.add_all([b1]) db_session.commit() # k1: 未导入 Key k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-SourceGroup", key_name="Key-G1", key_value="sk-g1-secret", status="created", ) # k3: 已导入 Key k3 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G3", group_name="G3-SourceGroup", key_name="Key-G3", key_value="sk-g3-secret", status="imported", imported_website_id=w.id, imported_account_id="ACC-G3-EXISTING", imported_target_group_id="TG1", ) db_session.add_all([k1, k3]) db_session.commit() created_accounts = [] class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): # 模拟获取失败返回 None! return None def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): acc_id = f"NEW-{body['name']}" new_acc = {"id": acc_id, "name": body["name"], "group_ids": body["group_ids"]} created_accounts.append(new_acc) return new_acc monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) # 执行一键整理 response = organize_website_groups(wid=w.id, db=db_session) # 新行为:无 UpstreamKeyAccountLink 时不复用 legacy 字段,k3 也按新 Key 创建 assert response.success is True assert "已创建 2" in response.message items = response.items assert len(items) == 2 item_g1 = next(item for item in items if item.source_group_id == "G1") assert item_g1.status == "created" assert item_g1.account_id.startswith("NEW-U1-") assert item_g1.source_group_name == "G1-SourceGroup" item_g3 = next(item for item in items if item.source_group_id == "G3") assert item_g3.status == "created" db_session.refresh(k3) assert k3.imported_website_id == w.id assert k3.imported_account_id is not None def test_organize_groups_streaming_basic_event_sequence(db_session, monkeypatch): """流式一键整理:验证基本 NDJSON 事件序列、响应头、total_items 准确性。""" w = Website( name="W1", site_type="sub2api", 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) b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([ {"upstream_id": u1.id, "group_id": "G1"}, ]), enabled=True ) db_session.add(b1) k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-Name", key_name="Key-G1", key_value="sk-g1-secret", status="created", ) db_session.add(k1) db_session.commit() created_accounts = [] class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): acc_id = f"NEW-{body['name']}" new_acc = {"id": acc_id, "name": body["name"], "group_ids": body["group_ids"]} created_accounts.append(new_acc) return new_acc monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) from fastapi.testclient import TestClient from app.main import app from app.database import get_db from app.utils.auth import get_current_user app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_current_user] = lambda: None try: client = TestClient(app) resp = client.post(f"/api/websites/{w.id}/groups/organize/stream") assert resp.status_code == 200 assert resp.headers["content-type"] == "application/x-ndjson" assert resp.headers.get("cache-control") == "no-cache" assert resp.headers.get("x-accel-buffering") == "no" lines = [line if isinstance(line, str) else line.decode("utf-8") for line in resp.iter_lines() if line] parsed_events = [json.loads(line) for line in lines] # Event sequence: start → item → complete assert len(parsed_events) >= 2 assert parsed_events[0]["event"] == "start" assert parsed_events[0]["data"]["total_items"] == 1 item_events = [e for e in parsed_events if e["event"] == "item"] assert len(item_events) == 1 assert item_events[0]["data"]["source_group_id"] == "G1" assert item_events[0]["data"]["status"] == "created" complete_event = next(e for e in parsed_events if e["event"] == "complete") assert complete_event["data"]["success"] is True finally: app.dependency_overrides.clear() def test_organize_groups_streaming_409_concurrent_lock(db_session, monkeypatch): """流式一键整理:同一网站并发调用返回 409 错误。""" w = Website( name="W1", site_type="sub2api", base_url="http://w1", enabled=True, auth_config_json="{}", timeout_seconds=30 ) db_session.add(w) db_session.commit() db_session.refresh(w) # Mock to make the first call run long enough to trigger concurrency import time original_generator = None class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) from fastapi.testclient import TestClient from app.main import app from app.database import get_db from app.utils.auth import get_current_user app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_current_user] = lambda: None try: client = TestClient(app) # First request starts the task lock resp1 = client.post(f"/api/websites/{w.id}/groups/organize/stream") assert resp1.status_code == 200 # Second request while first is still "running" (lock not yet released) # But the first request already completed, so we need to simulate # by calling the task lock manually. from app.routers.websites import _try_start_task, _finish_task ok, _task_token = _try_start_task(w.id, "organize") assert ok try: resp2 = client.post(f"/api/websites/{w.id}/groups/organize/stream") assert resp2.status_code == 200 lines = [line if isinstance(line, str) else line.decode("utf-8") for line in resp2.iter_lines() if line] parsed_events = [json.loads(line) for line in lines] assert len(parsed_events) == 1 assert parsed_events[0]["event"] == "error" assert "正在执行中" in parsed_events[0]["data"]["message"] finally: _finish_task(w.id, "organize", _task_token) # Task lock released: third request should succeed resp3 = client.post(f"/api/websites/{w.id}/groups/organize/stream") assert resp3.status_code == 200 finally: app.dependency_overrides.clear() def test_organize_groups_streaming_non_sub2api_returns_error_event(db_session, monkeypatch): """流式一键整理:非 sub2api 网站返回 error 事件而非 HTTP 错误。""" w = Website( name="W1", site_type="other", base_url="http://w1", enabled=True, auth_config_json="{}", timeout_seconds=30 ) db_session.add(w) db_session.commit() db_session.refresh(w) from fastapi.testclient import TestClient from app.main import app from app.database import get_db from app.utils.auth import get_current_user app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_current_user] = lambda: None try: client = TestClient(app) resp = client.post(f"/api/websites/{w.id}/groups/organize/stream") assert resp.status_code == 200 lines = [line if isinstance(line, str) else line.decode("utf-8") for line in resp.iter_lines() if line] parsed_events = [json.loads(line) for line in lines] assert len(parsed_events) == 1 assert parsed_events[0]["event"] == "error" finally: app.dependency_overrides.clear() def test_organize_groups_streaming_total_items_matches_items(db_session, monkeypatch): """流式一键整理:total_items 应当等于实际产生的 item 数量。""" w = Website( name="W1", site_type="sub2api", 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) # Two bindings, each with one key b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G1"}]), enabled=True ) b2 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG2", target_group_name="TG2-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G2"}]), enabled=True ) db_session.add_all([b1, b2]) k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-Name", key_name="Key-G1", key_value="sk-g1", status="created", ) k2 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G2", group_name="G2-Name", key_name="Key-G2", key_value="sk-g2", status="created", ) db_session.add_all([k1, k2]) db_session.commit() class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [ {"id": "TG1", "name": "TG1-Group", "platform": "openai"}, {"id": "TG2", "name": "TG2-Group", "platform": "openai"}, ] def list_accounts(self): return [] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): return {"id": "NEW-" + body["name"], "name": body["name"], "group_ids": body["group_ids"]} monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) from fastapi.testclient import TestClient from app.main import app from app.database import get_db from app.utils.auth import get_current_user app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_current_user] = lambda: None try: client = TestClient(app) resp = client.post(f"/api/websites/{w.id}/groups/organize/stream") assert resp.status_code == 200 lines = [line if isinstance(line, str) else line.decode("utf-8") for line in resp.iter_lines() if line] parsed_events = [json.loads(line) for line in lines] start_event = next(e for e in parsed_events if e["event"] == "start") item_events = [e for e in parsed_events if e["event"] == "item"] assert start_event["data"]["total_items"] == len(item_events) finally: app.dependency_overrides.clear() def test_organize_groups_original_json_endpoint_still_works(db_session, monkeypatch): """一键整理 JSON 接口的回归测试:HTTP 状态码和错误处理仍然正确。""" # Non-sub2api → HTTP 400 w = Website( name="W1", site_type="other", base_url="http://w1", enabled=True, auth_config_json="{}", timeout_seconds=30 ) db_session.add(w) db_session.commit() db_session.refresh(w) from fastapi.testclient import TestClient from app.main import app from app.database import get_db from app.utils.auth import get_current_user app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_current_user] = lambda: None try: client = TestClient(app) resp = client.post(f"/api/websites/{w.id}/groups/organize") assert resp.status_code == 400 assert "只支持" in resp.json()["detail"] or "only sub2api" in resp.json()["detail"] finally: app.dependency_overrides.clear() def test_organize_groups_json_409_concurrent_lock(db_session, monkeypatch): """一键整理 JSON 接口:并发调用返回 HTTP 409。""" w = Website( name="W1", site_type="sub2api", base_url="http://w1", enabled=True, auth_config_json="{}", timeout_seconds=30, ) db_session.add(w) db_session.commit() db_session.refresh(w) from fastapi.testclient import TestClient from app.main import app from app.database import get_db from app.utils.auth import get_current_user app.dependency_overrides[get_db] = lambda: db_session app.dependency_overrides[get_current_user] = lambda: None try: client = TestClient(app) # Hold lock manually to simulate concurrent request from app.routers.websites import _try_start_task, _finish_task ok, _task_token = _try_start_task(w.id, "organize") assert ok try: resp = client.post(f"/api/websites/{w.id}/groups/organize") assert resp.status_code == 409 assert "正在执行中" in resp.json()["detail"] finally: _finish_task(w.id, "organize", _task_token) # Lock released: second request should succeed resp2 = client.post(f"/api/websites/{w.id}/groups/organize") assert resp2.status_code == 200 finally: app.dependency_overrides.clear() def test_organize_groups_force_alignment(db_session, monkeypatch): """测试强对齐场景: 2. 已存在账号同时有非 SmartUp 分组,整理后保留非 SmartUp 分组。 3. 已存在账号已经完全一致,整理不重复调用更新接口。 4. 多目标分组、多上游来源时,每个账号按自己的当前绑定关系对齐。 """ w = Website( name="W1", site_type="sub2api", 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) # 两个绑定关系: # 绑定 1:目标分组 TG1 ↔ G1 # 绑定 2:目标分组 TG2 ↔ G2 b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G1"}]), enabled=True ) b2 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG2", target_group_name="TG2-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G2"}]), enabled=True ) db_session.add_all([b1, b2]) db_session.commit() # 3. 建立 3 个 Key,对应 3 个不同的远端账号状态 # k1: 原本在 TG_old (旧 SmartUp 分组),但根据绑定 1 它现在应该在 TG1 k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-Group", key_name="Key-G1", key_value="sk-g1-secret", status="imported", imported_website_id=w.id, imported_account_id="ACC-G1", imported_target_group_id="TG_old", # 旧 SmartUp 分组 ) # k2: 已经在 TG2 (正确),且同时有非 SmartUp 分组 (999) 和另一个旧 SmartUp 分组 (TG1 - 应该被移除) k2 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G2", group_name="G2-Group", key_name="Key-G2", key_value="sk-g2-secret", status="imported", imported_website_id=w.id, imported_account_id="ACC-G2", imported_target_group_id="TG2", ) db_session.add_all([k1, k2]) db_session.commit() updated_accounts = [] class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [ {"id": "TG1", "name": "TG1-Group", "platform": "openai"}, {"id": "TG2", "name": "TG2-Group", "platform": "openai"}, ] def list_accounts(self): return [ { "id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG_old"], # 包含旧 SmartUp 分组 TG_old,没有 TG1 "platform": "openai", }, { "id": "ACC-G2", "name": "SmartUp-G2", "group_ids": ["TG2", 999, "TG1"], # TG2 是当前正确分组,999 是非托管分组,TG1 是旧托管分组(应移除) "platform": "openai", } ] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def update_account(self, account_id, body): updated_accounts.append((account_id, body)) return {"id": account_id, "group_ids": body.get("group_ids")} monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) # 执行一键整理 response = organize_website_groups(wid=w.id, db=db_session) assert response.success is True # ACC-G1 (G1) 的 group_ids 应变更为 ["TG1"] (TG2 被移除了) # ACC-G2 (G2) 的 group_ids 应变更为 [999, "TG2"] (TG1 被移除了,999 保留) assert len(updated_accounts) == 2 # 验证 ACC-G1 变更 up_g1 = next(item for item in updated_accounts if item[0] == "ACC-G1") assert set(str(g) for g in up_g1[1]["group_ids"]) == {"TG1"} # 验证 ACC-G2 变更 up_g2 = next(item for item in updated_accounts if item[0] == "ACC-G2") assert set(str(g) for g in up_g2[1]["group_ids"]) == {"999", "TG2"} # 验证本地 DB 的标记已被同步更新为正确的 TG1 / TG2 db_session.refresh(k1) db_session.refresh(k2) assert k1.imported_target_group_id == "TG1" assert k2.imported_target_group_id == "TG2" # 验证结果消息汇总包含 aligned 0, migrated 2 (因为两个都发生了迁移/更新) assert "已存在已迁移 2" in response.message # 让我们跑一次完全一致的再次整理,验证不触发 update_account (已对齐) updated_accounts.clear() class MockClientAligned(MockClient): def list_accounts(self): return [ { "id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai", }, { "id": "ACC-G2", "name": "SmartUp-G2", "group_ids": [999, "TG2"], "platform": "openai", } ] monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClientAligned) response_aligned = organize_website_groups(wid=w.id, db=db_session) assert response_aligned.success is True # 既然已经一致,不会触发 update_account assert len(updated_accounts) == 0 assert "已存在且已对齐 2" in response_aligned.message def test_organize_groups_corrects_mismatched_platform_instead_of_recreating(db_session, monkeypatch): """测试一键整理遇到已存在账号平台错配时,只调用 update_account() 修正平台,不创建新账号。""" from app.models.snapshot import UpstreamRateSnapshot w = Website( name="W1", site_type="sub2api", 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) # 绑定关系:目标分组 TG1 ↔ G1 b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G1"}]), enabled=True ) db_session.add(b1) # G1: 已导入,导入的目标分组为 TG1,账号为 ACC-G1 k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-Group", key_name="Key-G1", key_value="sk-g1-secret", status="imported", imported_website_id=w.id, imported_account_id="ACC-G1", imported_target_group_id="TG1", ) db_session.add(k1) # 写入快照指定平台为 anthropic snapshot = UpstreamRateSnapshot( upstream_id=u1.id, snapshot_json=json.dumps({ "groups": { "G1": { "group_name": "G1-Group", "rate": 0.1, "platform": "anthropic" } } }) ) db_session.add(snapshot) db_session.commit() updated_accounts = [] create_called = False mock_platform = "openai" class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): return [ { "id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": mock_platform, } ] def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): nonlocal create_called create_called = True return {"id": "NEW-ACC", "name": body["name"], "group_ids": body["group_ids"]} def update_account(self, account_id, body): nonlocal mock_platform updated_accounts.append((account_id, body)) if "platform" in body: mock_platform = body["platform"] return {"id": account_id, "platform": body.get("platform")} monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) response = organize_website_groups(wid=w.id, db=db_session) assert response.success is True assert not create_called # 新行为:平台以 binding 为准(openai),与远端账号一致 → 不对平台做任何修正 assert len(updated_accounts) == 0 assert response.items[0].status == "exists" assert "已存在" in response.items[0].message def _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn): """公用工厂:搭建平台错配场景并注入可定制的 list_accounts。""" from app.models.snapshot import UpstreamRateSnapshot w = Website( name="W-PlatChk", site_type="sub2api", base_url="http://wp", enabled=True, auth_config_json="{}", timeout_seconds=30, ) u1 = Upstream(name="U-PlatChk", base_url="http://up") db_session.add_all([w, u1]) db_session.commit() db_session.refresh(w) db_session.refresh(u1) b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG1", target_group_name="TG1-Group", source_groups_json=json.dumps([{"upstream_id": u1.id, "group_id": "G1"}]), enabled=True, platform="openai", ) db_session.add(b1) k1 = UpstreamGeneratedKey( upstream_id=u1.id, group_id="G1", group_name="G1-Group", key_name="Key-G1", key_value="sk-g1-secret", status="imported", imported_website_id=w.id, imported_account_id="ACC-G1", imported_target_group_id="TG1", ) db_session.add(k1) snapshot = UpstreamRateSnapshot( upstream_id=u1.id, snapshot_json=json.dumps({ "groups": { "G1": {"group_name": "G1-Group", "rate": 0.1, "platform": "anthropic"} } }), ) db_session.add(snapshot) db_session.commit() call_count = {"n": 0} class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [{"id": "TG1", "name": "TG1-Group", "platform": "openai"}] def list_accounts(self): call_count["n"] += 1 return list_accounts_fn(call_count["n"]) def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def update_account(self, account_id, body): return {"id": account_id} monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr( "app.routers.websites.sync_account_priorities_for_website", lambda db, wid: [] ) return w def test_organize_groups_platform_update_remote_silently_ignores(db_session, monkeypatch): """平台以 binding 为准 → openai 与远端一致,无需更新,直接 aligned。""" def list_accounts_fn(call_n): return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] w = _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn) response = organize_website_groups(wid=w.id, db=db_session) assert response.success is True assert response.items[0].status == "exists" def test_organize_groups_platform_update_list_returns_none(db_session, monkeypatch): """平台以 binding 为准 → openai 与远端一致,无需更新,直接 aligned。""" def list_accounts_fn(call_n): return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] w = _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn) response = organize_website_groups(wid=w.id, db=db_session) assert response.success is True assert response.items[0].status == "exists" def test_organize_groups_platform_update_account_missing_from_list(db_session, monkeypatch): """平台以 binding 为准 → openai 与远端一致,无需更新,直接 aligned。""" def list_accounts_fn(call_n): return [{"id": "ACC-G1", "name": "SmartUp-G1", "group_ids": ["TG1"], "platform": "openai"}] w = _seed_platform_mismatch(db_session, monkeypatch, list_accounts_fn) response = organize_website_groups(wid=w.id, db=db_session) assert response.success is True assert response.items[0].status == "exists" def test_one_key_multi_platform_two_bindings_creates_two_accounts(db_session, monkeypatch): """同一 Key 绑定到 openai 和 anthropic 两个目标分组 → 创建两个独立账号。""" from app.models.upstream_key import UpstreamKeyAccountLink w = Website(name="W-Multi", base_url="http://w-multi", enabled=True, auth_config_json="{}", timeout_seconds=30) u = Upstream(name="U-Multi", base_url="http://u-multi") db_session.add_all([w, u]) db_session.commit() k = UpstreamGeneratedKey( upstream_id=u.id, group_id="G1", group_name="G1", key_name="Key-Shared", key_value="sk-shared-secret", status="created", ) db_session.add(k) b1 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG-OpenAI", source_groups_json=json.dumps([{"upstream_id": u.id, "group_id": "G1"}]), enabled=True, platform="openai", ) b2 = WebsiteGroupBinding( website_id=w.id, target_group_id="TG-Anthropic", source_groups_json=json.dumps([{"upstream_id": u.id, "group_id": "G1"}]), enabled=True, platform="anthropic", ) db_session.add_all([b1, b2]) db_session.commit() created_accounts = [] class MockClient: def __init__(self, **kwargs): pass def __enter__(self): return self def __exit__(self, *a): pass def get_groups(self, *a, **kw): return [ {"id": "TG-OpenAI", "name": "OpenAI", "platform": "openai"}, {"id": "TG-Anthropic", "name": "Anthropic", "platform": "anthropic"}, ] def list_accounts(self): return created_accounts def extract_id(self, val): return val.get("id") if isinstance(val, dict) else str(val) def create_account(self, body): acc_id = f"ACC-{body['platform']}-{len(created_accounts)+1}" acc = {"id": acc_id, "name": body["name"], "group_ids": body["group_ids"], "platform": body["platform"]} created_accounts.append(acc) return acc monkeypatch.setattr("app.routers.websites.Sub2ApiWebsiteClient", MockClient) monkeypatch.setattr("app.routers.websites.sync_account_priorities_for_website", lambda db, wid: []) res1 = organize_website_groups(wid=w.id, db=db_session) assert res1.success is True assert len(created_accounts) == 2 assert {a["platform"] for a in created_accounts} == {"openai", "anthropic"} links = db_session.query(UpstreamKeyAccountLink).filter( UpstreamKeyAccountLink.website_id == w.id, UpstreamKeyAccountLink.status == "active" ).all() assert len(links) == 2 assert {l.platform for l in links} == {"openai", "anthropic"} assert all(l.upstream_key_id == k.id for l in links) created_before = len(created_accounts) res2 = organize_website_groups(wid=w.id, db=db_session) assert res2.success is True assert len(created_accounts) == created_before assert all(i.status == "exists" for i in res2.items)