diff --git a/tests/unit/test_generation_tasks.py b/tests/unit/test_generation_tasks.py index a4c23206f..bd74fd9e5 100755 --- a/tests/unit/test_generation_tasks.py +++ b/tests/unit/test_generation_tasks.py @@ -126,9 +126,7 @@ class TestCreateGenerationTaskUseCase: """ID 会自动生成""" mock_repo.create.side_effect = lambda t: t - cmd = CreateGenerationTaskCommand( - project_id="proj-1", asset_library_id="lib-1" - ) + cmd = CreateGenerationTaskCommand(project_id="proj-1", asset_library_id="lib-1") uc = CreateGenerationTaskUseCase(mock_repo) task = uc.execute(cmd) @@ -186,12 +184,8 @@ class TestListUserTasksFilteredUseCase: assert isinstance(result, ListGenerationTasksResult) assert len(result.items) == 2 assert result.total == 2 - mock_repo.list_by_user_filtered.assert_called_once_with( - "user-1", status=None, limit=None, offset=0 - ) - mock_repo.count_by_user_filtered.assert_called_once_with( - "user-1", status=None - ) + mock_repo.list_by_user_filtered.assert_called_once_with("user-1", status=None, limit=None, offset=0) + mock_repo.count_by_user_filtered.assert_called_once_with("user-1", status=None) def test_list_with_status_filter(self, mock_repo): """按状态筛选""" @@ -201,12 +195,8 @@ class TestListUserTasksFilteredUseCase: uc = ListUserTasksFilteredUseCase(mock_repo) uc.execute("user-1", status="running") - mock_repo.list_by_user_filtered.assert_called_once_with( - "user-1", status="running", limit=None, offset=0 - ) - mock_repo.count_by_user_filtered.assert_called_once_with( - "user-1", status="running" - ) + mock_repo.list_by_user_filtered.assert_called_once_with("user-1", status="running", limit=None, offset=0) + mock_repo.count_by_user_filtered.assert_called_once_with("user-1", status="running") def test_list_with_pagination(self, mock_repo): """分页查询""" @@ -217,9 +207,7 @@ class TestListUserTasksFilteredUseCase: result = uc.execute("user-1", limit=10, offset=20) assert result.total == 100 - mock_repo.list_by_user_filtered.assert_called_once_with( - "user-1", status=None, limit=10, offset=20 - ) + mock_repo.list_by_user_filtered.assert_called_once_with("user-1", status=None, limit=10, offset=20) def test_list_empty_result(self, mock_repo): """空结果""" diff --git a/tests/unit/test_jobs.py b/tests/unit/test_jobs.py index 61ba82790..c9d178c47 100755 --- a/tests/unit/test_jobs.py +++ b/tests/unit/test_jobs.py @@ -183,9 +183,7 @@ class TestUpdateJobProgressUseCase: mock_repo.get.return_value = job mock_repo.update.side_effect = lambda j: j - cmd = UpdateJobProgressCommand( - job_id=job.id, progress=50.0, current_stage="处理中" - ) + cmd = UpdateJobProgressCommand(job_id=job.id, progress=50.0, current_stage="处理中") uc = UpdateJobProgressUseCase(mock_repo) result = uc.execute(cmd) diff --git a/tests/unit/test_verification_code_service.py b/tests/unit/test_verification_code_service.py index 978473dc0..87549bb0f 100755 --- a/tests/unit/test_verification_code_service.py +++ b/tests/unit/test_verification_code_service.py @@ -120,9 +120,7 @@ class TestGenerateNormal: mock_repo.find_latest.return_value = None mock_repo.count_today.return_value = 0 - code, err = service.generate( - "test@example.com", CODE_TYPE_EMAIL_BIND, custom_code="888888" - ) + code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND, custom_code="888888") assert err is None assert code.code == "888888" @@ -132,9 +130,7 @@ class TestGenerateNormal: mock_repo.find_latest.return_value = None mock_repo.count_today.return_value = 0 - code, err = service.generate( - "test@example.com", CODE_TYPE_EMAIL_BIND, ttl_seconds=60 - ) + code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND, ttl_seconds=60) assert err is None # 过期时间 - 创建时间 ≈ 60 秒 @@ -174,9 +170,7 @@ class TestGenerateRateLimit: def test_resend_cooldown_blocked(self, service, mock_repo): """冷却期内发送被拒绝""" # 10 秒前刚发过一条 - recent = make_code( - created_at=datetime.now(timezone.utc) - timedelta(seconds=10) - ) + recent = make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=10)) mock_repo.find_latest.return_value = recent mock_repo.count_today.return_value = 1 @@ -188,6 +182,7 @@ class TestGenerateRateLimit: # 等待时间应接近 50 秒（60-10） # 提取数字验证范围 import re + match = re.search(r"(\d+)\s*秒", err) assert match wait = int(match.group(1)) @@ -196,9 +191,7 @@ class TestGenerateRateLimit: def test_resend_after_cooldown_ok(self, service, mock_repo): """超过冷却期可以重发""" # 2 分钟前发的，已过冷却 - old = make_code( - created_at=datetime.now(timezone.utc) - timedelta(seconds=120) - ) + old = make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=120)) mock_repo.find_latest.return_value = old mock_repo.count_today.return_value = 1 @@ -210,9 +203,7 @@ class TestGenerateRateLimit: def test_daily_limit_reached(self, service, mock_repo): """达到每日上限""" # 没有最近的（过了冷却），但今日已达上限 - old = make_code( - created_at=datetime.now(timezone.utc) - timedelta(hours=2) - ) + old = make_code(created_at=datetime.now(timezone.utc) - timedelta(hours=2)) mock_repo.find_latest.return_value = old mock_repo.count_today.return_value = DAILY_LIMIT @@ -253,13 +244,9 @@ class TestGenerateCustomRateLimitParams: def test_custom_cooldown(self, mock_repo): """自定义冷却时间""" - svc = VerificationCodeService( - repo=mock_repo, resend_cooldown=300, daily_limit=5 - ) + svc = VerificationCodeService(repo=mock_repo, resend_cooldown=300, daily_limit=5) # 60 秒前发的，默认冷却 60 秒就够了，但这里设了 300 秒 - recent = make_code( - created_at=datetime.now(timezone.utc) - timedelta(seconds=60) - ) + recent = make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=60)) mock_repo.find_latest.return_value = recent mock_repo.count_today.return_value = 1 @@ -270,9 +257,7 @@ class TestGenerateCustomRateLimitParams: def test_custom_daily_limit(self, mock_repo): """自定义每日上限""" - svc = VerificationCodeService( - repo=mock_repo, resend_cooldown=60, daily_limit=3 - ) + svc = VerificationCodeService(repo=mock_repo, resend_cooldown=60, daily_limit=3) mock_repo.find_latest.return_value = None mock_repo.count_today.return_value = 3 @@ -308,9 +293,7 @@ class TestVerifyParamValidation: mock_repo.find_latest.return_value = code mock_repo.count_today.return_value = 0 - ok, err = service.verify( - " test@example.com ", CODE_TYPE_EMAIL_BIND, " 123456 " - ) + ok, err = service.verify(" test@example.com ", CODE_TYPE_EMAIL_BIND, " 123456 ") assert ok assert err is None @@ -329,9 +312,7 @@ class TestVerifyNormal: code = make_code(code="123456") mock_repo.find_latest.return_value = code - ok, err = service.verify( - "test@example.com", CODE_TYPE_EMAIL_BIND, "123456", consume=True - ) + ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456", consume=True) assert ok assert err is None @@ -344,9 +325,7 @@ class TestVerifyNormal: code = make_code(code="123456") mock_repo.find_latest.return_value = code - ok, err = service.verify( - "test@example.com", CODE_TYPE_EMAIL_BIND, "123456", consume=False - ) + ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456", consume=False) assert ok assert err is None @@ -438,9 +417,7 @@ class TestVerifyCodeTypeIsolation: mock_repo.find_latest.side_effect = side_effect - ok, err = service.verify( - "test@example.com", CODE_TYPE_EMAIL_BIND, "123456" - ) + ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456") assert not ok assert "不存在或已过期" in err @@ -468,4 +445,5 @@ class TestConstants: def test_valid_code_types_count(self): """5 种验证码类型""" from packages.application.auth.verification_code_service import VALID_CODE_TYPES + assert len(VALID_CODE_TYPES) == 5 diff --git a/tests/unit/test_wechat_oauth_service.py b/tests/unit/test_wechat_oauth_service.py index f4f67e392..c9783662f 100755 --- a/tests/unit/test_wechat_oauth_service.py +++ b/tests/unit/test_wechat_oauth_service.py @@ -21,7 +21,6 @@ from packages.application.auth.wechat_oauth_service import ( get_wechat_oauth_service, ) - # ============================================================ # MemoryStateStore # ============================================================ @@ -81,30 +80,22 @@ class TestIsConfigured: def test_fully_configured(self): """三项都配置了""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") assert svc.is_configured() is True def test_missing_app_id(self): """缺 app_id""" - svc = WechatOAuthService( - app_id="", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="", app_secret="secret", redirect_uri="https://example.com/cb") assert svc.is_configured() is False def test_missing_app_secret(self): """缺 app_secret""" - svc = WechatOAuthService( - app_id="wx123", app_secret="", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="", redirect_uri="https://example.com/cb") assert svc.is_configured() is False def test_missing_redirect_uri(self): """缺 redirect_uri""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="") assert svc.is_configured() is False def test_none_configured(self): @@ -123,9 +114,7 @@ class TestGenerateAuthUrl: def test_configured_mode(self): """配置完整时生成正式微信授权链接""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") url, state = svc.generate_auth_url() assert "open.weixin.qq.com" in url @@ -149,17 +138,13 @@ class TestGenerateAuthUrl: def test_custom_scope(self): """自定义 scope""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") url, _ = svc.generate_auth_url(scope="snsapi_userinfo") assert "scope=snsapi_userinfo" in url def test_state_is_unique(self): """每次生成的 state 不同""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") _, state1 = svc.generate_auth_url() _, state2 = svc.generate_auth_url() assert state1 != state2 @@ -187,27 +172,21 @@ class TestHandleCallback: def test_missing_code(self): """缺少授权码""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") user_info, err = svc.handle_callback("", "some-state") assert user_info is None assert "缺少授权码" in err def test_invalid_state(self): """state 无效或已过期""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") user_info, err = svc.handle_callback("code123", "invalid-state") assert user_info is None assert "state" in err def test_empty_state(self): """空 state""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") user_info, err = svc.handle_callback("code123", "") assert user_info is None assert "state" in err @@ -228,9 +207,7 @@ class TestHandleCallback: def test_configured_mode_success(self): """配置完整时正常调用微信 API""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") _, state = svc.generate_auth_url() with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get: @@ -265,9 +242,7 @@ class TestHandleCallback: def test_access_token_failed(self): """access_token 接口返回错误""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") _, state = svc.generate_auth_url() with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get: @@ -286,9 +261,7 @@ class TestHandleCallback: def test_userinfo_failed(self): """userinfo 接口返回错误""" - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") _, state = svc.generate_auth_url() with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get: @@ -313,9 +286,7 @@ class TestHandleCallback: """网络异常""" import requests - svc = WechatOAuthService( - app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb" - ) + svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb") _, state = svc.generate_auth_url() with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get: