From bbdb53e22668c330bbe2399d15a85da640db4d93 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 18 Aug 2026 15:14:27 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E6=B7=BB=E5=8A=A0=20health.py=20psycop?= =?UTF-8?q?g3=20=E8=BF=9E=E6=8E=A5=E6=A3=80=E6=9F=A5=E5=8D=95=E5=85=83?= =?UTF-8?q?=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 覆盖 _check_database() 和 _check_migrations() 中的 psycopg.connect 调用： - 成功连接路径（覆盖 lines 52, 127） - 连接失败路径 - 内存数据库跳过路径 - startup_check 端点综合测试 预期增量覆盖率从 33% 提升至 100% --- tests/unit/test_health_routes.py | 184 +++++++++++++++++++++++++++++++ 1 file changed, 184 insertions(+) create mode 100644 tests/unit/test_health_routes.py 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