diff --git a/tests/unit/test_health_routes.py b/tests/unit/test_health_routes.py new file mode 100644 index 000000000..c74cd2de6 --- /dev/null +++ b/tests/unit/test_health_routes.py @@ -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