test: 添加 health.py psycopg3 连接检查单元测试
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 46s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m15s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m3s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m8s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 54s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m13s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m36s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 46s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m15s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m3s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m8s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 54s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m13s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m36s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
覆盖 _check_database() 和 _check_migrations() 中的 psycopg.connect 调用: - 成功连接路径(覆盖 lines 52, 127) - 连接失败路径 - 内存数据库跳过路径 - startup_check 端点综合测试 预期增量覆盖率从 33% 提升至 100%
This commit is contained in:
@@ -0,0 +1,184 @@
|
||||
"""Unit tests for apps/api/app/api/routes/health.py
|
||||
|
||||
覆盖 _check_database() 和 _check_migrations() 中 psycopg3 连接逻辑。
|
||||
确保增量覆盖率 ≥ 60%(目标覆盖 lines 52, 127)。
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCheckDatabase:
|
||||
"""Tests for _check_database() health check function."""
|
||||
|
||||
@patch("apps.api.app.api.routes.health.settings")
|
||||
@patch("apps.api.app.api.routes.health.psycopg.connect")
|
||||
async def test_check_database_success(self, mock_connect, mock_settings):
|
||||
"""PostgreSQL 连接成功时返回 healthy。"""
|
||||
mock_settings.USE_IN_MEMORY_DB = False
|
||||
mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
|
||||
|
||||
# Mock connection and cursor
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_cursor.__exit__ = MagicMock(return_value=False)
|
||||
mock_cursor.fetchone.return_value = (1,)
|
||||
mock_conn.cursor.return_value = mock_cursor
|
||||
mock_connect.return_value = mock_conn
|
||||
|
||||
from apps.api.app.api.routes.health import _check_database
|
||||
|
||||
result = await _check_database()
|
||||
|
||||
assert result["status"] == "healthy"
|
||||
assert result["type"] == "postgresql"
|
||||
assert result["message"] == "Database connection successful"
|
||||
mock_connect.assert_called_once_with(
|
||||
"postgresql+psycopg://test:test@localhost/test", connect_timeout=3
|
||||
)
|
||||
mock_cursor.execute.assert_called_once_with("SELECT 1")
|
||||
mock_conn.close.assert_called_once()
|
||||
|
||||
@patch("apps.api.app.api.routes.health.settings")
|
||||
@patch("apps.api.app.api.routes.health.psycopg.connect")
|
||||
async def test_check_database_connection_failure(self, mock_connect, mock_settings):
|
||||
"""PostgreSQL 连接失败时返回 unhealthy。"""
|
||||
mock_settings.USE_IN_MEMORY_DB = False
|
||||
mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
|
||||
mock_connect.side_effect = Exception("connection refused")
|
||||
|
||||
from apps.api.app.api.routes.health import _check_database
|
||||
|
||||
result = await _check_database()
|
||||
|
||||
assert result["status"] == "unhealthy"
|
||||
assert result["type"] == "postgresql"
|
||||
assert "connection refused" in result["message"]
|
||||
|
||||
@patch("apps.api.app.api.routes.health.settings")
|
||||
async def test_check_database_in_memory(self, mock_settings):
|
||||
"""使用内存数据库时跳过 PostgreSQL 检查。"""
|
||||
mock_settings.USE_IN_MEMORY_DB = True
|
||||
|
||||
from apps.api.app.api.routes.health import _check_database
|
||||
|
||||
result = await _check_database()
|
||||
|
||||
assert result["status"] == "healthy"
|
||||
assert result["type"] == "in_memory"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCheckMigrations:
|
||||
"""Tests for _check_migrations() health check function."""
|
||||
|
||||
@patch("apps.api.app.api.routes.health.settings")
|
||||
@patch("apps.api.app.api.routes.health.psycopg.connect")
|
||||
async def test_check_migrations_success(self, mock_connect, mock_settings):
|
||||
"""所有迁移表存在时返回 healthy。"""
|
||||
mock_settings.USE_IN_MEMORY_DB = False
|
||||
mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_cursor.__exit__ = MagicMock(return_value=False)
|
||||
mock_cursor.fetchone.return_value = (5,) # 5 tables found
|
||||
mock_conn.cursor.return_value = mock_cursor
|
||||
mock_connect.return_value = mock_conn
|
||||
|
||||
from apps.api.app.api.routes.health import _check_migrations
|
||||
|
||||
result = await _check_migrations()
|
||||
|
||||
assert result["status"] == "healthy"
|
||||
assert result["message"] == "Database migrations applied"
|
||||
mock_connect.assert_called_once_with(
|
||||
"postgresql+psycopg://test:test@localhost/test", connect_timeout=3
|
||||
)
|
||||
mock_conn.close.assert_called_once()
|
||||
|
||||
@patch("apps.api.app.api.routes.health.settings")
|
||||
@patch("apps.api.app.api.routes.health.psycopg.connect")
|
||||
async def test_check_migrations_missing_tables(self, mock_connect, mock_settings):
|
||||
"""迁移表不完整时返回 unhealthy。"""
|
||||
mock_settings.USE_IN_MEMORY_DB = False
|
||||
mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_cursor.__exit__ = MagicMock(return_value=False)
|
||||
mock_cursor.fetchone.return_value = (2,) # Only 2 of 5 tables
|
||||
mock_conn.cursor.return_value = mock_cursor
|
||||
mock_connect.return_value = mock_conn
|
||||
|
||||
from apps.api.app.api.routes.health import _check_migrations
|
||||
|
||||
result = await _check_migrations()
|
||||
|
||||
assert result["status"] == "unhealthy"
|
||||
assert "Missing tables" in result["message"]
|
||||
assert "2/5" in result["message"]
|
||||
|
||||
@patch("apps.api.app.api.routes.health.settings")
|
||||
@patch("apps.api.app.api.routes.health.psycopg.connect")
|
||||
async def test_check_migrations_connection_failure(self, mock_connect, mock_settings):
|
||||
"""数据库连接失败时返回 unhealthy。"""
|
||||
mock_settings.USE_IN_MEMORY_DB = False
|
||||
mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
|
||||
mock_connect.side_effect = Exception("connection refused")
|
||||
|
||||
from apps.api.app.api.routes.health import _check_migrations
|
||||
|
||||
result = await _check_migrations()
|
||||
|
||||
assert result["status"] == "unhealthy"
|
||||
assert "Migration check failed" in result["message"]
|
||||
|
||||
@patch("apps.api.app.api.routes.health.settings")
|
||||
async def test_check_migrations_in_memory(self, mock_settings):
|
||||
"""使用内存数据库时跳过迁移检查。"""
|
||||
mock_settings.USE_IN_MEMORY_DB = True
|
||||
|
||||
from apps.api.app.api.routes.health import _check_migrations
|
||||
|
||||
result = await _check_migrations()
|
||||
|
||||
assert result["status"] == "healthy"
|
||||
assert "no migrations needed" in result["message"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestStartupCheck:
|
||||
"""Tests for startup_check() endpoint."""
|
||||
|
||||
@patch("apps.api.app.api.routes.health._check_migrations")
|
||||
@patch("apps.api.app.api.routes.health._check_database")
|
||||
async def test_startup_all_healthy(self, mock_db, mock_mig):
|
||||
"""所有检查通过时返回 started。"""
|
||||
mock_db.return_value = {"status": "healthy"}
|
||||
mock_mig.return_value = {"status": "healthy"}
|
||||
|
||||
from apps.api.app.api.routes.health import startup_check
|
||||
|
||||
result = await startup_check()
|
||||
|
||||
assert result["status"] == "started"
|
||||
|
||||
@patch("apps.api.app.api.routes.health._check_migrations")
|
||||
@patch("apps.api.app.api.routes.health._check_database")
|
||||
async def test_startup_db_unhealthy(self, mock_db, mock_mig):
|
||||
"""数据库不健康时返回 starting + 503。"""
|
||||
mock_db.return_value = {"status": "unhealthy", "message": "fail"}
|
||||
mock_mig.return_value = {"status": "healthy"}
|
||||
|
||||
from apps.api.app.api.routes.health import startup_check
|
||||
|
||||
result = await startup_check()
|
||||
|
||||
assert result["status"] == "starting"
|
||||
assert result.status_code == 503
|
||||
Reference in New Issue
Block a user