From 9b047db6f6113db493747c4eaa87e3c77d7ef51f Mon Sep 17 00:00:00 2001 From: Aaron Liang <76561968+AaronL725@users.noreply.github.com> Date: Wed, 15 Jul 2026 12:02:57 +0800 Subject: [PATCH] fix: harden postprocessing and recovery boundaries Preserve local CPA success while surfacing hotload-copy warnings, retry transient DuckMail/YYDS message details, serialize pending recovery on target files, isolate optional NSFW failures, and restore main-module constant forwarding. --- account_outputs.py | 10 +- cpa_export.py | 2 + grok_register_ttk.py | 20 +- mail_service.py | 18 +- registration_flow.py | 23 +- tests/test_minimal_boundary_regressions.py | 157 +++++++++ tools/apply_minimal_regression_fixes.py | 392 +++++++++++++++++++++ 7 files changed, 606 insertions(+), 16 deletions(-) create mode 100644 tests/test_minimal_boundary_regressions.py create mode 100644 tools/apply_minimal_regression_fixes.py diff --git a/account_outputs.py b/account_outputs.py index ccc0ba2..c4bb045 100644 --- a/account_outputs.py +++ b/account_outputs.py @@ -3,6 +3,7 @@ import json import os import tempfile import time +from contextlib import ExitStack from datetime import datetime, timezone from filelock import FileLock @@ -67,8 +68,13 @@ def retry_pending_file(pending_path, output_path=None, log_callback=None): if os.path.normcase(pending_path) == os.path.normcase(target_path): raise ValueError("pending 输入文件与输出文件不能是同一个文件") - lock_path = pending_path + ".lock" - with FileLock(lock_path, timeout=30): + lock_paths = sorted( + {pending_path + ".lock", target_path + ".lock"}, + key=lambda value: os.path.normcase(os.path.abspath(value)), + ) + with ExitStack() as stack: + for lock_path in lock_paths: + stack.enter_context(FileLock(lock_path, timeout=30)) if not os.path.isfile(pending_path): return {"restored": 0, "remaining": 0, "output_path": target_path} with open(pending_path, "r", encoding="utf-8") as handle: diff --git a/cpa_export.py b/cpa_export.py index aa1b150..23fb3e9 100644 --- a/cpa_export.py +++ b/cpa_export.py @@ -176,6 +176,8 @@ def export_cpa_xai_for_account(email, password, page=None, cookies=None, sso=Non log("[cpa] hotload copy -> %s" % target) except Exception as exc: result["cpa_copy_error"] = str(exc) + result["warning"] = True + result["partial"] = True log("[cpa] hotload copy failed: %s" % exc) if not result.get("ok"): fail_path = settings.auth_dir / "cpa_auth_failed.txt" diff --git a/grok_register_ttk.py b/grok_register_ttk.py index 4e3597a..2dbf631 100755 --- a/grok_register_ttk.py +++ b/grok_register_ttk.py @@ -53,7 +53,7 @@ import browser_runtime as _browser_runtime import mail_service as _mail_service import registration_browser as _registration_browser from app_config import ( - CONFIG_FILE, DEFAULT_CONFIG, ConfigError, config, load_config, save_config, + DEFAULT_CONFIG, ConfigError, config, load_config, save_config, validate_config, validate_config_structure, validate_run_requirements, ) @@ -299,6 +299,10 @@ for _name in ['generate_random_birthdate', 'response_preview', 'is_cloudflare_bl def __getattr__(name): + if name == "CONFIG_FILE": + return _app_config.CONFIG_FILE + if name == "SIGNUP_URL": + return _registration_browser.SIGNUP_URL if name in {"browser", "page", "browser_proxy_bridge", "browser_started_with_proxy", "cf_clearance"}: return getattr(_registration_browser, name) if name in {"_cf_domain_index", "_cloudmail_domain_index"}: @@ -308,6 +312,14 @@ def __getattr__(name): class _CompatibilityModule(types.ModuleType): def __setattr__(self, name, value): + if name == "CONFIG_FILE": + _app_config.CONFIG_FILE = str(value) + self.__dict__.pop(name, None) + return + if name == "SIGNUP_URL": + _registration_browser.SIGNUP_URL = str(value) + self.__dict__.pop(name, None) + return if name == "config": if value is not _app_config.config: if not isinstance(value, dict): @@ -585,7 +597,11 @@ def maybe_export_cpa_xai_after_success(email, password, sso="", log_callback=Non if result.get("ok"): exported_path = result.get("hotload_path") or result.get("path") or "" suffix = f": {exported_path}" if exported_path else "" - logger(f"[+] CPA OIDC 导出成功{suffix}") + if result.get("warning") or result.get("partial") or result.get("cpa_copy_error"): + detail = result.get("cpa_copy_error") or "后处理未完整完成" + logger(f"[!] CPA OIDC 凭证已生成，但存在后处理警告{suffix}: {detail}") + else: + logger(f"[+] CPA OIDC 导出成功{suffix}") elif not result.get("skipped"): logger(f"[!] CPA OIDC 导出失败，账号已保留: {result.get('error') or result}") return result diff --git a/mail_service.py b/mail_service.py index ba4df92..0f8fd79 100644 --- a/mail_service.py +++ b/mail_service.py @@ -449,7 +449,7 @@ def duckmail_get_oai_code( cancel_callback=None, ): deadline = time.time() + timeout - seen_ids = set() + seen_attempts = {} while time.time() < deadline: raise_if_cancelled(cancel_callback) try: @@ -461,12 +461,15 @@ def duckmail_get_oai_code( continue for msg in messages: msg_id = msg.get("id") or msg.get("msgid") - if not msg_id or msg_id in seen_ids: + if not msg_id: continue - seen_ids.add(msg_id) recipients = [t.get("address", "").lower() for t in (msg.get("to") or [])] if email.lower() not in recipients: continue + attempt = int(seen_attempts.get(msg_id, 0)) + if attempt >= 5: + continue + seen_attempts[msg_id] = attempt + 1 try: detail = get_message_detail(dev_token, msg_id) except Exception as exc: @@ -790,7 +793,7 @@ def yyds_get_oai_code( cancel_callback=None, ): deadline = time.time() + timeout - seen_ids = set() + seen_attempts = {} while time.time() < deadline: raise_if_cancelled(cancel_callback) try: @@ -802,12 +805,15 @@ def yyds_get_oai_code( continue for msg in messages: msg_id = msg.get("id") - if not msg_id or msg_id in seen_ids: + if not msg_id: continue - seen_ids.add(msg_id) to_addrs = [t.get("address", "").lower() for t in (msg.get("to") or [])] if address.lower() not in to_addrs: continue + attempt = int(seen_attempts.get(msg_id, 0)) + if attempt >= 5: + continue + seen_attempts[msg_id] = attempt + 1 try: detail = yyds_get_message_detail(msg_id, token=token, jwt=jwt) except Exception as exc: diff --git a/registration_flow.py b/registration_flow.py index f77dc89..822af4f 100644 --- a/registration_flow.py +++ b/registration_flow.py @@ -111,11 +111,14 @@ def register_one_account(callbacks, ops, enable_nsfw=True, max_mail_retry=3): sso = ops.wait_for_sso_cookie() if enable_nsfw: callbacks.log("[*] 6. 开启 NSFW") - nsfw_ok, nsfw_msg = ops.enable_nsfw(sso) - if nsfw_ok: - callbacks.log(f"[+] NSFW 开启成功: {nsfw_msg}") - else: - callbacks.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") + try: + nsfw_ok, nsfw_msg = ops.enable_nsfw(sso) + if nsfw_ok: + callbacks.log(f"[+] NSFW 开启成功: {nsfw_msg}") + else: + callbacks.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") + except Exception as exc: + callbacks.log(f"[!] NSFW 开启异常，继续保存账号: {exc}") return RegistrationResult( ok=True, email=email, @@ -283,7 +286,15 @@ def run_batch(count, callbacks, observer, ops, enable_nsfw=True, cleanup_interva isinstance(state, dict) and state.get("enabled") and not state.get("ok") for state in output.pools.values() ) - cpa_warning = bool(output.cpa and not output.cpa.get("ok") and not output.cpa.get("skipped")) + cpa_warning = bool( + output.cpa + and not output.cpa.get("skipped") + and ( + not output.cpa.get("ok") + or output.cpa.get("warning") + or output.cpa.get("cpa_copy_error") + ) + ) if pool_warning or cpa_warning: result.postprocess_warning_count += 1 except ops.cancelled_exception: diff --git a/tests/test_minimal_boundary_regressions.py b/tests/test_minimal_boundary_regressions.py new file mode 100644 index 0000000..1b57689 --- /dev/null +++ b/tests/test_minimal_boundary_regressions.py @@ -0,0 +1,157 @@ +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_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() diff --git a/tools/apply_minimal_regression_fixes.py b/tools/apply_minimal_regression_fixes.py new file mode 100644 index 0000000..a9ee0fb --- /dev/null +++ b/tools/apply_minimal_regression_fixes.py @@ -0,0 +1,392 @@ +#!/usr/bin/env python3 +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + + +def read(path): + return (ROOT / path).read_text(encoding="utf-8") + + +def write(path, text): + (ROOT / path).write_text(text, encoding="utf-8") + + +def replace_once(text, old, new, label): + if new in text: + return text + count = text.count(old) + if count != 1: + raise RuntimeError(f"{label}: expected one match, got {count}") + return text.replace(old, new, 1) + + +# 1. CPA hotload copy failures remain local-success, but become partial warnings. +path = "cpa_export.py" +text = read(path) +old = ''' except Exception as exc: + result["cpa_copy_error"] = str(exc) + log("[cpa] hotload copy failed: %s" % exc) +''' +new = ''' except Exception as exc: + result["cpa_copy_error"] = str(exc) + result["warning"] = True + result["partial"] = True + log("[cpa] hotload copy failed: %s" % exc) +''' +text = replace_once(text, old, new, "CPA copy warning flags") +write(path, text) + + +# 2. Public flow isolates optional NSFW exceptions and counts partial CPA warnings. +path = "registration_flow.py" +text = read(path) +old = ''' if enable_nsfw: + callbacks.log("[*] 6. 开启 NSFW") + nsfw_ok, nsfw_msg = ops.enable_nsfw(sso) + if nsfw_ok: + callbacks.log(f"[+] NSFW 开启成功: {nsfw_msg}") + else: + callbacks.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") +''' +new = ''' if enable_nsfw: + callbacks.log("[*] 6. 开启 NSFW") + try: + nsfw_ok, nsfw_msg = ops.enable_nsfw(sso) + if nsfw_ok: + callbacks.log(f"[+] NSFW 开启成功: {nsfw_msg}") + else: + callbacks.log(f"[!] NSFW 未开启，继续保存账号: {nsfw_msg}") + except Exception as exc: + callbacks.log(f"[!] NSFW 开启异常，继续保存账号: {exc}") +''' +text = replace_once(text, old, new, "NSFW exception isolation") +old = ''' cpa_warning = bool(output.cpa and not output.cpa.get("ok") and not output.cpa.get("skipped")) +''' +new = ''' cpa_warning = bool( + output.cpa + and not output.cpa.get("skipped") + and ( + not output.cpa.get("ok") + or output.cpa.get("warning") + or output.cpa.get("cpa_copy_error") + ) + ) +''' +text = replace_once(text, old, new, "CPA partial warning statistics") +write(path, text) + + +# 3. DuckMail and YYDS retry the same message up to five times. +path = "mail_service.py" +text = read(path) +old = ''' deadline = time.time() + timeout + seen_ids = set() + while time.time() < deadline: +''' +new = ''' deadline = time.time() + timeout + seen_attempts = {} + while time.time() < deadline: +''' +# The same prelude exists in exactly DuckMail and YYDS. +if "seen_ids = set()" in text: + if text.count(old) != 2: + raise RuntimeError(f"mail retry preludes: expected two matches, got {text.count(old)}") + text = text.replace(old, new, 2) + +old = ''' msg_id = msg.get("id") or msg.get("msgid") + if not msg_id or msg_id in seen_ids: + continue + seen_ids.add(msg_id) + recipients = [t.get("address", "").lower() for t in (msg.get("to") or [])] + if email.lower() not in recipients: + continue + try: +''' +new = ''' msg_id = msg.get("id") or msg.get("msgid") + if not msg_id: + continue + recipients = [t.get("address", "").lower() for t in (msg.get("to") or [])] + if email.lower() not in recipients: + continue + attempt = int(seen_attempts.get(msg_id, 0)) + if attempt >= 5: + continue + seen_attempts[msg_id] = attempt + 1 + try: +''' +text = replace_once(text, old, new, "DuckMail message retries") +old = ''' msg_id = msg.get("id") + if not msg_id or msg_id in seen_ids: + continue + seen_ids.add(msg_id) + to_addrs = [t.get("address", "").lower() for t in (msg.get("to") or [])] + if address.lower() not in to_addrs: + continue + try: +''' +new = ''' msg_id = msg.get("id") + if not msg_id: + continue + to_addrs = [t.get("address", "").lower() for t in (msg.get("to") or [])] + if address.lower() not in to_addrs: + continue + attempt = int(seen_attempts.get(msg_id, 0)) + if attempt >= 5: + continue + seen_attempts[msg_id] = attempt + 1 + try: +''' +text = replace_once(text, old, new, "YYDS message retries") +write(path, text) + + +# 4. Pending recovery locks both pending and target paths in deterministic order. +path = "account_outputs.py" +text = read(path) +text = replace_once( + text, + "import time\nfrom datetime import datetime, timezone\n", + "import time\nfrom contextlib import ExitStack\nfrom datetime import datetime, timezone\n", + "ExitStack import", +) +old = ''' lock_path = pending_path + ".lock" + with FileLock(lock_path, timeout=30): + if not os.path.isfile(pending_path): +''' +new = ''' lock_paths = sorted( + {pending_path + ".lock", target_path + ".lock"}, + key=lambda value: os.path.normcase(os.path.abspath(value)), + ) + with ExitStack() as stack: + for lock_path in lock_paths: + stack.enter_context(FileLock(lock_path, timeout=30)) + if not os.path.isfile(pending_path): +''' +text = replace_once(text, old, new, "pending and target locks") +write(path, text) + + +# 5. Preserve main-module constant read/write compatibility without broadening proxies. +path = "grok_register_ttk.py" +text = read(path) +text = replace_once( + text, + '''from app_config import ( + CONFIG_FILE, DEFAULT_CONFIG, ConfigError, config, load_config, save_config, +''', + '''from app_config import ( + DEFAULT_CONFIG, ConfigError, config, load_config, save_config, +''', + "remove copied CONFIG_FILE", +) +old = '''def __getattr__(name): + if name in {"browser", "page", "browser_proxy_bridge", "browser_started_with_proxy", "cf_clearance"}: + return getattr(_registration_browser, name) +''' +new = '''def __getattr__(name): + if name == "CONFIG_FILE": + return _app_config.CONFIG_FILE + if name == "SIGNUP_URL": + return _registration_browser.SIGNUP_URL + if name in {"browser", "page", "browser_proxy_bridge", "browser_started_with_proxy", "cf_clearance"}: + return getattr(_registration_browser, name) +''' +text = replace_once(text, old, new, "constant read compatibility") +old = ''' def __setattr__(self, name, value): + if name == "config": +''' +new = ''' def __setattr__(self, name, value): + if name == "CONFIG_FILE": + _app_config.CONFIG_FILE = str(value) + self.__dict__.pop(name, None) + return + if name == "SIGNUP_URL": + _registration_browser.SIGNUP_URL = str(value) + self.__dict__.pop(name, None) + return + if name == "config": +''' +text = replace_once(text, old, new, "constant write compatibility") +old = ''' if result.get("ok"): + exported_path = result.get("hotload_path") or result.get("path") or "" + suffix = f": {exported_path}" if exported_path else "" + logger(f"[+] CPA OIDC 导出成功{suffix}") + elif not result.get("skipped"): +''' +new = ''' if result.get("ok"): + exported_path = result.get("hotload_path") or result.get("path") or "" + suffix = f": {exported_path}" if exported_path else "" + if result.get("warning") or result.get("partial") or result.get("cpa_copy_error"): + detail = result.get("cpa_copy_error") or "后处理未完整完成" + logger(f"[!] CPA OIDC 凭证已生成，但存在后处理警告{suffix}: {detail}") + else: + logger(f"[+] CPA OIDC 导出成功{suffix}") + elif not result.get("skipped"): +''' +text = replace_once(text, old, new, "CPA partial success logging") +write(path, text) + + +# Focused regression tests only for the changed boundaries. +tests = r'''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_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() +''' +write("tests/test_minimal_boundary_regressions.py", tests) + +print("minimal regression fixes applied")