feat(phase6): 测试补全 + 集成测试 + repository 落地示范
- tests/unit/modules/test_repository.py:base_repository 在 SQLite 内存库的 CRUD 测试 - tests/integration/api/test_smoke.py:登录链路 / 只读接口 / 未授权拦截(标记 integration,默认跳过) - conftest 提供 sqlite_db(隔离 DB)与 app/client/admin_headers(集成)fixture - pytest.ini 默认跳过 integration,Makefile test-integration 可显式运行 - 修复 base_repository.get_by_id/get_by_ids 对字符串 ID 的兼容(URL 路径参数) - 示范 modules/system/level service 改用 level_repo,CRUD 全链路实测通过 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
"""集成冒烟测试:登录链路 + 各模块只读接口(需 MySQL/Redis 运行,默认跳过)。"""
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_flow(client):
|
||||
"""完整登录链路:/captcha → 解析验证码 → /login 获取 token。"""
|
||||
import redis as redis_lib
|
||||
|
||||
from config import REDIS_INDEX
|
||||
|
||||
r = await client.get("/captcha")
|
||||
body = r.json()
|
||||
assert body["code"] == 0
|
||||
idkey = body.get("idkey")
|
||||
assert idkey
|
||||
|
||||
# 从 Redis 读取验证码明文
|
||||
redis_client = redis_lib.Redis(db=REDIS_INDEX)
|
||||
code = redis_client.get(idkey).decode()
|
||||
|
||||
resp = await client.post("/login", json={
|
||||
"username": "admin", "password": "123456",
|
||||
"captcha": code, "idKey": idkey,
|
||||
})
|
||||
login = resp.json()
|
||||
assert login["code"] == 0
|
||||
assert login["data"]["access_token"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_read_endpoints(client, admin_headers):
|
||||
"""管理员 token 可访问代表性只读接口。"""
|
||||
for path in ("/user/page", "/dept/list", "/menu/list", "/index/menu", "/job/page"):
|
||||
resp = await client.get(path, headers=admin_headers)
|
||||
assert resp.json()["code"] == 0, f"{path} failed"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unauthorized_blocked(client):
|
||||
"""无 token 访问受保护接口应返回 401。"""
|
||||
resp = await client.get("/user/page")
|
||||
assert resp.json()["code"] == 401
|
||||
@@ -0,0 +1,32 @@
|
||||
"""集成测试 fixture:真实 FastAPI 应用 + ASGI client(需 MySQL/Redis 运行)。"""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
|
||||
from core import jwt as core_jwt
|
||||
from core.app import create_app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app():
|
||||
return create_app()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(app):
|
||||
# 手动运行 Redis lifespan(当前 httpx 版本不支持 ASGITransport lifespan 参数)
|
||||
from core.redis_client import redis_lifespan
|
||||
|
||||
transport = httpx.ASGITransport(app=app)
|
||||
async with redis_lifespan(app):
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_headers():
|
||||
"""userId=1 管理员 token(绕过 permission_required 权限校验)。"""
|
||||
token = core_jwt.create_token(
|
||||
{"userId": 1, "username": "admin", "realname": "admin"})
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
Reference in New Issue
Block a user