Spaces:
Paused
Paused
File size: 14,275 Bytes
a011d67 90694b4 a011d67 90694b4 1f4cfcb 90694b4 a011d67 1f4cfcb a011d67 1f4cfcb a011d67 1f4cfcb a011d67 1f4cfcb a011d67 1f4cfcb a011d67 1f4cfcb a011d67 1f4cfcb a011d67 1f4cfcb a011d67 1f4cfcb a011d67 90694b4 | 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 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 | import inspect
import threading
import pytest
from autoteam import api, chatgpt_api, codex_auth, invite, manager
def test_launch_browser_stops_playwright_when_browser_launch_fails(tmp_path, monkeypatch):
class FakePlaywright:
def __init__(self):
self.stopped = False
self.chromium = self
def launch(self, **_kwargs):
raise RuntimeError("proxy launch failed")
def stop(self):
self.stopped = True
class FakeSyncPlaywright:
def __init__(self, playwright):
self._playwright = playwright
def start(self):
return self._playwright
fake_playwright = FakePlaywright()
monkeypatch.setattr(chatgpt_api, "SCREENSHOT_DIR", tmp_path)
monkeypatch.setattr(chatgpt_api, "get_playwright_launch_options", lambda: {"proxy": {"server": "http://proxy"}})
monkeypatch.setattr(chatgpt_api, "sync_playwright", lambda: FakeSyncPlaywright(fake_playwright))
client = chatgpt_api.ChatGPTTeamAPI()
with pytest.raises(RuntimeError, match="proxy launch failed"):
client._launch_browser()
assert fake_playwright.stopped is True
assert client.playwright is None
assert client.browser is None
assert client.context is None
assert client.page is None
def test_launch_browser_uses_unified_context_options(monkeypatch):
captured = {}
class FakePage:
def close(self):
captured["closed_page"] = True
class FakeContext:
def close(self):
captured["closed_context"] = True
def new_page(self):
return FakePage()
class FakePlaywright:
def __init__(self):
self.chromium = self
def launch(self, **_kwargs):
return self
def new_context(self, **kwargs):
captured["kwargs"] = kwargs
return FakeContext()
def stop(self):
captured["stopped"] = True
class FakeSyncPlaywright:
def start(self):
return FakePlaywright()
monkeypatch.setattr(chatgpt_api, "sync_playwright", lambda: FakeSyncPlaywright())
monkeypatch.setattr(chatgpt_api, "get_playwright_launch_options", lambda: {})
monkeypatch.setattr(chatgpt_api, "get_playwright_context_options", lambda: {"viewport": {"width": 100, "height": 200}})
client = chatgpt_api.ChatGPTTeamAPI()
client._launch_browser()
assert captured["kwargs"] == {"viewport": {"width": 100, "height": 200}}
def test_chatgpt_stop_closes_page_context_browser_and_playwright():
calls = []
class FakeClosable:
def __init__(self, name):
self.name = name
def close(self):
calls.append(f"close:{self.name}")
class FakePlaywright:
def stop(self):
calls.append("stop:playwright")
client = chatgpt_api.ChatGPTTeamAPI()
client.page = FakeClosable("page")
client.context = FakeClosable("context")
client.browser = FakeClosable("browser")
client.playwright = FakePlaywright()
client.stop()
client.stop()
assert calls == ["close:page", "close:context", "close:browser", "stop:playwright"]
assert client.page is None
assert client.context is None
assert client.browser is None
assert client.playwright is None
def test_start_with_session_stops_partially_started_browser_session():
calls = []
class FakeClosable:
def __init__(self, name):
self.name = name
def close(self):
calls.append(f"close:{self.name}")
class FakePlaywright:
def stop(self):
calls.append("stop:playwright")
client = chatgpt_api.ChatGPTTeamAPI()
def fail_after_browser_started(_session_token):
client.page = FakeClosable("page")
client.context = FakeClosable("context")
client.browser = FakeClosable("browser")
client.playwright = FakePlaywright()
raise RuntimeError("token fetch failed")
client._start_browser_session = fail_after_browser_started
with pytest.raises(RuntimeError, match="token fetch failed"):
client.start_with_session("session-token", "account-id", require_browser=True)
assert calls == ["close:page", "close:context", "close:browser", "stop:playwright"]
assert client.page is None
assert client.context is None
assert client.browser is None
assert client.playwright is None
def test_api_fetch_browser_fallback_cleans_partial_browser_session():
calls = []
class FakeClosable:
def __init__(self, name):
self.name = name
def close(self):
calls.append(f"close:{self.name}")
class FakePlaywright:
def stop(self):
calls.append("stop:playwright")
client = chatgpt_api.ChatGPTTeamAPI()
client.session_token = "session-token"
client.http_transport = FakeClosable("transport")
def fail_after_browser_started(_session_token):
client.page = FakeClosable("page")
client.context = FakeClosable("context")
client.browser = FakeClosable("browser")
client.playwright = FakePlaywright()
raise RuntimeError("cloudflare fallback failed")
client._start_browser_session = fail_after_browser_started
with pytest.raises(RuntimeError, match="cloudflare fallback failed"):
client._ensure_browser_session()
assert calls == [
"close:transport",
"close:page",
"close:context",
"close:browser",
"stop:playwright",
]
assert client.http_transport is None
assert client.page is None
assert client.context is None
assert client.browser is None
assert client.playwright is None
def test_complete_registration_closes_page_context_browser_when_invite_registration_raises(monkeypatch):
calls = []
captured = {}
class FakeClosable:
def __init__(self, name):
self.name = name
def close(self):
calls.append(f"close:{self.name}")
class FakeContext(FakeClosable):
def __init__(self):
super().__init__("context")
def new_page(self):
return FakeClosable("page")
class FakeBrowser(FakeClosable):
def __init__(self):
super().__init__("browser")
def new_context(self, **kwargs):
captured["kwargs"] = kwargs
return FakeContext()
class FakeChromium:
def launch(self, **_kwargs):
return FakeBrowser()
class FakePlaywright:
chromium = FakeChromium()
class FakeSyncPlaywright:
def __enter__(self):
return FakePlaywright()
def __exit__(self, *_args):
return False
def fail_register(*_args, **_kwargs):
raise RuntimeError("registration crashed")
monkeypatch.setattr(manager, "sync_playwright", lambda: FakeSyncPlaywright())
monkeypatch.setattr(manager, "get_playwright_context_options", lambda: {"locale": "zh-CN"})
monkeypatch.setattr(invite, "register_with_invite", fail_register)
with pytest.raises(RuntimeError, match="registration crashed"):
manager._complete_registration(
"child@example.com",
"password",
"https://invite.example",
mail_client=object(),
)
assert calls == ["close:page", "close:context", "close:browser"]
assert captured["kwargs"] == {"locale": "zh-CN"}
def test_register_direct_once_cleans_browser_context_on_unhandled_page_error(monkeypatch):
calls = []
captured = {}
class FakePage:
url = "about:blank"
def goto(self, *_args, **_kwargs):
raise RuntimeError("navigation exploded")
def close(self):
calls.append("close:page")
class FakeContext:
def new_page(self):
return FakePage()
def close(self):
calls.append("close:context")
class FakeBrowser:
def new_context(self, **kwargs):
captured["kwargs"] = kwargs
return FakeContext()
def close(self):
calls.append("close:browser")
class FakeChromium:
def launch(self, **_kwargs):
return FakeBrowser()
class FakePlaywright:
chromium = FakeChromium()
class FakeSyncPlaywright:
def __enter__(self):
return FakePlaywright()
def __exit__(self, *_args):
return False
monkeypatch.setattr(manager, "sync_playwright", lambda: FakeSyncPlaywright())
monkeypatch.setattr(manager, "get_playwright_context_options", lambda: {"timezone_id": "Asia/Shanghai"})
with pytest.raises(RuntimeError, match="navigation exploded"):
manager._register_direct_once(object(), "child@example.com", "password")
assert calls == ["close:page", "close:context", "close:browser"]
assert captured["kwargs"] == {"timezone_id": "Asia/Shanghai"}
def test_registration_and_oauth_paths_use_unified_playwright_cleanup():
assert "browser.close()" not in inspect.getsource(manager._complete_registration)
assert "browser.close()" not in inspect.getsource(manager._register_direct_once)
assert "browser.close()" not in inspect.getsource(invite.run)
assert "browser.close()" not in inspect.getsource(codex_auth.login_codex_via_browser)
assert "close_playwright_objects" in inspect.getsource(manager._register_direct_once)
assert "close_playwright_objects" in inspect.getsource(codex_auth.login_codex_via_browser)
assert "get_playwright_context_options()" in inspect.getsource(chatgpt_api.ChatGPTTeamAPI._launch_browser)
assert "get_playwright_context_options()" in inspect.getsource(invite.run)
assert "get_playwright_context_options()" in inspect.getsource(codex_auth.login_codex_via_browser)
assert "get_playwright_context_options()" in inspect.getsource(manager._complete_registration)
assert "get_playwright_context_options()" in inspect.getsource(manager._register_direct_once)
assert "viewport={\"width\": 1280, \"height\": 800}" not in inspect.getsource(chatgpt_api.ChatGPTTeamAPI._launch_browser)
assert "user_agent=\"Mozilla/5.0" not in inspect.getsource(chatgpt_api.ChatGPTTeamAPI._launch_browser)
def test_session_codex_auth_flow_stops_chatgpt_when_page_start_fails(monkeypatch):
instances = []
class FakeContext:
def new_page(self):
raise RuntimeError("new page failed")
class FakeChatGPTTeamAPI:
def __init__(self):
self.context = FakeContext()
self.stopped = False
instances.append(self)
def start_with_session(self, *_args, **_kwargs):
return None
def stop(self):
self.stopped = True
monkeypatch.setattr("autoteam.chatgpt_api.ChatGPTTeamAPI", FakeChatGPTTeamAPI)
flow = codex_auth.SessionCodexAuthFlow(
email="admin@example.com",
session_token="session-token",
account_id="account-id",
workspace_name="workspace",
)
with pytest.raises(RuntimeError, match="new page failed"):
flow.start()
assert len(instances) == 1
assert instances[0].stopped is True
assert flow.chatgpt is None
assert flow.page is None
def test_post_admin_login_start_stops_api_when_begin_login_fails(monkeypatch):
instances = []
class FakeChatGPTTeamAPI:
def __init__(self):
self.stopped = False
instances.append(self)
def begin_admin_login(self, _email):
raise RuntimeError("proxy launch failed")
def stop(self):
self.stopped = True
monkeypatch.setattr(api, "_playwright_lock", threading.Lock())
monkeypatch.setattr(api, "_admin_login_api", None)
monkeypatch.setattr(api, "_admin_login_step", None)
monkeypatch.setattr(api._pw_executor, "run", lambda func, *args, **kwargs: func(*args, **kwargs))
monkeypatch.setattr("autoteam.chatgpt_api.ChatGPTTeamAPI", FakeChatGPTTeamAPI)
with pytest.raises(api.HTTPException) as exc:
api.post_admin_login_start(api.AdminEmailParams(email="admin@example.com"))
assert exc.value.status_code == 400
assert "proxy launch failed" in str(exc.value.detail)
assert len(instances) == 1
assert instances[0].stopped is True
assert api._admin_login_api is None
assert api._playwright_lock.locked() is False
def test_get_team_members_stops_chatgpt_when_start_fails(monkeypatch):
instances = []
class FakeChatGPTTeamAPI:
def __init__(self):
self.stopped = False
instances.append(self)
def start(self):
raise RuntimeError("http proxy failed")
def stop(self):
self.stopped = True
monkeypatch.setattr(api, "_playwright_lock", threading.Lock())
monkeypatch.setattr(api._pw_executor, "run", lambda func, *args, **kwargs: func(*args, **kwargs))
monkeypatch.setattr("autoteam.admin_state.get_admin_session_token", lambda: "session")
monkeypatch.setattr("autoteam.admin_state.get_chatgpt_account_id", lambda: "acc-1")
monkeypatch.setattr("autoteam.chatgpt_api.ChatGPTTeamAPI", FakeChatGPTTeamAPI)
with pytest.raises(api.HTTPException) as exc:
api.get_team_members()
assert exc.value.status_code == 502
assert "http proxy failed" in str(exc.value.detail)
assert len(instances) == 1
assert instances[0].stopped is True
assert api._playwright_lock.locked() is False
def test_parse_playwright_probe_stdout_uses_last_json_line():
assert api._parse_playwright_probe_stdout("noise\n{\"count\": 4}\n") == {"count": 4}
def test_auto_check_team_member_count_returns_minus_one_when_probe_fails(monkeypatch):
calls = []
def fake_probe(*_args, **_kwargs):
calls.append(True)
raise TimeoutError("probe timeout")
monkeypatch.setattr(api, "_run_playwright_probe", fake_probe)
assert api._auto_check_team_member_count(timeout_seconds=0.01, retries=2) == -1
assert len(calls) == 2
def test_auto_check_team_member_count_reads_probe_count(monkeypatch):
monkeypatch.setattr(api, "_run_playwright_probe", lambda *_args, **_kwargs: {"count": "5"})
assert api._auto_check_team_member_count(timeout_seconds=0.01, retries=1) == 5
|