xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 16 Pull Requests 12 Actions 34.2k Packages Projects Releases 7 Wiki Activity

style: auto-fix black formatting

Browse Source
This commit is contained in:
xiaoxia
2026-07-22 18:23:49 +08:00
parent 37c74616f9
commit ed02631fdd
4 changed files with 35 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_generation_tasks.py
+6 -18
View File
@@ -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):
"""空结果"""
tests/unit/test_jobs.py
+1 -3
View File
@@ -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)
tests/unit/test_verification_code_service.py
+14 -36
View File
@@ -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
tests/unit/test_wechat_oauth_service.py
+14 -43
View File
@@ -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: