learningcircuit--local-deep-research/tests/database/test_journal_migrations_encrypted.py

"""Journal-quality migration regression test against a SQLCipher-encrypted DB.

The existing `test_encrypted_database_orm.py` exercises ORM operations
but never explicitly walks the new journal-quality chain. This test
creates a fresh user DB via :class:`DatabaseManager`, runs migrations
to head, inserts a Journal row carrying every kept column, closes the
engine, reopens it with the same key, and reads the row back.

Why this matters: SQLCipher-keyed engines route every statement through
the sqlcipher_utils key-first ordering, and batch_alter_table rebuilds
the journals table. A key-management regression would show up here.
"""

from __future__ import annotations

import sys
import tempfile
from pathlib import Path

import pytest

pytest.importorskip(
    "sqlcipher3",
    reason="SQLCipher is not available on this platform; the encrypted "
    "migration test requires it to exercise the encrypted engine path.",
)

sys.path.insert(
    0,
    str(Path(__file__).parent.parent.parent.resolve()),
)

from local_deep_research.database.encrypted_db import DatabaseManager
from local_deep_research.database.models import Journal


@pytest.fixture
def temp_data_dir(monkeypatch):
    with tempfile.TemporaryDirectory() as tmpdir:
        path = Path(tmpdir)
        monkeypatch.setattr(
            "local_deep_research.database.encrypted_db.get_data_directory",
            lambda: path,
        )
        yield path


@pytest.fixture
def db_manager(temp_data_dir):
    m = DatabaseManager()
    yield m
    for username in list(m.connections.keys()):
        m.close_user_database(username)


def test_journal_roundtrip_through_encrypted_migrations(db_manager):
    """Create → migrate → write → reopen → read on a keyed DB."""
    assert db_manager.has_encryption, (
        "sqlcipher3 imports but DatabaseManager reported has_encryption=False; "
        "check for LDR_BOOTSTRAP_ALLOW_UNENCRYPTED or a broken SQLCipher install."
    )
    username = "journalman"
    password = "StrongPassword1!"

    # Fresh encrypted DB — create_user_database runs migrations to head
    # via initialize_database/run_migrations.
    db_manager.create_user_database(username, password)

    with db_manager.get_session(username) as session:
        row = Journal(
            name="Journal Of Encrypted Test Cases",
            name_lower="journal of encrypted test cases",
            quality=9,
            score_source="llm",
            quality_model="gpt-test-2026",
            quality_analysis_time=1_700_000_000,
        )
        session.add(row)
        session.commit()
        row_id = row.id

    # Close and reopen — new engine, same key — and verify persistence.
    db_manager.close_user_database(username)
    db_manager.open_user_database(username, password)
    with db_manager.get_session(username) as session:
        persisted = session.query(Journal).filter_by(id=row_id).one()
        assert persisted.name == "Journal Of Encrypted Test Cases"
        assert persisted.name_lower == "journal of encrypted test cases"
        assert persisted.quality == 9
        assert persisted.score_source == "llm"
        assert persisted.quality_model == "gpt-test-2026"


def test_journal_column_set_after_encrypted_migration(db_manager):
    """Post-migration schema must match the 7-column final shape."""
    assert db_manager.has_encryption, (
        "sqlcipher3 imports but DatabaseManager reported has_encryption=False; "
        "check for LDR_BOOTSTRAP_ALLOW_UNENCRYPTED or a broken SQLCipher install."
    )
    from sqlalchemy import inspect

    username = "shapetester"
    password = "StrongPassword1!"
    db_manager.create_user_database(username, password)

    engine = db_manager.connections[username]
    cols = {c["name"] for c in inspect(engine).get_columns("journals")}
    assert cols == {
        "id",
        "name",
        "name_lower",
        "quality",
        "score_source",
        "quality_model",
        "quality_analysis_time",
    }, cols