Spaces:
Paused
Paused
File size: 5,441 Bytes
3b9a70a 05bf6da 3b9a70a 5aff1af 3b9a70a 05bf6da 3b9a70a 5aff1af 3b9a70a bce9200 05bf6da bce9200 5aff1af bce9200 05bf6da bce9200 5aff1af bce9200 05bf6da | 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 | from autoteam import manager
class _FakeChatGPT:
def __init__(self):
self.browser = True
self.started = 0
self.stopped = 0
def start(self):
self.browser = True
self.started += 1
def stop(self):
self.browser = False
self.stopped += 1
class _FakeMailClient:
def login(self):
return None
def test_cmd_fill_tries_other_reusable_accounts_before_creating_new(monkeypatch):
import autoteam.config as config
chatgpt = _FakeChatGPT()
count_values = iter([2, 3])
events = []
monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", False)
monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
monkeypatch.setattr(manager, "CloudMailClient", lambda: _FakeMailClient())
monkeypatch.setattr(manager, "get_team_member_count", lambda _chatgpt: next(count_values))
monkeypatch.setattr(
manager,
"get_standby_accounts",
lambda: [
{"email": "old-1@example.com", "_quota_recovered": True},
{"email": "old-2@example.com", "_quota_recovered": True},
],
)
def fake_reinvite(_chatgpt, _mail, acc):
events.append(("reinvite", acc["email"]))
return acc["email"] == "old-2@example.com"
monkeypatch.setattr(manager, "reinvite_account", fake_reinvite)
monkeypatch.setattr(
manager,
"create_new_account",
lambda _chatgpt, _mail: events.append(("create", None)) or True,
)
monkeypatch.setattr(manager, "sync_to_cpa", lambda: events.append(("sync", None)))
monkeypatch.setattr(manager, "cmd_status", lambda: events.append(("status", None)))
manager.cmd_fill(target=3)
assert events == [
("reinvite", "old-1@example.com"),
("reinvite", "old-2@example.com"),
("sync", None),
("status", None),
]
assert chatgpt.stopped == 1
def test_cmd_fill_skips_google_accounts_during_auto_reuse(monkeypatch):
import autoteam.config as config
chatgpt = _FakeChatGPT()
count_values = iter([2, 3])
events = []
monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", False)
monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
monkeypatch.setattr(manager, "CloudMailClient", lambda: _FakeMailClient())
monkeypatch.setattr(manager, "get_team_member_count", lambda _chatgpt: next(count_values))
monkeypatch.setattr(
manager,
"get_standby_accounts",
lambda: [
{"email": "bubblehuntr@gmail.com", "_quota_recovered": True},
{"email": "old-2@example.com", "_quota_recovered": True},
],
)
def fake_reinvite(_chatgpt, _mail, acc):
events.append(("reinvite", acc["email"]))
return True
monkeypatch.setattr(manager, "reinvite_account", fake_reinvite)
monkeypatch.setattr(
manager,
"create_new_account",
lambda _chatgpt, _mail: events.append(("create", None)) or True,
)
monkeypatch.setattr(manager, "sync_to_cpa", lambda: events.append(("sync", None)))
monkeypatch.setattr(manager, "cmd_status", lambda: events.append(("status", None)))
manager.cmd_fill(target=3)
assert events == [
("reinvite", "old-2@example.com"),
("sync", None),
("status", None),
]
assert chatgpt.stopped == 1
def test_auto_reuse_skip_reason_detects_google_provider_and_gmail():
assert manager._auto_reuse_skip_reason({"email": "bubblehuntr@gmail.com"}) == "Google 登录账号暂不支持自动复用"
assert (
manager._auto_reuse_skip_reason({"email": "user@example.com", "login_provider": "google"})
== "Google 登录账号暂不支持自动复用"
)
assert manager._auto_reuse_skip_reason({"email": "user@example.com"}) is None
def test_cmd_fill_passes_direct_parallel_and_releases_failed_validation(monkeypatch):
import autoteam.config as config
chatgpt = _FakeChatGPT()
counts = iter([2, 2])
events = []
monkeypatch.setattr(config, "ROTATE_SKIP_REUSE", True)
monkeypatch.setattr(manager, "ChatGPTTeamAPI", lambda: chatgpt)
monkeypatch.setattr(manager, "CloudMailClient", lambda: _FakeMailClient())
monkeypatch.setattr(manager, "get_team_member_count", lambda _chatgpt: next(counts))
monkeypatch.setattr(manager, "get_standby_accounts", lambda: [])
monkeypatch.setattr(
manager,
"create_new_account",
lambda _chatgpt, _mail, *, parallel=None: events.append(("create", parallel)) or "new@example.com",
)
monkeypatch.setattr(manager, "_validate_managed_account_operational", lambda *_args, **_kwargs: False)
monkeypatch.setattr(
manager,
"remove_from_team",
lambda _chatgpt, email, *, return_status=False, **_kwargs: events.append(("remove", email)) or "removed",
)
monkeypatch.setattr(
manager,
"update_account",
lambda email, **kwargs: events.append(("update", email, kwargs.get("status"), kwargs.get("_reason"))),
)
monkeypatch.setattr(manager, "sync_to_cpa", lambda: events.append(("sync", None)))
monkeypatch.setattr(manager, "cmd_status", lambda: events.append(("status", None)))
manager.cmd_fill(target=3, direct_parallel=3)
assert ("create", 3) in events
assert ("remove", "new@example.com") in events
assert ("update", "new@example.com", manager.STATUS_STANDBY, "fill_new_account_not_ready") in events
|