签入版本

tests/integration/__init__.py

@@ -0,0 +1 @@
+"""集成测试。"""

tests/integration/conftest.py

@@ -0,0 +1,28 @@
+"""集成测试配置：MySQL/Redis 不可用或认证失败时自动跳过。"""
+import pytest
+
+
+def _services_available() -> bool:
+    """尝试真实连接 MySQL 与 Redis，任一不可用则跳过集成测试。"""
+    try:
+        # 数据库连通性 + 认证（select(1) 跨方言，Oracle 下自动渲染 SELECT 1 FROM DUAL）
+        from sqlalchemy import select
+        from extensions.sqlalchemy import engine
+        with engine.connect() as conn:
+            conn.execute(select(1))
+        # Redis 连通性
+        from core.redis import redis
+        redis.ping()
+        return True
+    except Exception:
+        return False
+
+
+SERVICES_AVAILABLE = _services_available()
+
+
+@pytest.fixture(autouse=True)
+def _skip_without_services(request):
+    """集成测试依赖 MySQL/Redis，服务不可用或认证失败时跳过。"""
+    if "integration" in request.keywords and not SERVICES_AVAILABLE:
+        pytest.skip("MySQL/Redis 服务不可用或认证失败，跳过集成测试")

tests/integration/test_api.py

@@ -0,0 +1,34 @@
+"""API 集成测试。"""
+import json
+
+import pytest
+from tornado.testing import AsyncHTTPTestCase
+
+# 使用 src 包路径导入工厂，避免与根目录入口 app.py 同名冲突
+from src.app import create_app
+
+pytestmark = pytest.mark.integration
+
+
+class ApiSmokeTest(AsyncHTTPTestCase):
+    def get_app(self):
+        return create_app()
+
+    def test_health(self):
+        resp = self.fetch("/api/v1/health")
+        assert resp.code == 200
+        body = json.loads(resp.body)
+        assert body["code"] == 0
+        assert body["data"]["status"] == "ok"
+
+    def test_captcha(self):
+        resp = self.fetch("/api/v1/captcha")
+        assert resp.code == 200
+        body = json.loads(resp.body)
+        assert body["code"] == 0
+        assert body["data"].startswith("data:image/png;base64,")
+
+    def test_protected_requires_token(self):
+        resp = self.fetch("/api/v1/user/list")
+        body = json.loads(resp.body)
+        assert body["code"] == 401

tests/integration/test_database.py

@@ -0,0 +1,64 @@
+"""数据库连接与 Repository 集成测试。"""
+import pytest
+from sqlalchemy import select
+
+from extensions.sqlalchemy import engine
+
+pytestmark = pytest.mark.integration
+
+
+def test_db_connect():
+    with engine.connect() as conn:
+        result = conn.execute(select(1))
+        assert result.scalar() == 1
+
+
+def test_repo_get_by_id_robust():
+    """get_by_id 对非法 ID 容错，对已存在 ID 正常返回。"""
+    from modules.role.repository import RoleRepository
+    repo = RoleRepository()
+    try:
+        assert repo.get_by_id(None) is None
+        assert repo.get_by_id("") is None
+        assert repo.get_by_id("abc") is None
+        assert repo.get_by_id(0) is None
+        assert repo.get_by_id(-1) is None
+        first = repo.filter_by().order_by(RoleRepository.model.id.asc()).first()
+        assert first is not None
+        assert repo.get_by_id(first.id) is not None
+    finally:
+        repo.close()
+
+
+def test_repo_exists_by_field():
+    """exists_by_field 唯一性校验（仅未删除，可排除自身）。"""
+    from modules.role.models import Role
+    from modules.role.repository import RoleRepository
+    repo = RoleRepository()
+    try:
+        role = repo.create(Role(name="唯一名XYZ", code="uniq_xyz", status=1, sort=0))
+        assert repo.exists_by_field("code", "uniq_xyz")                    # 自身命中
+        assert not repo.exists_by_field("code", "uniq_xyz", exclude_id=role.id)  # 排除自身后无重复
+        assert not repo.exists_by_field("code", "__不存在的编码__")
+        repo.db.query(Role).filter(Role.id == role.id).delete()
+        repo.db.commit()
+    finally:
+        repo.close()
+
+
+def test_repo_batch_delete():
+    """batch_delete 逗号分隔软删，返回实际删除条数。"""
+    from modules.role.models import Role
+    from modules.role.repository import RoleRepository
+    repo = RoleRepository()
+    try:
+        role = repo.create(Role(name="测试删除角色", code="test_del", status=1, sort=0))
+        assert repo.batch_delete(f"{role.id}") == 1
+        assert repo.get_by_id(role.id) is None          # 软删后不可见
+        assert repo.get_by_id(role.id, soft_delete=False) is not None  # 物理仍在
+        assert repo.batch_delete("abc,,0") == 0
+        # 清理
+        repo.db.query(Role).filter(Role.id == role.id).delete()
+        repo.db.commit()
+    finally:
+        repo.close()

tests/integration/test_redis.py

@@ -0,0 +1,15 @@
+"""Redis 集成测试。"""
+import pytest
+
+from core.redis import redis
+
+pytestmark = pytest.mark.integration
+
+TEST_KEY = "test:djangoadmin:ping"
+
+
+def test_redis_set_get():
+    redis.set(TEST_KEY, "pong")
+    assert redis.get(TEST_KEY) == "pong"
+    redis.delete(TEST_KEY)
+    assert redis.get(TEST_KEY) is None