xiaoxia-saas/tests/unit/test_health_routes.py

"""Unit tests for apps/api/app/api/routes/health.py

覆盖 _check_database() 和 _check_migrations() 中 psycopg3 连接逻辑。
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest


def _make_cursor(fetchone_result=None):
    """Create a mock cursor with context manager support."""
    cur = MagicMock()
    cur.__enter__ = MagicMock(return_value=cur)
    cur.__exit__ = MagicMock(return_value=False)
    if fetchone_result is not None:
        cur.fetchone.return_value = fetchone_result
    return cur


def _make_conn(cursor_result=None):
    conn = MagicMock()
    conn.cursor.return_value = cursor_result or _make_cursor()
    return conn


@pytest.mark.asyncio
class TestCheckDatabase:

    async def test_check_database_success(self):
        mock_cur = _make_cursor(fetchone_result=(1,))
        mock_conn = _make_conn(mock_cur)
        mock_settings = MagicMock()
        mock_settings.USE_IN_MEMORY_DB = False
        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"

        from apps.api.app.api.routes import health

        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
            mock_psycopg.connect.return_value = mock_conn
            result = await health._check_database()

        assert result["status"] == "healthy"
        assert result["type"] == "postgresql"
        assert result["message"] == "Database connection successful"
        mock_psycopg.connect.assert_called_once_with(
            "postgresql://test:test@localhost/test", connect_timeout=3
        )
        mock_cur.execute.assert_called_once_with("SELECT 1")
        mock_conn.close.assert_called_once()

    async def test_check_database_connection_failure(self):
        mock_settings = MagicMock()
        mock_settings.USE_IN_MEMORY_DB = False
        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"

        from apps.api.app.api.routes import health

        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
            mock_psycopg.connect.side_effect = Exception("connection refused")
            result = await health._check_database()

        assert result["status"] == "unhealthy"
        assert result["type"] == "postgresql"
        assert "connection refused" in result["message"]

    async def test_check_database_in_memory(self):
        mock_settings = MagicMock()
        mock_settings.USE_IN_MEMORY_DB = True

        from apps.api.app.api.routes import health

        with patch.object(health, "settings", mock_settings):
            result = await health._check_database()

        assert result["status"] == "healthy"
        assert result["type"] == "in_memory"


@pytest.mark.asyncio
class TestCheckMigrations:

    async def test_check_migrations_success(self):
        mock_cur = _make_cursor(fetchone_result=(5,))
        mock_conn = _make_conn(mock_cur)
        mock_settings = MagicMock()
        mock_settings.USE_IN_MEMORY_DB = False
        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"

        from apps.api.app.api.routes import health

        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
            mock_psycopg.connect.return_value = mock_conn
            result = await health._check_migrations()

        assert result["status"] == "healthy"
        assert result["message"] == "Database migrations applied"
        mock_psycopg.connect.assert_called_once_with(
            "postgresql://test:test@localhost/test", connect_timeout=3
        )
        mock_conn.close.assert_called_once()

    async def test_check_migrations_missing_tables(self):
        mock_cur = _make_cursor(fetchone_result=(2,))
        mock_conn = _make_conn(mock_cur)
        mock_settings = MagicMock()
        mock_settings.USE_IN_MEMORY_DB = False
        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"

        from apps.api.app.api.routes import health

        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
            mock_psycopg.connect.return_value = mock_conn
            result = await health._check_migrations()

        assert result["status"] == "unhealthy"
        assert "Missing tables" in result["message"]
        assert "2/5" in result["message"]

    async def test_check_migrations_connection_failure(self):
        mock_settings = MagicMock()
        mock_settings.USE_IN_MEMORY_DB = False
        mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"

        from apps.api.app.api.routes import health

        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
            mock_psycopg.connect.side_effect = Exception("connection refused")
            result = await health._check_migrations()

        assert result["status"] == "unhealthy"
        assert "Migration check failed" in result["message"]

    async def test_check_migrations_in_memory(self):
        mock_settings = MagicMock()
        mock_settings.USE_IN_MEMORY_DB = True

        from apps.api.app.api.routes import health

        with patch.object(health, "settings", mock_settings):
            result = await health._check_migrations()

        assert result["status"] == "healthy"
        assert "no migrations needed" in result["message"]


@pytest.mark.asyncio
class TestStartupCheck:

    async def test_startup_all_healthy(self):
        from apps.api.app.api.routes import health

        with patch.object(health, "_check_migrations", new_callable=AsyncMock) as mock_mig,              patch.object(health, "_check_database", new_callable=AsyncMock) as mock_db:
            mock_db.return_value = {"status": "healthy"}
            mock_mig.return_value = {"status": "healthy"}
            result = await health.startup_check()

        assert result["status"] == "started"

    async def test_startup_db_unhealthy(self):
        import json

        from apps.api.app.api.routes import health

        with patch.object(health, "_check_migrations", new_callable=AsyncMock) as mock_mig,              patch.object(health, "_check_database", new_callable=AsyncMock) as mock_db:
            mock_db.return_value = {"status": "unhealthy", "message": "fail"}
            mock_mig.return_value = {"status": "healthy"}
            result = await health.startup_check()

        assert result.status_code == 503
        body = json.loads(result.body)
        assert body["status"] == "starting"