Spaces:
Paused
Paused
File size: 7,515 Bytes
ceeba19 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 | import threading
import pytest
from fastapi import HTTPException
from autoteam import api
@pytest.fixture(autouse=True)
def _reset_main_codex_state():
yield
api._admin_login_api = None
api._admin_login_step = None
api._main_codex_flow = None
api._main_codex_step = None
api._main_codex_action = None
if api._playwright_lock.locked():
api._playwright_lock.release()
def test_finish_admin_login_refreshes_local_main_auth_without_remote_sync(monkeypatch):
events = []
class FakeAdminLoginAPI:
def complete_admin_login(self):
events.append("admin_complete")
return {
"email": "owner@example.com",
"session_token": "session-token",
"account_id": "acc-1",
"workspace_name": "Idapro",
}
def stop(self):
events.append("admin_stop")
lock = threading.Lock()
assert lock.acquire(blocking=False) is True
def fake_refresh_main_auth_file():
events.append("refresh_main_auth")
return {"email": "owner@example.com", "auth_file": "/tmp/codex-main-acc-1.json", "plan_type": "team"}
def fail_remote_sync(_filepath):
raise AssertionError("admin login must not sync main Codex auth to remote targets")
monkeypatch.setattr(api, "_playwright_lock", lock)
monkeypatch.setattr(api, "_admin_login_api", FakeAdminLoginAPI())
monkeypatch.setattr(api, "_admin_login_step", "workspace_required")
monkeypatch.setattr(api, "_main_codex_flow", None)
monkeypatch.setattr(api, "_main_codex_step", None)
monkeypatch.setattr(api, "_main_codex_action", None)
monkeypatch.setattr(api._pw_executor, "run", lambda func, *args, **kwargs: func(*args, **kwargs))
monkeypatch.setattr("autoteam.codex_auth.refresh_main_auth_file", fake_refresh_main_auth_file)
monkeypatch.setattr("autoteam.sync_targets.sync_main_codex_to_configured_targets", fail_remote_sync)
monkeypatch.setattr(
"autoteam.admin_state.get_admin_state_summary",
lambda: {
"configured": False,
"email": "",
"password_saved": False,
"session_present": False,
"account_id": "",
"workspace_name": "",
},
)
result = api._finish_admin_login({"step": "completed"})
assert result["status"] == "completed"
assert result["codex"] == {"in_progress": False, "step": None, "action": None}
assert result["info"]["main_auth"] == {
"email": "owner@example.com",
"auth_file": "/tmp/codex-main-acc-1.json",
"plan_type": "team",
}
assert api._admin_login_api is None
assert api._admin_login_step is None
assert lock.locked() is False
assert events == ["admin_complete", "admin_stop", "refresh_main_auth"]
def test_finish_main_codex_flow_returns_login_message(monkeypatch):
events = []
class FakeMainCodexFlow:
def complete(self):
events.append("complete")
return {
"email": "owner@example.com",
"auth_file": "/tmp/codex-main-acc-1.json",
"plan_type": "team",
}
def stop(self):
events.append("stop")
lock = threading.Lock()
assert lock.acquire(blocking=False) is True
monkeypatch.setattr(api, "_playwright_lock", lock)
monkeypatch.setattr(api, "_main_codex_flow", FakeMainCodexFlow())
monkeypatch.setattr(api, "_main_codex_step", "code_required")
monkeypatch.setattr(api, "_main_codex_action", "login")
monkeypatch.setattr(api._pw_executor, "run", lambda func, *args, **kwargs: func(*args, **kwargs))
result = api._finish_main_codex_flow()
assert result == {
"status": "completed",
"message": "主号 Codex 已登录",
"codex": {"in_progress": False, "step": None, "action": None},
"info": {
"email": "owner@example.com",
"auth_file": "/tmp/codex-main-acc-1.json",
"plan_type": "team",
},
}
assert api._main_codex_flow is None
assert api._main_codex_step is None
assert api._main_codex_action is None
assert lock.locked() is False
assert events == ["complete", "stop"]
def test_post_main_codex_login_starts_login_flow(monkeypatch):
events = []
lock = threading.Lock()
monkeypatch.setattr(api, "_playwright_lock", lock)
monkeypatch.setattr(api, "_main_codex_flow", None)
monkeypatch.setattr(api, "_main_codex_step", None)
monkeypatch.setattr(api, "_main_codex_action", None)
def fake_start(action="sync"):
events.append(action)
if lock.locked():
lock.release()
return "completed", {
"status": "completed",
"message": "主号 Codex 已登录",
"codex": {"in_progress": False, "step": None, "action": None},
"info": {"auth_file": "/tmp/codex-main-acc-1.json"},
}
monkeypatch.setattr(api, "_start_main_codex_flow", fake_start)
result = api.post_main_codex_login()
assert events == ["login"]
assert result["message"] == "主号 Codex 已登录"
assert lock.locked() is False
def test_main_codex_sync_requires_enabled_target(monkeypatch):
monkeypatch.setenv("SYNC_TARGET_CPA", "false")
monkeypatch.setenv("SYNC_TARGET_SUB2API", "false")
monkeypatch.delenv("CPA_URL", raising=False)
monkeypatch.delenv("CPA_KEY", raising=False)
monkeypatch.delenv("SUB2API_URL", raising=False)
monkeypatch.delenv("SUB2API_EMAIL", raising=False)
monkeypatch.delenv("SUB2API_PASSWORD", raising=False)
with pytest.raises(HTTPException) as exc_info:
api._require_sync_target_configs("同步主号 Codex")
assert exc_info.value.status_code == 400
assert "启用至少一个远端同步目标" in exc_info.value.detail
def test_post_main_codex_delete_remote_files_returns_deleted_names(monkeypatch):
monkeypatch.setattr(api, "_require_sync_target_configs", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
"autoteam.sync_targets.get_enabled_sync_targets",
lambda env=None: ["cpa", "sub2api"],
)
monkeypatch.setattr(
"autoteam.sync_targets.delete_main_codex_from_configured_targets",
lambda: {
"cpa": {"deleted": ["codex-main-acc-1.json"], "count": 1},
"sub2api": {"deleted": ["sub2api-codex-main-acc-1.json"], "count": 1},
},
)
result = api.post_main_codex_delete_remote_files()
assert result == {
"message": "已从 CPA + Sub2API 删除 2 个主号认证文件",
"deleted": ["codex-main-acc-1.json", "sub2api-codex-main-acc-1.json"],
"results": {
"cpa": {"deleted": ["codex-main-acc-1.json"], "count": 1},
"sub2api": {"deleted": ["sub2api-codex-main-acc-1.json"], "count": 1},
},
}
def test_post_main_codex_delete_cpa_requires_enabled_target(monkeypatch):
def fake_require(*_args, **_kwargs):
raise HTTPException(
status_code=400,
detail="删除主号 Codex 远端文件 前请先在配置面板启用至少一个远端同步目标（CPA 或 Sub2API）",
)
monkeypatch.setattr(api, "_require_sync_target_configs", fake_require)
with pytest.raises(HTTPException) as exc_info:
api.post_main_codex_delete_cpa()
assert exc_info.value.status_code == 400
assert "启用至少一个远端同步目标" in exc_info.value.detail
|