167 lines
7.2 KiB
Python
167 lines
7.2 KiB
Python
"""验证后处理警告、邮件重试、目标锁和常量兼容边界。"""
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import account_outputs
|
|
import app_config
|
|
import cpa_export
|
|
import grok_register_ttk as app
|
|
import mail_service
|
|
import registration_browser
|
|
from registration_flow import RegistrationCallbacks, RegistrationOperations, run_batch
|
|
|
|
|
|
class Cancelled(Exception):
|
|
pass
|
|
|
|
|
|
class Retryable(Exception):
|
|
pass
|
|
|
|
|
|
def make_ops(enable_nsfw=None, export_cpa=None):
|
|
return RegistrationOperations(
|
|
start_browser=lambda: None,
|
|
restart_browser=lambda: None,
|
|
browser_missing=lambda: False,
|
|
open_signup_page=lambda: None,
|
|
fill_email_and_submit=lambda: ("user@example.com", "mail-token"),
|
|
save_mail_credential=lambda email, token: True,
|
|
fill_code_and_submit=lambda email, token: "ABC-123",
|
|
fill_profile_and_submit=lambda: {"given_name": "A", "family_name": "B", "password": "pw"},
|
|
wait_for_sso_cookie=lambda: "sso-token",
|
|
enable_nsfw=enable_nsfw or (lambda sso: (True, "ok")),
|
|
persist_account_line=lambda email, password, sso: None,
|
|
queue_unsaved_result=lambda payload, error: True,
|
|
add_tokens=lambda sso, email: {
|
|
"local": {"enabled": False, "ok": None, "error": None},
|
|
"remote": {"enabled": False, "ok": None, "error": None},
|
|
},
|
|
export_cpa=export_cpa or (lambda email, password, sso: {"ok": False, "skipped": True}),
|
|
cleanup=lambda reason: None,
|
|
sleep=lambda seconds: None,
|
|
cancelled_exception=Cancelled,
|
|
retry_exception=Retryable,
|
|
)
|
|
|
|
|
|
class MinimalBoundaryRegressionTests(unittest.TestCase):
|
|
def callbacks(self, logs=None):
|
|
logs = logs if logs is not None else []
|
|
return RegistrationCallbacks(log=logs.append, cancelled=lambda: False)
|
|
|
|
def test_cpa_copy_failure_is_partial_warning_and_batch_warning(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
config = {
|
|
"cpa_export_enabled": True,
|
|
"cpa_auth_dir": directory,
|
|
"cpa_copy_to_hotload": True,
|
|
"cpa_hotload_dir": str(Path(directory) / "hotload"),
|
|
}
|
|
mint = lambda **kwargs: {"ok": True, "path": str(Path(directory) / "xai-test.json")}
|
|
with patch.object(cpa_export, "_load_mint_and_export", return_value=mint), \
|
|
patch.object(cpa_export.shutil, "copy2", side_effect=OSError("copy failed")):
|
|
result = cpa_export.export_cpa_xai_for_account("user@example.com", "pw", config=config)
|
|
self.assertTrue(result["ok"])
|
|
self.assertTrue(result["warning"])
|
|
self.assertTrue(result["partial"])
|
|
self.assertIn("copy failed", result["cpa_copy_error"])
|
|
|
|
batch = run_batch(
|
|
1,
|
|
self.callbacks(),
|
|
lambda *args: None,
|
|
make_ops(export_cpa=lambda *args: {
|
|
"ok": True,
|
|
"warning": True,
|
|
"partial": True,
|
|
"cpa_copy_error": "copy failed",
|
|
}),
|
|
)
|
|
self.assertEqual(batch.success_count, 1)
|
|
self.assertEqual(batch.postprocess_warning_count, 1)
|
|
|
|
def test_duckmail_retries_same_message_after_detail_failure(self):
|
|
message = {"id": "m1", "to": [{"address": "user@example.com"}]}
|
|
with patch.object(mail_service, "get_messages", return_value=[message]), \
|
|
patch.object(mail_service, "get_message_detail", side_effect=[RuntimeError("temporary"), {"subject": "ABC-123 xAI"}]), \
|
|
patch.object(mail_service, "sleep_with_cancel", return_value=None), \
|
|
patch.object(mail_service.time, "time", side_effect=[0, 0, 0]):
|
|
code = mail_service.duckmail_get_oai_code("token", "user@example.com", timeout=1, poll_interval=0)
|
|
self.assertEqual(code, "ABC-123")
|
|
|
|
def test_yyds_retries_same_message_after_detail_failure(self):
|
|
message = {"id": "m1", "to": [{"address": "user@example.com"}]}
|
|
with patch.object(mail_service, "yyds_get_messages", return_value=[message]), \
|
|
patch.object(mail_service, "yyds_get_message_detail", side_effect=[RuntimeError("temporary"), {"subject": "ABC-123 xAI"}]), \
|
|
patch.object(mail_service, "sleep_with_cancel", return_value=None), \
|
|
patch.object(mail_service.time, "time", side_effect=[0, 0, 0]):
|
|
code = mail_service.yyds_get_oai_code("token", "user@example.com", timeout=1, poll_interval=0)
|
|
self.assertEqual(code, "ABC-123")
|
|
|
|
def test_confirmation_subject_code_beats_lowercase_css_token(self):
|
|
code = mail_service.extract_verification_code(
|
|
".mj-column-per-100 { width: 100%; }",
|
|
"SpaceXAI confirmation code: XOE-G02",
|
|
)
|
|
self.assertEqual(code, "XOE-G02")
|
|
|
|
def test_pending_recovery_acquires_pending_and_target_locks_in_fixed_order(self):
|
|
acquired = []
|
|
|
|
class FakeLock:
|
|
def __init__(self, path, timeout=30):
|
|
self.path = path
|
|
def __enter__(self):
|
|
acquired.append(self.path)
|
|
return self
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
pending = Path(directory) / "one.pending.jsonl"
|
|
target = Path(directory) / "accounts.txt"
|
|
pending.write_text(json.dumps({"email": "u@example.com", "password": "pw", "sso": "sso"}) + "\n", encoding="utf-8")
|
|
with patch.object(account_outputs, "FileLock", FakeLock):
|
|
account_outputs.retry_pending_file(str(pending), output_path=str(target))
|
|
expected = sorted(
|
|
[str(pending.resolve()) + ".lock", str(target.resolve()) + ".lock"],
|
|
key=lambda value: __import__("os").path.normcase(__import__("os").path.abspath(value)),
|
|
)
|
|
self.assertEqual(acquired, expected)
|
|
|
|
def test_nsfw_exception_does_not_discard_registered_account(self):
|
|
logs = []
|
|
batch = run_batch(
|
|
1,
|
|
self.callbacks(logs),
|
|
lambda *args: None,
|
|
make_ops(enable_nsfw=lambda sso: (_ for _ in ()).throw(RuntimeError("nsfw down"))),
|
|
)
|
|
self.assertEqual(batch.success_count, 1)
|
|
self.assertEqual(batch.fail_count, 0)
|
|
self.assertTrue(any("NSFW 开启异常" in line for line in logs))
|
|
|
|
def test_main_constant_assignments_forward_to_owner_modules(self):
|
|
old_config_file = app_config.CONFIG_FILE
|
|
old_signup_url = registration_browser.SIGNUP_URL
|
|
try:
|
|
app.CONFIG_FILE = "other.json"
|
|
app.SIGNUP_URL = "https://example.test/signup"
|
|
self.assertEqual(app_config.CONFIG_FILE, "other.json")
|
|
self.assertEqual(app.CONFIG_FILE, "other.json")
|
|
self.assertEqual(registration_browser.SIGNUP_URL, "https://example.test/signup")
|
|
self.assertEqual(app.SIGNUP_URL, "https://example.test/signup")
|
|
finally:
|
|
app.CONFIG_FILE = old_config_file
|
|
app.SIGNUP_URL = old_signup_url
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|