From b8b1eb72d153c028bd93bf71050cf01ccd66f305 Mon Sep 17 00:00:00 2001 From: Mison Date: Tue, 24 Mar 2026 01:49:54 +0800 Subject: [PATCH] test: finalize restored task coverage --- tests/test_register_phase_runner.py | 249 ---------------------------- tests/test_tempmail_service.py | 4 +- 2 files changed, 2 insertions(+), 251 deletions(-) delete mode 100644 tests/test_register_phase_runner.py diff --git a/tests/test_register_phase_runner.py b/tests/test_register_phase_runner.py deleted file mode 100644 index 64cbd7a..0000000 --- a/tests/test_register_phase_runner.py +++ /dev/null @@ -1,249 +0,0 @@ -from types import SimpleNamespace - -import src.core.register as register_module -from src.core.register import PhaseContext, PhaseResult, RegistrationEngine, RegistrationResult -from src.services import EmailServiceType -from src.services.base import BaseEmailService, RateLimitedEmailServiceError - - -class DummySettings: - openai_client_id = "client-id" - openai_auth_url = "https://auth.example.test" - openai_token_url = "https://token.example.test" - openai_redirect_uri = "https://callback.example.test" - openai_scope = "openid profile email" - - -class FakeResponse: - def __init__(self, status_code=200, payload=None, text=""): - self.status_code = status_code - self._payload = payload or {} - self.text = text - - def json(self): - return self._payload - - -class FakeCookies(dict): - def get(self, key, default=None): - return super().get(key, default) - - -class FakePasswordSession: - def __init__(self, response): - self.response = response - self.cookies = FakeCookies({"oai-did": "did-1"}) - self.post_calls = [] - self.get_calls = [] - - def post(self, url, **kwargs): - self.post_calls.append({"url": url, **kwargs}) - return self.response - - def get(self, url, **kwargs): - self.get_calls.append({"url": url, **kwargs}) - return FakeResponse(status_code=200) - - -def _build_engine(monkeypatch): - monkeypatch.setattr(register_module, "get_settings", lambda: DummySettings()) - email_service = SimpleNamespace(service_type=EmailServiceType.DUCK_MAIL) - return RegistrationEngine(email_service=email_service) - - -class RateLimitedEmailService(BaseEmailService): - def __init__(self): - super().__init__(EmailServiceType.DUCK_MAIL, "duck-test") - - def create_email(self, config=None): - error = RateLimitedEmailServiceError("请求失败: 429", retry_after=7) - self.update_status(False, error) - raise error - - def get_verification_code(self, email, email_id=None, timeout=120, pattern=r"(? bool: - return False - - def check_health(self) -> bool: - return True - - -def test_run_executes_nine_explicit_phases(monkeypatch): - engine = _build_engine(monkeypatch) - order = [] - phase_names = [ - "ip_check", - "email_prepare", - "signup", - "otp_primary", - "account_create", - "oauth_reenter", - "otp_secondary", - "workspace_resolve", - "oauth_callback", - ] - - def make_phase(name): - def _phase(result, context): - order.append(name) - if name == "email_prepare": - result.email = "tester@example.com" - if name == "workspace_resolve": - result.workspace_id = "ws-1" - context.callback_url = "https://callback.example.test?code=abc&state=xyz" - if name == "oauth_callback": - context.token_info = { - "account_id": "acct-1", - "access_token": "access-token", - "refresh_token": "refresh-token", - "id_token": "id-token", - } - return PhaseResult(phase=name, success=True, data={"phase": name}) - - return _phase - - for phase_name in phase_names: - monkeypatch.setattr(engine, f"_phase_{phase_name}", make_phase(phase_name)) - - engine.session = SimpleNamespace( - cookies=FakeCookies({"__Secure-next-auth.session-token": "session-token"}) - ) - - result = engine.run() - - assert result.success is True - assert order == phase_names - assert [item.phase for item in engine.phase_history] == phase_names - assert all(isinstance(item, PhaseResult) for item in engine.phase_history) - assert result.email == "tester@example.com" - assert result.workspace_id == "ws-1" - assert result.account_id == "acct-1" - assert result.session_token == "session-token" - assert result.source == "register" - assert "registration_mode" not in result.metadata - - -def test_run_stops_on_first_failed_phase(monkeypatch): - engine = _build_engine(monkeypatch) - order = [] - - def success_phase(name): - def _phase(result, context): - order.append(name) - return PhaseResult(phase=name, success=True) - - return _phase - - def failed_signup(result, context): - order.append("signup") - return PhaseResult( - phase="signup", - success=False, - error_message="提交注册表单失败: 协议错误", - ) - - monkeypatch.setattr(engine, "_phase_ip_check", success_phase("ip_check")) - monkeypatch.setattr(engine, "_phase_email_prepare", success_phase("email_prepare")) - monkeypatch.setattr(engine, "_phase_signup", failed_signup) - - for phase_name in ["otp_primary", "account_create", "oauth_reenter", "otp_secondary", "workspace_resolve", "oauth_callback"]: - monkeypatch.setattr( - engine, - f"_phase_{phase_name}", - lambda result, context, name=phase_name: PhaseResult( - phase=name, - success=True, - ), - ) - - result = engine.run() - - assert result.success is False - assert result.error_message == "提交注册表单失败: 协议错误" - assert order == ["ip_check", "email_prepare", "signup"] - assert [item.phase for item in engine.phase_history] == order - - -def test_email_prepare_phase_exposes_provider_backoff(monkeypatch): - monkeypatch.setattr(register_module, "get_settings", lambda: DummySettings()) - engine = RegistrationEngine(email_service=RateLimitedEmailService()) - - phase_result = engine._phase_email_prepare( - RegistrationResult(success=False, logs=[]), - PhaseContext(), - ) - - assert phase_result.success is False - assert phase_result.error_code == "EMAIL_PROVIDER_RATE_LIMITED" - assert phase_result.retryable is True - assert phase_result.next_action == "switch_provider" - assert phase_result.provider_backoff is not None - assert phase_result.provider_backoff.failures == 1 - assert phase_result.provider_backoff.delay_seconds == 30 - assert phase_result.provider_backoff.retry_after == 7 - - -def test_submit_login_password_step_returns_continue_url(monkeypatch): - engine = _build_engine(monkeypatch) - engine.email = "tester@example.com" - engine.password = "Pass12345" - engine.session = FakePasswordSession( - FakeResponse(status_code=200, payload={"continue_url": "https://continue.example.test"}) - ) - monkeypatch.setattr(engine, "_check_sentinel", lambda did: None) - - step_result = engine._submit_login_password_step() - - assert step_result.success is True - assert step_result.http_status == 200 - assert step_result.continue_url == "https://continue.example.test" - assert engine.session.get_calls == [ - {"url": "https://continue.example.test", "timeout": 15} - ] - - -def test_otp_secondary_timeout_uses_independent_anchor_and_returns_explicit_error(monkeypatch): - engine = _build_engine(monkeypatch) - engine._is_existing_account = False - engine._otp_sent_at = 100.0 - - captured = {} - - monkeypatch.setattr( - register_module.time, - "time", - lambda: 500.0, - ) - monkeypatch.setattr( - engine, - "_submit_login_password_step", - lambda: SimpleNamespace(success=True, http_status=200), - ) - - def fake_get_verification_code(otp_sent_at=None, timeout=120): - captured["otp_sent_at"] = otp_sent_at - captured["timeout"] = timeout - return None - - monkeypatch.setattr(engine, "_get_verification_code", fake_get_verification_code) - - phase_result = engine._phase_otp_secondary( - RegistrationResult(success=False, logs=[]), - PhaseContext(reenter_ready=True), - ) - - assert captured == { - "otp_sent_at": 500.0, - "timeout": 120, - } - assert phase_result.success is False - assert phase_result.error_code == "OTP_TIMEOUT_SECONDARY" - assert phase_result.error_message == "等待第二次验证码超时" - assert phase_result.retryable is True - assert phase_result.next_action == "extend_timeout" - assert phase_result.data == {"otp_sent_at": 500.0} diff --git a/tests/test_tempmail_service.py b/tests/test_tempmail_service.py index 4a82398..c9c0e27 100644 --- a/tests/test_tempmail_service.py +++ b/tests/test_tempmail_service.py @@ -43,14 +43,14 @@ def test_get_verification_code_ignores_messages_not_newer_than_otp_anchor(monkey "from": "noreply@openai.com", "subject": "Old verification code", "body": "111111", - "date": 1999, + "received_at": 1999, }, { "id": "new-mail", "from": "noreply@openai.com", "subject": "New verification code", "body": "654321", - "date": 2001, + "received_at": 2001, }, ] },