bambuddy/backend/tests/unit/test_db_dialect.py

"""Unit tests for database dialect helpers and PostgreSQL compatibility."""

from unittest.mock import AsyncMock, patch

import pytest


class TestDialectDetection:
    """Test is_sqlite() and is_postgres() detection."""

    def test_sqlite_detected(self):
        with patch("backend.app.core.config.settings") as mock_settings:
            mock_settings.database_url = "sqlite+aiosqlite:///path/to/db.sqlite"
            from backend.app.core.db_dialect import is_postgres, is_sqlite

            assert is_sqlite() is True
            assert is_postgres() is False

    def test_postgres_detected(self):
        with patch("backend.app.core.config.settings") as mock_settings:
            mock_settings.database_url = "postgresql+asyncpg://user:pass@host:5432/db"
            from backend.app.core.db_dialect import is_postgres, is_sqlite

            assert is_postgres() is True
            assert is_sqlite() is False


class TestRunPragma:
    """Test that PRAGMAs only run on SQLite."""

    @pytest.mark.asyncio
    async def test_pragma_runs_on_sqlite(self):
        with patch("backend.app.core.db_dialect.is_sqlite", return_value=True):
            from backend.app.core.db_dialect import run_pragma

            mock_conn = AsyncMock()
            await run_pragma(mock_conn, "PRAGMA journal_mode = WAL")
            mock_conn.execute.assert_called_once()

    @pytest.mark.asyncio
    async def test_pragma_skipped_on_postgres(self):
        with patch("backend.app.core.db_dialect.is_sqlite", return_value=False):
            from backend.app.core.db_dialect import run_pragma

            mock_conn = AsyncMock()
            await run_pragma(mock_conn, "PRAGMA journal_mode = WAL")
            mock_conn.execute.assert_not_called()


class TestTimezoneStripping:
    """Test that the before_cursor_execute event strips timezone info."""

    def test_strip_aware_datetime(self):
        """Verify the timezone stripping logic works correctly."""
        import datetime

        aware = datetime.datetime(2026, 4, 3, 10, 0, 0, tzinfo=datetime.timezone.utc)
        naive = aware.replace(tzinfo=None)

        def _strip(val):
            if isinstance(val, datetime.datetime) and val.tzinfo is not None:
                return val.replace(tzinfo=None)
            return val

        assert _strip(aware) == naive
        assert _strip(aware).tzinfo is None
        assert _strip(naive) == naive
        assert _strip("not a datetime") == "not a datetime"
        assert _strip(None) is None

    def test_strip_in_dict_params(self):
        """Verify timezone stripping works on dict parameters."""
        import datetime

        aware = datetime.datetime(2026, 4, 3, 10, 0, 0, tzinfo=datetime.timezone.utc)

        def _strip(val):
            if isinstance(val, datetime.datetime) and val.tzinfo is not None:
                return val.replace(tzinfo=None)
            return val

        params = {"name": "test", "created_at": aware, "count": 5}
        result = {k: _strip(v) for k, v in params.items()}
        assert result["created_at"].tzinfo is None
        assert result["name"] == "test"
        assert result["count"] == 5

    def test_strip_in_tuple_params(self):
        """Verify timezone stripping works on tuple parameters."""
        import datetime

        aware = datetime.datetime(2026, 4, 3, 10, 0, 0, tzinfo=datetime.timezone.utc)

        def _strip(val):
            if isinstance(val, datetime.datetime) and val.tzinfo is not None:
                return val.replace(tzinfo=None)
            return val

        params = ("test", aware, 5)
        result = tuple(_strip(v) for v in params)
        assert result[1].tzinfo is None
        assert result[0] == "test"

    def test_naive_datetime_unchanged(self):
        """Naive datetimes should pass through untouched."""
        import datetime

        naive = datetime.datetime(2026, 4, 3, 10, 0, 0)

        def _strip(val):
            if isinstance(val, datetime.datetime) and val.tzinfo is not None:
                return val.replace(tzinfo=None)
            return val

        result = _strip(naive)
        assert result == naive
        assert result.tzinfo is None


class TestCrossDatabaseConversion:
    """Test SQLite→Postgres type conversion logic used in cross-database import."""

    def test_boolean_conversion(self):
        """SQLite stores booleans as 0/1, Postgres needs Python bool."""
        assert bool(0) is False
        assert bool(1) is True

    def test_datetime_string_conversion(self):
        """SQLite stores datetimes as strings, Postgres needs datetime objects."""
        from datetime import datetime

        val = "2026-04-02 11:01:52.105147"
        result = datetime.fromisoformat(val)
        assert result.year == 2026
        assert result.month == 4
        assert result.microsecond == 105147

    def test_datetime_with_timezone_string(self):
        """SQLite may store timezone-aware strings."""
        from datetime import datetime

        val = "2026-04-02T11:01:52+00:00"
        result = datetime.fromisoformat(val)
        assert result.year == 2026

    def test_json_serialization_for_backup(self):
        """JSON/list/dict values must be serialized for SQLite backup."""
        import json

        values = [{"key": "val"}, [1, 2, 3], "plain string", 42, None]
        for val in values:
            if isinstance(val, (list, dict)):
                serialized = json.dumps(val)
                assert isinstance(serialized, str)
            else:
                assert val == val  # noqa: PLR0124 — no conversion needed


class TestSafeExecutePattern:
    """Test _safe_execute error handling logic."""

    def test_safe_execute_catches_expected_exceptions(self):
        """Verify _safe_execute catches both OperationalError and ProgrammingError."""
        from sqlalchemy.exc import OperationalError, ProgrammingError

        for exc_type in (OperationalError, ProgrammingError):
            try:
                raise exc_type("test", [], Exception("column already exists"))
            except (OperationalError, ProgrammingError):
                pass

    def test_safe_execute_would_not_catch_integrity_error(self):
        """IntegrityError should NOT be caught by _safe_execute."""
        from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError

        with pytest.raises(IntegrityError):
            try:
                raise IntegrityError("test", [], Exception("unique violation"))
            except (OperationalError, ProgrammingError):
                pass

    @pytest.mark.asyncio
    async def test_safe_execute_reraises_non_idempotency_errors(self):
        """Non-idempotency errors must propagate so startup fails loudly."""
        from sqlalchemy.exc import OperationalError
        from sqlalchemy.ext.asyncio import create_async_engine

        from backend.app.core.database import _safe_execute

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            with pytest.raises(OperationalError):
                await _safe_execute(conn, "SELECT * FROM nonexistent_table_xyz")
        await engine.dispose()

    @pytest.mark.asyncio
    async def test_safe_execute_swallows_already_exists(self):
        """Idempotency errors (already exists) must be silently ignored."""
        from sqlalchemy import text
        from sqlalchemy.ext.asyncio import create_async_engine

        from backend.app.core.database import _safe_execute

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.execute(text("CREATE TABLE t (id INTEGER)"))
            # Second CREATE must not raise
            await _safe_execute(conn, "CREATE TABLE t (id INTEGER)")
        await engine.dispose()

    @pytest.mark.asyncio
    async def test_provider_email_lowercasing_migration(self):
        """SEC-3: provider_email normalisation lowers mixed-case values, leaves NULL intact.

        The production migration runs this UPDATE directly (not via _safe_execute)
        so any failure is always fatal and visible at startup.
        """
        from sqlalchemy import text
        from sqlalchemy.ext.asyncio import create_async_engine

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.execute(text("CREATE TABLE user_oidc_links (id INTEGER PRIMARY KEY, provider_email TEXT)"))
            await conn.execute(text("INSERT INTO user_oidc_links VALUES (1, 'User@Example.COM')"))
            await conn.execute(text("INSERT INTO user_oidc_links VALUES (2, 'already@lower.com')"))
            await conn.execute(text("INSERT INTO user_oidc_links VALUES (3, NULL)"))

            async with conn.begin_nested():
                await conn.execute(
                    text(
                        "UPDATE user_oidc_links SET provider_email = LOWER(provider_email) "
                        "WHERE provider_email IS NOT NULL AND provider_email != LOWER(provider_email)"
                    )
                )

            result = await conn.execute(text("SELECT provider_email FROM user_oidc_links ORDER BY id"))
            rows = [r[0] for r in result.fetchall()]
        await engine.dispose()

        assert rows[0] == "user@example.com"
        assert rows[1] == "already@lower.com"
        assert rows[2] is None

    @pytest.mark.asyncio
    async def test_safe_execute_swallows_no_such_column_for_rename(self):
        """'no such column' is swallowed for RENAME COLUMN idempotency.

        When a column has already been renamed, re-running the RENAME COLUMN
        migration raises 'no such column' — that must be silently swallowed.
        DML safety is guaranteed by never passing DML through _safe_execute.
        """
        from sqlalchemy.ext.asyncio import create_async_engine

        from backend.app.core.database import _safe_execute

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.execute(__import__("sqlalchemy").text("CREATE TABLE t (id INTEGER, new_col INTEGER)"))
            # Column 'old_col' does not exist — simulates re-running a RENAME COLUMN migration
            # Must NOT raise.
            await _safe_execute(conn, "ALTER TABLE t RENAME COLUMN old_col TO new_col")
        await engine.dispose()

    @pytest.mark.asyncio
    async def test_safe_execute_swallows_does_not_exist_for_rename_postgres(self):
        """'does not exist' (PostgreSQL UndefinedColumnError) is swallowed for RENAME COLUMN idempotency."""
        from unittest.mock import AsyncMock, MagicMock

        from sqlalchemy.exc import ProgrammingError

        from backend.app.core.database import _safe_execute

        fake_exc = ProgrammingError('column "quantity_printed" does not exist', [], Exception())

        nested_cm = MagicMock()
        nested_cm.__aenter__ = AsyncMock(return_value=nested_cm)
        nested_cm.execute = AsyncMock(side_effect=fake_exc)
        nested_cm.__aexit__ = AsyncMock(return_value=False)

        mock_conn = MagicMock()
        mock_conn.begin_nested.return_value = nested_cm
        mock_conn.execute = AsyncMock(side_effect=fake_exc)

        # Must NOT raise — "does not exist" is in the swallow-list
        await _safe_execute(
            mock_conn, "ALTER TABLE project_bom_items RENAME COLUMN quantity_printed TO quantity_acquired"
        )

    @pytest.mark.asyncio
    async def test_safe_execute_swallows_duplicate_key(self):
        """'duplicate key' errors (PostgreSQL unique-constraint violations on re-run)
        must be silently swallowed for idempotent DDL migrations."""
        from unittest.mock import AsyncMock, MagicMock

        from sqlalchemy.exc import OperationalError

        from backend.app.core.database import _safe_execute

        fake_exc = OperationalError("duplicate key value violates unique constraint", [], Exception())

        # begin_nested() is called synchronously (not awaited) and returns an
        # async context manager. Use MagicMock so the call returns a regular
        # object, then attach __aenter__/__aexit__ for the async with protocol.
        nested_cm = MagicMock()
        nested_cm.__aenter__ = AsyncMock(return_value=nested_cm)
        # Raise on execute inside the context, simulating PG duplicate key
        nested_cm.execute = AsyncMock(side_effect=fake_exc)
        nested_cm.__aexit__ = AsyncMock(return_value=False)

        mock_conn = MagicMock()
        mock_conn.begin_nested.return_value = nested_cm
        mock_conn.execute = AsyncMock(side_effect=fake_exc)

        # Must NOT raise — "duplicate key" is in the swallow-list
        await _safe_execute(mock_conn, "CREATE UNIQUE INDEX ...")

    @pytest.mark.asyncio
    async def test_check_constraint_false_true_on_sqlite(self):
        """CheckConstraint with FALSE/TRUE literals is enforced on SQLite (3.23+)."""
        from sqlalchemy import text
        from sqlalchemy.exc import IntegrityError
        from sqlalchemy.ext.asyncio import create_async_engine

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.execute(
                text("""
                CREATE TABLE ck_test (
                    id INTEGER PRIMARY KEY,
                    auto_link BOOLEAN,
                    require_ev BOOLEAN,
                    email_claim TEXT,
                    CHECK (auto_link = FALSE OR (require_ev = TRUE AND email_claim = 'email'))
                )
            """)
            )
            # Valid: auto_link=0 (FALSE)
            await conn.execute(text("INSERT INTO ck_test VALUES (1, 0, 0, 'upn')"))
            # Valid: auto_link=1, require_ev=1, email_claim='email'
            await conn.execute(text("INSERT INTO ck_test VALUES (2, 1, 1, 'email')"))

        async with engine.begin() as conn:
            # Invalid: auto_link=1 but conditions not met
            with pytest.raises(IntegrityError):
                await conn.execute(text("INSERT INTO ck_test VALUES (3, 1, 0, 'email')"))
        await engine.dispose()

    @pytest.mark.asyncio
    async def test_auto_link_sec1_backfill_resets_unsafe_rows(self):
        """SEC-1 backfill resets auto_link=TRUE on rows with unsafe combined state.

        Three cases:
          1. auto_link=TRUE + require_ev=FALSE → reset to FALSE (unsafe: permissive mode)
          2. auto_link=TRUE + custom claim → reset to FALSE (unsafe: no email_verified gate)
          3. auto_link=TRUE + require_ev=TRUE + standard claim → unchanged (safe)
        """
        from sqlalchemy import text
        from sqlalchemy.ext.asyncio import create_async_engine

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.execute(
                text(
                    "CREATE TABLE oidc_providers ("
                    "id INTEGER PRIMARY KEY, "
                    "auto_link_existing_accounts BOOLEAN, "
                    "require_email_verified BOOLEAN, "
                    "email_claim TEXT"
                    ")"
                )
            )
            # Row 1: unsafe — require_ev=FALSE
            await conn.execute(text("INSERT INTO oidc_providers VALUES (1, 1, 0, 'email')"))
            # Row 2: unsafe — custom claim
            await conn.execute(text("INSERT INTO oidc_providers VALUES (2, 1, 1, 'preferred_username')"))
            # Row 3: safe — require_ev=TRUE + standard claim
            await conn.execute(text("INSERT INTO oidc_providers VALUES (3, 1, 1, 'email')"))

            async with conn.begin_nested():
                await conn.execute(
                    text(
                        "UPDATE oidc_providers SET auto_link_existing_accounts = FALSE "
                        "WHERE auto_link_existing_accounts = TRUE "
                        "AND (require_email_verified = FALSE OR email_claim != 'email')"
                    )
                )

            result = await conn.execute(text("SELECT id, auto_link_existing_accounts FROM oidc_providers ORDER BY id"))
            rows = {r[0]: r[1] for r in result.fetchall()}
        await engine.dispose()

        assert rows[1] == 0, "unsafe (require_ev=FALSE) row must be reset to FALSE"
        assert rows[2] == 0, "unsafe (custom claim) row must be reset to FALSE"
        assert rows[3] == 1, "safe row must remain TRUE"

    @pytest.mark.asyncio
    async def test_safe_execute_reraises_does_not_exist_without_column(self):
        """'does not exist' without 'column' in the message must NOT be swallowed.

        This verifies that the narrowing from the broad 'does not exist' substring
        to the compound RENAME-COLUMN-only guard works correctly.  A missing-relation
        error must propagate so the operator sees a startup failure rather than a
        silent schema gap.
        """
        from unittest.mock import AsyncMock, MagicMock

        from sqlalchemy.exc import ProgrammingError

        from backend.app.core.database import _safe_execute

        # PostgreSQL error for a missing relation — contains "does not exist" but NOT "column"
        fake_exc = ProgrammingError('relation "oidc_providers" does not exist', [], Exception())

        nested_cm = MagicMock()
        nested_cm.__aenter__ = AsyncMock(return_value=nested_cm)
        nested_cm.execute = AsyncMock(side_effect=fake_exc)
        nested_cm.__aexit__ = AsyncMock(return_value=False)

        mock_conn = MagicMock()
        mock_conn.begin_nested.return_value = nested_cm
        mock_conn.execute = AsyncMock(side_effect=fake_exc)

        # Must RAISE — "column" is absent so this is not RENAME COLUMN idempotency
        with pytest.raises(ProgrammingError):
            await _safe_execute(
                mock_conn, "ALTER TABLE oidc_providers ADD COLUMN auto_link_existing_accounts BOOLEAN DEFAULT false"
            )

    @pytest.mark.asyncio
    async def test_oidc_boolean_default_migrations_sqlite_defaults(self):
        """auto_link defaults to 0 (FALSE) and require_email_verified defaults to 1 (TRUE) on SQLite.

        Verifies that the SQLite branch of the BOOLEAN DEFAULT dialect-branch uses
        the correct integer literals so new rows get safe defaults without explicit values.
        """
        from sqlalchemy import text
        from sqlalchemy.ext.asyncio import create_async_engine

        from backend.app.core.database import _safe_execute

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.execute(text("CREATE TABLE oidc_providers (id INTEGER PRIMARY KEY, name TEXT)"))
            await _safe_execute(
                conn, "ALTER TABLE oidc_providers ADD COLUMN auto_link_existing_accounts BOOLEAN DEFAULT 0"
            )
            await _safe_execute(conn, "ALTER TABLE oidc_providers ADD COLUMN require_email_verified BOOLEAN DEFAULT 1")
            await conn.execute(text("INSERT INTO oidc_providers (id, name) VALUES (1, 'test')"))
            result = await conn.execute(
                text("SELECT auto_link_existing_accounts, require_email_verified FROM oidc_providers WHERE id = 1")
            )
            row = result.fetchone()
        await engine.dispose()

        assert row[0] == 0, "auto_link_existing_accounts must default to 0 (FALSE) on SQLite"
        assert row[1] == 1, "require_email_verified must default to 1 (TRUE) on SQLite"

    @pytest.mark.asyncio
    async def test_safe_execute_column_not_exists_only_swallowed_for_rename(self):
        """'column … does not exist' is swallowed only when the SQL is RENAME COLUMN.

        The compound guard must NOT swallow the same error pattern when the SQL is
        an ADD COLUMN statement — that would indicate schema corruption, not idempotency.
        """
        from unittest.mock import AsyncMock, MagicMock

        from sqlalchemy.exc import ProgrammingError

        from backend.app.core.database import _safe_execute

        fake_exc = ProgrammingError('column "auto_link_existing_accounts" does not exist', [], Exception())

        nested_cm = MagicMock()
        nested_cm.__aenter__ = AsyncMock(return_value=nested_cm)
        nested_cm.execute = AsyncMock(side_effect=fake_exc)
        nested_cm.__aexit__ = AsyncMock(return_value=False)

        mock_conn = MagicMock()
        mock_conn.begin_nested.return_value = nested_cm
        mock_conn.execute = AsyncMock(side_effect=fake_exc)

        # ADD COLUMN statement — must RAISE even though message contains "column" + "does not exist"
        with pytest.raises(ProgrammingError):
            await _safe_execute(
                mock_conn, "ALTER TABLE oidc_providers ADD COLUMN auto_link_existing_accounts BOOLEAN DEFAULT false"
            )

        # RENAME COLUMN statement — must NOT raise (idempotency)
        await _safe_execute(
            mock_conn, "ALTER TABLE oidc_providers RENAME COLUMN auto_link_existing_accounts TO auto_link"
        )

    @pytest.mark.asyncio
    async def test_normalize_printer_ids_sqlite_uses_plain_comparison(self):
        """SQLite path executes plain string comparison (no cast)."""
        from sqlalchemy import text
        from sqlalchemy.ext.asyncio import create_async_engine

        from backend.app.core.database import _migrate_normalize_printer_ids

        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.execute(text("CREATE TABLE api_keys (id INTEGER PRIMARY KEY, printer_ids TEXT)"))
            await conn.execute(text("INSERT INTO api_keys VALUES (1, '[]')"))
            await conn.execute(text("INSERT INTO api_keys VALUES (2, '[1,2]')"))

            with patch("backend.app.core.database.is_sqlite", return_value=True):
                await _migrate_normalize_printer_ids(conn)

            result = await conn.execute(text("SELECT id, printer_ids FROM api_keys ORDER BY id"))
            rows = {r[0]: r[1] for r in result.fetchall()}
        await engine.dispose()

        assert rows[1] is None, "printer_ids='[]' must be normalised to NULL"
        assert rows[2] == "[1,2]", "non-empty printer_ids must be unchanged"

    @pytest.mark.asyncio
    async def test_normalize_printer_ids_postgres_uses_text_cast(self):
        """PostgreSQL path casts printer_ids to text for comparison (works for json and jsonb)."""
        from unittest.mock import AsyncMock, MagicMock

        from backend.app.core.database import _migrate_normalize_printer_ids

        nested_cm = MagicMock()
        nested_cm.__aenter__ = AsyncMock(return_value=nested_cm)
        nested_cm.__aexit__ = AsyncMock(return_value=False)
        mock_conn = MagicMock()
        mock_conn.begin_nested.return_value = nested_cm
        mock_conn.execute = AsyncMock()

        with patch("backend.app.core.database.is_sqlite", return_value=False):
            await _migrate_normalize_printer_ids(mock_conn)

        sql = mock_conn.execute.call_args[0][0].text
        assert "::text = '[]'" in sql, f"Expected ::text cast in SQL, got: {sql}"
        assert "printer_ids" in sql