mirror of
https://github.com/zc-zhangchen/any-auto-register.git
synced 2026-05-15 10:56:45 +08:00
334 lines
14 KiB
Python
334 lines
14 KiB
Python
"""验证每个平台在注册时使用独立（不同）邮箱地址的测试
|
||
|
||
测试逻辑：
|
||
1. 用一个 MockMailbox，每次 get_email() 返回全局唯一的地址（e1@test, e2@test ...）
|
||
2. 同时对多个平台并发发起注册请求（mock 掉实际网络请求）
|
||
3. 验证各平台收到的邮箱地址互不相同，且每次 register() 都调用了 get_email()
|
||
"""
|
||
import sys, os, types, threading, itertools, importlib
|
||
sys.path.insert(0, os.path.dirname(__file__))
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 1. 轻量 stub：阻止真实 IO 依赖被导入
|
||
# ─────────────────────────────────────────────────────────────
|
||
def _stub_module(name, **attrs):
|
||
mod = types.ModuleType(name)
|
||
mod.__dict__.update(attrs)
|
||
sys.modules[name] = mod
|
||
return mod
|
||
|
||
# curl_cffi stub
|
||
curl_mod = _stub_module("curl_cffi")
|
||
curl_req_mod = _stub_module("curl_cffi.requests")
|
||
class _FakeSession:
|
||
def __init__(self, *a, **kw): self.proxies = {}; self.headers = {}; self.cookies = _FC()
|
||
def get(self, *a, **kw): return _FR()
|
||
def post(self, *a, **kw): return _FR()
|
||
def update(self, *a, **kw): pass
|
||
class _FC:
|
||
def __iter__(self): return iter([])
|
||
def get(self, k, d=None): return d
|
||
class _FR:
|
||
status_code = 200
|
||
text = "{}"
|
||
content = b""
|
||
headers = {}
|
||
def json(self): return {}
|
||
curl_req_mod.Session = _FakeSession
|
||
curl_req_mod.get = lambda *a,**kw: _FR()
|
||
curl_mod.requests = curl_req_mod
|
||
|
||
# sqlmodel / sqlalchemy stubs
|
||
for m in ["sqlmodel", "sqlalchemy", "sqlalchemy.orm"]:
|
||
_stub_module(m)
|
||
sqlmodel_mod = sys.modules["sqlmodel"]
|
||
sqlmodel_mod.Field = lambda *a, **kw: None
|
||
sqlmodel_mod.SQLModel = object
|
||
sqlmodel_mod.create_engine = lambda *a, **kw: None
|
||
sqlmodel_mod.Session = object
|
||
sqlmodel_mod.select = lambda *a, **kw: None
|
||
|
||
# cbor2 / jwcrypto stubs (needed by kiro)
|
||
_stub_module("cbor2")
|
||
_stub_module("jwcrypto")
|
||
_stub_module("jwcrypto.jwk")
|
||
_stub_module("jwcrypto.jwe")
|
||
|
||
# requests stub
|
||
req_mod = _stub_module("requests")
|
||
req_mod.Session = _FakeSession
|
||
req_mod.get = lambda *a, **kw: _FR()
|
||
req_mod.post = lambda *a, **kw: _FR()
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 2. stub core.db so import doesn't touch SQLite
|
||
# ─────────────────────────────────────────────────────────────
|
||
db_mod = _stub_module("core.db")
|
||
from core.base_platform import Account, AccountStatus
|
||
db_mod.save_account = lambda account: account
|
||
db_mod.init_db = lambda: None
|
||
db_mod.get_session = lambda: iter([None])
|
||
db_mod.engine = None
|
||
|
||
class _FakeAccountModel:
|
||
pass
|
||
db_mod.AccountModel = _FakeAccountModel
|
||
db_mod.TaskLog = _FakeAccountModel
|
||
db_mod.ProxyModel = _FakeAccountModel
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 3. MockMailbox - 每次 get_email() 返回唯一地址
|
||
# ─────────────────────────────────────────────────────────────
|
||
from core.base_mailbox import BaseMailbox, MailboxAccount
|
||
|
||
class MockMailbox(BaseMailbox):
|
||
"""线程安全的 mock 邮箱：每次 get_email() 返回一个全局自增地址"""
|
||
_counter = itertools.count(1)
|
||
_lock = threading.Lock()
|
||
issued: list # 记录所有已发放的邮箱
|
||
|
||
def __init__(self):
|
||
self.issued = []
|
||
|
||
def get_email(self) -> MailboxAccount:
|
||
with self._lock:
|
||
n = next(self._counter)
|
||
addr = f"user{n:04d}@mockdomain.test"
|
||
with self._lock:
|
||
self.issued.append(addr)
|
||
return MailboxAccount(email=addr, account_id=str(n))
|
||
|
||
def get_current_ids(self, account: MailboxAccount) -> set:
|
||
return set()
|
||
|
||
def wait_for_code(self, account: MailboxAccount, keyword: str = "",
|
||
timeout: int = 120, before_ids: set = None) -> str:
|
||
return "123456" # 固定验证码，mock 不走真实邮件
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 4. Platform-level mock：替换真实 register 方法，仅记录邮箱
|
||
# ─────────────────────────────────────────────────────────────
|
||
from core.base_platform import BasePlatform, RegisterConfig
|
||
from core.registry import register as reg_decorator
|
||
|
||
REGISTERED_EMAILS: dict[str, list[str]] = {} # platform -> [email, ...]
|
||
_email_lock = threading.Lock()
|
||
|
||
|
||
def _make_mock_platform(platform_name: str, display: str):
|
||
"""动态创建一个 mock 平台，register() 调用 mailbox.get_email() 后记录结果"""
|
||
@reg_decorator
|
||
class _MockPlatform(BasePlatform):
|
||
name = platform_name
|
||
display_name = display
|
||
version = "1.0.0"
|
||
supported_executors = ["protocol", "headless", "headed"]
|
||
|
||
def __init__(self, config: RegisterConfig = None, mailbox=None):
|
||
super().__init__(config)
|
||
self.mailbox = mailbox
|
||
|
||
def register(self, email: str = None, password: str = None) -> Account:
|
||
mail_acct = self.mailbox.get_email() if self.mailbox else None
|
||
used_email = email or (mail_acct.email if mail_acct else "no-email")
|
||
with _email_lock:
|
||
REGISTERED_EMAILS.setdefault(platform_name, []).append(used_email)
|
||
return Account(
|
||
platform=platform_name,
|
||
email=used_email,
|
||
password=password or "mock-pw",
|
||
status=AccountStatus.REGISTERED,
|
||
)
|
||
|
||
def check_valid(self, account) -> bool:
|
||
return True
|
||
|
||
_MockPlatform.__name__ = f"Mock_{platform_name}"
|
||
_MockPlatform.__qualname__ = f"Mock_{platform_name}"
|
||
return _MockPlatform
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 5. 注册 mock 平台
|
||
# ─────────────────────────────────────────────────────────────
|
||
PLATFORMS = [
|
||
("cursor", "Cursor"),
|
||
("grok", "Grok"),
|
||
("trae", "Trae.ai"),
|
||
("kiro", "Kiro"),
|
||
("chatgpt", "ChatGPT"),
|
||
("tavily", "Tavily"),
|
||
("openblocklabs", "OpenBlockLabs"),
|
||
]
|
||
|
||
for pname, pdisplay in PLATFORMS:
|
||
_make_mock_platform(pname, pdisplay)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 6. 测试函数
|
||
# ─────────────────────────────────────────────────────────────
|
||
from core.registry import get as get_platform
|
||
|
||
COLOR_OK = "\033[92m"
|
||
COLOR_FAIL = "\033[91m"
|
||
COLOR_WARN = "\033[93m"
|
||
COLOR_BOLD = "\033[1m"
|
||
COLOR_OFF = "\033[0m"
|
||
|
||
|
||
def run_test_single_registration():
|
||
"""测试1：每个平台各注册1次，验证各自得到独立邮箱（互不相同）"""
|
||
print(f"\n{COLOR_BOLD}=== 测试1：各平台单次注册，验证邮箱各不相同 ==={COLOR_OFF}")
|
||
REGISTERED_EMAILS.clear()
|
||
all_emails = []
|
||
results = []
|
||
|
||
for pname, _ in PLATFORMS:
|
||
mailbox = MockMailbox()
|
||
PlatformCls = get_platform(pname)
|
||
config = RegisterConfig(executor_type="protocol")
|
||
platform = PlatformCls(config=config, mailbox=mailbox)
|
||
account = platform.register(email=None, password=None)
|
||
email_used = account.email
|
||
all_emails.append(email_used)
|
||
results.append((pname, email_used))
|
||
print(f" {pname:20s} → {email_used}")
|
||
|
||
# 验证：所有邮箱必须唯一
|
||
unique_emails = set(all_emails)
|
||
if len(unique_emails) == len(all_emails):
|
||
print(f"{COLOR_OK} ✓ PASS：{len(all_emails)} 个平台使用了 {len(unique_emails)} 个互不相同的邮箱{COLOR_OFF}")
|
||
return True
|
||
else:
|
||
duplicates = [e for e in all_emails if all_emails.count(e) > 1]
|
||
print(f"{COLOR_FAIL} ✗ FAIL：发现重复邮箱: {set(duplicates)}{COLOR_OFF}")
|
||
return False
|
||
|
||
|
||
def run_test_multi_registration_same_platform():
|
||
"""测试2：同一平台连续注册 3 次，每次必须得到不同邮箱"""
|
||
print(f"\n{COLOR_BOLD}=== 测试2：同平台多次注册，每次邮箱不同 ==={COLOR_OFF}")
|
||
all_pass = True
|
||
|
||
for pname, _ in PLATFORMS:
|
||
REGISTERED_EMAILS.clear()
|
||
mailbox = MockMailbox()
|
||
PlatformCls = get_platform(pname)
|
||
config = RegisterConfig(executor_type="protocol")
|
||
emails_this_run = []
|
||
|
||
for i in range(3):
|
||
platform = PlatformCls(config=config, mailbox=mailbox)
|
||
account = platform.register(email=None, password=None)
|
||
emails_this_run.append(account.email)
|
||
|
||
unique = set(emails_this_run)
|
||
if len(unique) == 3:
|
||
print(f" {COLOR_OK}✓ {pname:20s} 3次注册邮箱全不同: {emails_this_run}{COLOR_OFF}")
|
||
else:
|
||
print(f" {COLOR_FAIL}✗ {pname:20s} 存在重复邮箱! {emails_this_run}{COLOR_OFF}")
|
||
all_pass = False
|
||
|
||
return all_pass
|
||
|
||
|
||
def run_test_email_argument_override():
|
||
"""测试3：显式传入 email 时，平台应使用该 email，而不是从 mailbox 取"""
|
||
print(f"\n{COLOR_BOLD}=== 测试3：显式传入 email 时，平台使用该固定邮箱 ==={COLOR_OFF}")
|
||
all_pass = True
|
||
fixed_email = "fixed@example.com"
|
||
|
||
for pname, _ in PLATFORMS:
|
||
mailbox = MockMailbox()
|
||
PlatformCls = get_platform(pname)
|
||
config = RegisterConfig(executor_type="protocol")
|
||
platform = PlatformCls(config=config, mailbox=mailbox)
|
||
account = platform.register(email=fixed_email, password=None)
|
||
|
||
# 当显式传入 email 时，mock platform 直接用 email 参数（不调用 mailbox）
|
||
# 但实际各平台实现：若 email 非 None，则跳过 mailbox.get_email()
|
||
# 此处验证返回的 account.email 与传入一致
|
||
if account.email == fixed_email:
|
||
print(f" {COLOR_OK}✓ {pname:20s} 使用固定邮箱: {account.email}{COLOR_OFF}")
|
||
else:
|
||
print(f" {COLOR_FAIL}✗ {pname:20s} 未使用传入邮箱！got={account.email}{COLOR_OFF}")
|
||
all_pass = False
|
||
|
||
return all_pass
|
||
|
||
|
||
def run_test_concurrent_registrations():
|
||
"""测试4：并发注册时，各线程得到的邮箱地址互不相同"""
|
||
print(f"\n{COLOR_BOLD}=== 测试4：并发注册（同平台5线程），邮箱无冲突 ==={COLOR_OFF}")
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
all_pass = True
|
||
|
||
for pname, _ in PLATFORMS:
|
||
mailbox = MockMailbox()
|
||
PlatformCls = get_platform(pname)
|
||
config = RegisterConfig(executor_type="protocol")
|
||
collected = []
|
||
lock = threading.Lock()
|
||
|
||
def _do(idx):
|
||
platform = PlatformCls(config=config, mailbox=mailbox)
|
||
account = platform.register(email=None, password=None)
|
||
with lock:
|
||
collected.append(account.email)
|
||
|
||
with ThreadPoolExecutor(max_workers=5) as pool:
|
||
futs = [pool.submit(_do, i) for i in range(5)]
|
||
for f in as_completed(futs):
|
||
f.result()
|
||
|
||
unique = set(collected)
|
||
if len(unique) == 5:
|
||
print(f" {COLOR_OK}✓ {pname:20s} 5线程并发，5个唯一邮箱{COLOR_OFF}")
|
||
else:
|
||
print(f" {COLOR_FAIL}✗ {pname:20s} 并发冲突！emails={collected}{COLOR_OFF}")
|
||
all_pass = False
|
||
|
||
return all_pass
|
||
|
||
|
||
def run_test_real_platform_email_logic():
|
||
"""
|
||
测试5：验证真实平台插件代码中，email 参数为 None 时确实调用 mailbox.get_email()。
|
||
通过追踪 MockMailbox.issued 来确认。
|
||
"""
|
||
print(f"\n{COLOR_BOLD}=== 测试5：验证 mock 平台 register(email=None) 确实调用了 mailbox.get_email() ==={COLOR_OFF}")
|
||
all_pass = True
|
||
|
||
for pname, _ in PLATFORMS:
|
||
mailbox = MockMailbox()
|
||
PlatformCls = get_platform(pname)
|
||
config = RegisterConfig(executor_type="protocol")
|
||
platform = PlatformCls(config=config, mailbox=mailbox)
|
||
|
||
before = len(mailbox.issued)
|
||
platform.register(email=None, password=None)
|
||
after = len(mailbox.issued)
|
||
|
||
called = (after - before) >= 1
|
||
if called:
|
||
print(f" {COLOR_OK}✓ {pname:20s} register(email=None) 调用了 get_email()，分配地址: {mailbox.issued[-1]}{COLOR_OFF}")
|
||
else:
|
||
print(f" {COLOR_FAIL}✗ {pname:20s} register(email=None) 未调用 get_email()！{COLOR_OFF}")
|
||
all_pass = False
|
||
|
||
return all_pass
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────
|
||
# 7. 汇总入口
|
||
# ─────────────────────────────────────────────────────────────
|
||
if __name__ == "__main__":
|
||
print(f"{COLOR_BOLD}\n{'='*60}")
|
||
print(" 平台独立邮箱验证测试套件")
|
||
print(f" 平台列表: {[p for p,_ in PLATFORMS]}")
|
||
print(f"{'='*60}{COLOR_OFF}")
|
||
|
||
tests = [
|
||
|