FabledScribe/tests/test_services_db_maintenance.py

"""Tests for services/db_maintenance.py — mocks the engine (no real DB).

Verifies that VACUUM (ANALYZE) is issued once per allowlisted table, that the
allowlist is closed (an arbitrary name can't be vacuumed), and that a single
table failure doesn't abort the rest of the sweep.
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from scribe.services.db_maintenance import MAINTENANCE_TABLES, run_maintenance


def _mock_engine(exec_side_effect=None):
    """An engine whose connect() yields a conn with a recording exec_driver_sql."""
    conn = MagicMock()
    autocommit = MagicMock()
    autocommit.exec_driver_sql = AsyncMock(side_effect=exec_side_effect)
    conn.execution_options.return_value = autocommit  # sync on AsyncConnection
    cm = AsyncMock()
    cm.__aenter__ = AsyncMock(return_value=conn)
    cm.__aexit__ = AsyncMock(return_value=False)
    engine = MagicMock()
    engine.connect.return_value = cm
    return engine, autocommit


@pytest.mark.asyncio
async def test_vacuums_each_allowlisted_table_once():
    engine, autocommit = _mock_engine()
    with patch("scribe.services.db_maintenance.engine", engine), \
         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
        summary = await run_maintenance()

    issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
    assert issued == [f"VACUUM (ANALYZE) {t}" for t in MAINTENANCE_TABLES]
    assert all(r["ok"] for r in summary["tables"])
    assert len(summary["tables"]) == len(MAINTENANCE_TABLES)
    # AUTOCOMMIT was requested — VACUUM cannot run inside a transaction.
    conn = engine.connect.return_value.__aenter__.return_value
    conn.execution_options.assert_called_once_with(isolation_level="AUTOCOMMIT")


@pytest.mark.asyncio
async def test_allowlist_is_closed():
    """A caller-supplied name not on the allowlist is silently ignored."""
    engine, autocommit = _mock_engine()
    with patch("scribe.services.db_maintenance.engine", engine), \
         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
        summary = await run_maintenance(tables=["app_logs", "users; DROP TABLE notes"])

    issued = [c.args[0] for c in autocommit.exec_driver_sql.await_args_list]
    assert issued == ["VACUUM (ANALYZE) app_logs"]
    assert [r["table"] for r in summary["tables"]] == ["app_logs"]


@pytest.mark.asyncio
async def test_one_table_failure_does_not_abort_the_rest():
    # First table raises, the remaining tables still get vacuumed.
    calls = {"n": 0}

    def flaky(_sql):
        calls["n"] += 1
        if calls["n"] == 1:
            raise RuntimeError("boom")
        return MagicMock()

    engine, autocommit = _mock_engine(exec_side_effect=flaky)
    with patch("scribe.services.db_maintenance.engine", engine), \
         patch("scribe.services.db_maintenance.set_admin_setting", AsyncMock()):
        summary = await run_maintenance()

    assert autocommit.exec_driver_sql.await_count == len(MAINTENANCE_TABLES)
    assert summary["tables"][0]["ok"] is False
    assert summary["tables"][0]["error"] == "boom"
    assert all(r["ok"] for r in summary["tables"][1:])


def _health_session(db_bytes, rows):
    s = AsyncMock()
    s.__aenter__ = AsyncMock(return_value=s)
    s.__aexit__ = AsyncMock(return_value=False)
    size_res = MagicMock()
    size_res.scalar.return_value = db_bytes
    rows_res = MagicMock()
    rows_res.mappings.return_value.all.return_value = rows
    s.execute = AsyncMock(side_effect=[size_res, rows_res])
    return s


@pytest.mark.asyncio
async def test_table_health_shapes_rows_and_db_size():
    from datetime import datetime, timezone

    from scribe.services.db_maintenance import get_table_health
    vac = datetime(2026, 6, 14, 4, 0, tzinfo=timezone.utc)
    rows = [
        {"table_name": "notes", "live": 1000, "dead": 300, "dead_pct": 23.1,
         "total_bytes": 5_000_000, "mod_since_analyze": 50,
         "last_vacuum": vac, "last_analyze": None},
    ]
    with patch("scribe.services.db_maintenance.async_session",
               return_value=_health_session(42_000_000, rows)):
        health = await get_table_health()

    assert health["db_bytes"] == 42_000_000
    t = health["tables"][0]
    assert t["table"] == "notes"
    assert t["dead_pct"] == 23.1
    assert t["last_vacuum"] == vac.isoformat()
    assert t["last_analyze"] is None  # null timestamp passes through as None


@pytest.mark.asyncio
async def test_summary_is_persisted_as_admin_setting():
    engine, _ = _mock_engine()
    setter = AsyncMock()
    with patch("scribe.services.db_maintenance.engine", engine), \
         patch("scribe.services.db_maintenance.set_admin_setting", setter):
        await run_maintenance()
    setter.assert_awaited_once()
    assert setter.await_args.args[0] == "db_maintenance_last_run"