learningcircuit--local-deep-research/tests/journal_quality/test_db.py

"""
Tests for journal_reference_db.py — the read-only SQLite accessor.

Tests use the actual bundled journal_quality.db file for integration
testing against real data.
"""

import pytest

from local_deep_research.journal_quality.db import (
    get_journal_reference_db,
)

# The ref_db fixture (skip-if-missing pattern) lives in conftest.py and
# is shared with test_db_accessors.py.


# ---------------------------------------------------------------------------
# lookup_source
# ---------------------------------------------------------------------------


class TestLookupSource:
    """Source lookup by ID, ISSN, and name."""

    def test_lookup_by_name(self, ref_db):
        result = ref_db.lookup_source(name="Nature")
        assert result is not None
        assert result["name"] == "Nature"
        assert result["h_index"] > 1000

    def test_lookup_by_name_case_insensitive(self, ref_db):
        result = ref_db.lookup_source(name="NATURE")
        assert result is not None

    def test_lookup_by_name_the_prefix_added(self, ref_db):
        """'Astrophysical Journal Letters' should find 'The Astrophysical Journal Letters'."""
        result = ref_db.lookup_source(name="Astrophysical Journal Letters")
        if result:
            assert "astrophysical" in result["name"].lower()

    def test_lookup_by_name_the_prefix_stripped(self, ref_db):
        """'The Lancet' should also find 'Lancet' if stored without 'the'."""
        result = ref_db.lookup_source(name="The Lancet")
        # Either found directly or via prefix stripping
        if result:
            assert "lancet" in result["name"].lower()

    def test_lookup_by_issn(self, ref_db):
        # Nature ISSN-L
        result = ref_db.lookup_source(issn="0028-0836")
        assert result is not None
        assert "nature" in result["name"].lower()

    def test_lookup_by_openalex_id(self, ref_db):
        # Look up a known journal by name first, get its ID
        nature = ref_db.lookup_source(name="Nature")
        if nature and nature.get("openalex_source_id"):
            result = ref_db.lookup_source(
                source_id=nature["openalex_source_id"]
            )
            assert result is not None

    def test_lookup_not_found(self, ref_db):
        result = ref_db.lookup_source(name="ZZZ Nonexistent Journal 12345")
        assert result is None

    def test_lookup_none_args(self, ref_db):
        result = ref_db.lookup_source()
        assert result is None


# ---------------------------------------------------------------------------
# is_predatory
# ---------------------------------------------------------------------------


class TestIsPredatory:
    """Predatory check against the reference DB."""

    def test_legitimate_journal_not_predatory(self, ref_db):
        is_pred, _ = ref_db.is_predatory(journal_name="Nature")
        assert is_pred is False

    def test_no_args_returns_false(self, ref_db):
        is_pred, _ = ref_db.is_predatory()
        assert is_pred is False


# ---------------------------------------------------------------------------
# is_whitelisted
# ---------------------------------------------------------------------------


class TestIsWhitelisted:
    """Whitelist check (DOAJ or high h-index)."""

    def test_high_h_index_is_whitelisted(self, ref_db):
        # Nature has h_index > 10
        assert ref_db.is_whitelisted(name="Nature") is True

    def test_unknown_not_whitelisted(self, ref_db):
        assert ref_db.is_whitelisted(name="ZZZ Unknown 12345") is False


# ---------------------------------------------------------------------------
# Dashboard queries
# ---------------------------------------------------------------------------


class TestGetSummary:
    """Summary statistics."""

    def test_returns_dict_with_expected_keys(self, ref_db):
        summary = ref_db.get_summary()
        assert "total" in summary
        assert summary["total"] > 200000
        assert "avg_quality" in summary
        assert "predatory_count" in summary
        assert "doaj_count" in summary

    def test_avg_quality_in_range(self, ref_db):
        summary = ref_db.get_summary()
        assert 1 <= summary["avg_quality"] <= 10


class TestGetQualityDistribution:
    """Quality histogram."""

    def test_returns_dict(self, ref_db):
        dist = ref_db.get_quality_distribution()
        assert isinstance(dist, dict)
        assert len(dist) > 0

    def test_values_are_positive(self, ref_db):
        dist = ref_db.get_quality_distribution()
        for v in dist.values():
            assert v > 0


class TestGetSourceDistribution:
    """Score source breakdown."""

    def test_returns_dict(self, ref_db):
        dist = ref_db.get_source_distribution()
        assert isinstance(dist, dict)
        assert "openalex" in dist


class TestGetJournalsPage:
    """Paginated journal list."""

    def test_first_page(self, ref_db):
        journals, total = ref_db.get_journals_page(page=1, per_page=10)
        assert len(journals) == 10
        assert total > 200000

    def test_search_filter(self, ref_db):
        journals, total = ref_db.get_journals_page(search="nature", per_page=10)
        assert total > 0
        assert all("nature" in j["name"].lower() for j in journals)

    def test_tier_filter(self, ref_db):
        journals, total = ref_db.get_journals_page(tier="elite", per_page=10)
        assert all(j["quality"] >= 9 for j in journals)

    def test_sort_by_h_index(self, ref_db):
        journals, _ = ref_db.get_journals_page(
            sort="h_index", order="desc", per_page=5
        )
        h_values = [j["h_index"] for j in journals if j["h_index"] is not None]
        assert h_values == sorted(h_values, reverse=True)

    def test_invalid_sort_column_defaults_to_quality(self, ref_db):
        # Should not crash
        journals, _ = ref_db.get_journals_page(
            sort="'; DROP TABLE sources; --", per_page=5
        )
        assert len(journals) == 5

    def test_pagination_offset(self, ref_db):
        page1, _ = ref_db.get_journals_page(page=1, per_page=5)
        page2, _ = ref_db.get_journals_page(page=2, per_page=5)
        # Different pages should have different journals
        names1 = {j["name"] for j in page1}
        names2 = {j["name"] for j in page2}
        assert names1 != names2


class TestGetInstitutionsPage:
    """Tests for JournalQualityDB.get_institutions_page."""

    def test_invalid_order_defaults_to_desc(self, ref_db):
        """A tainted ``order`` string must not crash or reach SQL.

        Regression guard for the allowlist added in
        fix(db): validate order param in get_institutions_page. The DB
        layer treats anything other than "asc" / "desc" as "desc", so
        the two calls below must return identical institution lists.
        """
        bad, _ = ref_db.get_institutions_page(
            order="'; DROP TABLE institutions; --",
            per_page=5,
        )
        desc, _ = ref_db.get_institutions_page(
            order="desc",
            per_page=5,
        )
        assert [i["openalex_id"] for i in bad] == [
            i["openalex_id"] for i in desc
        ]


# ---------------------------------------------------------------------------
# Build function
# ---------------------------------------------------------------------------


class TestBuildReferenceDb:
    """Test the build_reference_db function."""

    def test_build_creates_db(self, tmp_path):
        """Build a DB into a temp directory."""
        from local_deep_research.config.paths import (
            get_journal_data_directory,
        )
        from local_deep_research.journal_quality.db import build_db

        data_dir = get_journal_data_directory()
        if not (data_dir / "openalex_sources.json.gz").exists():
            pytest.skip("OpenAlex data file not found")

        output = tmp_path / "test_quality.db"
        build_db(data_dir=data_dir, output_path=output)
        assert output.exists()
        assert output.stat().st_size > 1_000_000  # at least 1MB

    def test_built_db_is_queryable(self, tmp_path):
        """Built DB should be queryable via sqlite3."""
        import sqlite3

        from local_deep_research.config.paths import (
            get_journal_data_directory,
        )
        from local_deep_research.journal_quality.db import build_db

        data_dir = get_journal_data_directory()
        if not (data_dir / "openalex_sources.json.gz").exists():
            pytest.skip("OpenAlex data file not found")

        output = tmp_path / "test_quality.db"
        build_db(data_dir=data_dir, output_path=output)

        # Open read-only via URI so chmod 0o444 doesn't matter
        conn = sqlite3.connect(f"file:{output}?mode=ro", uri=True)
        count = conn.execute("SELECT COUNT(*) FROM sources").fetchone()[0]
        assert count > 200000
        conn.close()

    def test_built_db_indexes_score_source(self):
        """The dashboard's /api/journals endpoint filters by
        score_source via equality — the column needs an index or
        each filtered page does a full scan of the ~217K-row table.
        Asserted against an in-memory build so the test runs without
        the 350 MB OpenAlex snapshot.
        """
        from sqlalchemy import create_engine, inspect
        from local_deep_research.journal_quality.models import (
            JournalQualityBase,
        )

        engine = create_engine("sqlite:///:memory:")
        try:
            JournalQualityBase.metadata.create_all(engine)
            indexed_cols = {
                col
                for idx in inspect(engine).get_indexes("sources")
                for col in idx["column_names"]
            }
            assert "score_source" in indexed_cols, (
                "sources.score_source must be indexed — it's a dashboard "
                "filter predicate; without the index the query does a "
                "full-table scan of ~217K rows."
            )
        finally:
            engine.dispose()

    def test_score_source_check_constraint_rejects_invalid(self):
        """CHECK constraint on ``sources.score_source`` must reject
        values outside the allowlist — defense-in-depth against any
        future writer (refactor, migration, import script) that
        accidentally bypasses the API layer's validation and tries
        to insert garbage.
        """
        from sqlalchemy import create_engine, text
        from sqlalchemy.exc import IntegrityError
        from local_deep_research.journal_quality.models import (
            JournalQualityBase,
        )

        engine = create_engine("sqlite:///:memory:")
        # SQLite only enforces CHECK constraints when foreign-key/CHECK
        # handling is explicitly on for the connection. create_engine's
        # default SQLite connection does enforce CHECK — verify via a
        # round-trip.
        JournalQualityBase.metadata.create_all(engine)

        try:
            # Minimal columns needed for a valid row — the NOT NULL flags
            # come from the model, not this test. is_in_doaj /
            # is_predatory both default to False but the insert still has
            # to satisfy NOT NULL.
            cols = "name, name_lower, is_in_doaj, is_predatory, score_source"

            with engine.begin() as conn:
                # Happy path: known-valid value goes through.
                conn.execute(
                    text(
                        f"INSERT INTO sources ({cols}) "
                        "VALUES ('Test Journal', 'test journal', "
                        "0, 0, 'openalex')"
                    )
                )

            with engine.begin() as conn:
                with pytest.raises(
                    IntegrityError, match=r"(?i)CHECK|score_source"
                ):
                    conn.execute(
                        text(
                            f"INSERT INTO sources ({cols}) "
                            "VALUES ('Bad Journal', 'bad journal', "
                            "0, 0, 'garbage')"
                        )
                    )
        finally:
            engine.dispose()

    def test_built_db_stamps_schema_version(self, tmp_path):
        """``build_db`` must stamp ``PRAGMA user_version`` to the
        current ``JOURNAL_QUALITY_SCHEMA_VERSION`` constant. The
        startup-time ``_validate_existing_db`` check reads this
        pragma to detect schema drift and force a rebuild — if the
        stamp goes missing silently, stale DBs survive forever past
        a schema change and users see mysterious query failures.

        Reads the constant dynamically rather than hardcoding a
        number so this test stays green across future version
        bumps without needing a mechanical update.
        """
        import sqlite3

        from local_deep_research.config.paths import (
            get_journal_data_directory,
        )
        from local_deep_research.journal_quality.db import (
            JOURNAL_QUALITY_SCHEMA_VERSION,
            build_db,
        )

        data_dir = get_journal_data_directory()
        if not (data_dir / "openalex_sources.json.gz").exists():
            pytest.skip("OpenAlex data file not found")

        output = tmp_path / "test_quality.db"
        build_db(data_dir=data_dir, output_path=output)

        conn = sqlite3.connect(f"file:{output}?mode=ro", uri=True)
        try:
            stamped = conn.execute("PRAGMA user_version").fetchone()[0]
        finally:
            conn.close()

        assert stamped == JOURNAL_QUALITY_SCHEMA_VERSION, (
            f"PRAGMA user_version={stamped}, expected "
            f"{JOURNAL_QUALITY_SCHEMA_VERSION}. If this fails, "
            "_validate_existing_db won't detect schema drift and "
            "stale DBs will never rebuild after the next bump."
        )


# ---------------------------------------------------------------------------
# _populate_sources unit tests (hijacked + DOAJ-only second pass)
# ---------------------------------------------------------------------------


class TestPopulateSources:
    """Cover the two correctness fixes in _populate_sources directly,
    without requiring the 50 MB on-disk DB."""

    def _build_in_memory(self, sources, doaj, pred):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.journal_quality.db import _populate_sources
        from local_deep_research.journal_quality.models import (
            JournalQualityBase,
        )

        engine = create_engine("sqlite:///:memory:")
        JournalQualityBase.metadata.create_all(engine)
        Session = sessionmaker(bind=engine)
        with Session() as s:
            _populate_sources(s, sources, doaj, pred)
            s.commit()
        return engine

    def test_hijacked_journal_flagged(self):
        from sqlalchemy import select
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.journal_quality.models import Source

        sources = {"S1": {"n": "Fake Cloned Journal", "i": "1234-5678", "h": 5}}
        pred = {
            "journals": set(),
            "publishers": set(),
            "hijacked": {"fake cloned journal"},
            "long_pubs": [],
        }
        engine = self._build_in_memory(sources, {}, pred)
        try:
            with sessionmaker(bind=engine)() as s:
                row = s.scalars(
                    select(Source).where(
                        Source.name_lower == "fake cloned journal"
                    )
                ).first()
                assert row is not None
                assert row.is_predatory is True
                assert row.predatory_source == "stop-predatory-hijacked"
        finally:
            engine.dispose()

    def test_doaj_only_journal_inserted(self):
        from sqlalchemy import select
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.journal_quality.models import Source

        doaj = {
            "9999-0001": {
                "name": "Some Small OA Journal",
                "publisher": "Small Press",
            }
        }
        pred = {
            "journals": set(),
            "publishers": set(),
            "hijacked": set(),
            "long_pubs": [],
        }
        engine = self._build_in_memory({}, doaj, pred)
        try:
            with sessionmaker(bind=engine)() as s:
                row = s.scalars(
                    select(Source).where(
                        Source.name_lower == "some small oa journal"
                    )
                ).first()
                assert row is not None
                assert row.score_source == "doaj"
                assert row.is_in_doaj is True
                assert row.openalex_source_id is None
        finally:
            engine.dispose()

    def test_doaj_crossref_flags_existing_openalex_source(self):
        """First-pass DOAJ cross-reference, both directions.

        Against a single non-empty DOAJ dump, an OpenAlex source whose
        normalized ISSN IS in the dump must be flagged ``is_in_doaj=True``,
        and one whose ISSN is NOT must stay ``is_in_doaj=False`` — both keeping
        ``score_source='openalex'`` (they came from OpenAlex; DOAJ only adds
        the flag). Testing both sides against the same dump pins the actual
        per-ISSN match: a regression that flagged every source whenever the
        dump is non-empty (``is_in_doaj = bool(doaj_data)``) passes a
        positive-only test but fails the negative assertion here. Partner to
        ``test_doaj_only_journal_inserted`` (second pass, DOAJ-only venues).
        Guards the ``doaj_data.get(issn)`` cross-ref at
        ``db.py::_populate_sources``.
        """
        from sqlalchemy import select
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.journal_quality.models import Source
        from local_deep_research.utilities.citation_normalizer import (
            normalize_issn,
        )

        # S1's ISSN is hyphenated while the DOAJ dump is keyed by the
        # normalized (no-dash) form, so the match exercises normalize_issn on
        # both sides. S2 is a real OpenAlex source whose ISSN is absent from
        # the dump — the negative control.
        sources = {
            "S1": {"n": "PLoS ONE", "i": "1932-6203", "h": 200, "p": "PLOS"},
            "S2": {"n": "Obscure Closed Journal", "i": "0000-0019", "h": 40},
        }
        doaj = {
            normalize_issn("1932-6203"): {
                "name": "PLoS ONE",
                "publisher": "PLOS",
            }
        }
        pred = {
            "journals": set(),
            "publishers": set(),
            "hijacked": set(),
            "long_pubs": [],
        }
        engine = self._build_in_memory(sources, doaj, pred)
        try:
            with sessionmaker(bind=engine)() as s:
                hit = s.scalars(
                    select(Source).where(
                        Source.issn == normalize_issn("1932-6203")
                    )
                ).first()
                assert hit is not None
                assert hit.is_in_doaj is True
                # Came from OpenAlex, so the score source stays "openalex";
                # DOAJ only contributes the is_in_doaj flag on this pass.
                assert hit.score_source == "openalex"
                assert hit.openalex_source_id == "S1"

                # Negative control: in OpenAlex, not in the (non-empty) dump.
                miss = s.scalars(
                    select(Source).where(
                        Source.issn == normalize_issn("0000-0019")
                    )
                ).first()
                assert miss is not None
                assert miss.is_in_doaj is False
                assert miss.score_source == "openalex"
                assert miss.openalex_source_id == "S2"
        finally:
            engine.dispose()

    def test_quartile_derivation_global_per_type(self):
        """Sources are bucketed Q1–Q4 by cited_by_count percentile within
        each source_type. Display-only signal — quality stays h-index-driven.
        """
        from sqlalchemy import select
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.journal_quality.models import Source

        # 8 journals with cited_by_count 100..800. With 8 records the
        # percentile splits land at: ranks 6,7 → Q1, ranks 4,5 → Q2,
        # ranks 2,3 → Q3, ranks 0,1 → Q4. Plus a 9th journal with NULL
        # cited_by_count to confirm it gets NULL quartile.
        sources = {
            f"S{i}": {
                "n": f"Journal {i}",
                "i": f"0000-{i:04d}",
                "h": 5,
                "cb": (i + 1) * 100,
            }
            for i in range(8)
        }
        sources["S99"] = {"n": "Citationless Journal", "i": "9999-9999", "h": 5}

        pred = {
            "journals": set(),
            "publishers": set(),
            "hijacked": set(),
            "long_pubs": [],
        }
        engine = self._build_in_memory(sources, {}, pred)

        try:
            with sessionmaker(bind=engine)() as s:
                rows = s.scalars(
                    select(Source).order_by(Source.cited_by_count)
                ).all()

                quartiled = [r for r in rows if r.cited_by_count is not None]
                assert len(quartiled) == 8

                # Highest two should be Q1, lowest two should be Q4.
                assert quartiled[-1].quartile == "Q1"
                assert quartiled[-2].quartile == "Q1"
                assert quartiled[0].quartile == "Q4"
                assert quartiled[1].quartile == "Q4"

                # Every quartile bucket is represented.
                assert {r.quartile for r in quartiled} == {
                    "Q1",
                    "Q2",
                    "Q3",
                    "Q4",
                }

                # NULL cited_by_count → NULL quartile.
                citationless = s.scalars(
                    select(Source).where(Source.name == "Citationless Journal")
                ).first()
                assert citationless is not None
                assert citationless.quartile is None

                # Quartile now feeds into quality (fixed in Round 5 review):
                # Q1 → 8 (STRONG), Q2 → 7, Q3 → 6, Q4 → 5. h_index=5 is below
                # the quartile-bump threshold so Q1 does not promote to 10.
                quartile_to_quality = {"Q1": 8, "Q2": 7, "Q3": 6, "Q4": 5}
                for r in quartiled:
                    assert r.quality == quartile_to_quality[r.quartile], (
                        f"Journal with quartile={r.quartile} should score "
                        f"{quartile_to_quality[r.quartile]} not {r.quality}"
                    )
        finally:
            engine.dispose()

    def test_quartile_separates_journals_from_conferences(self):
        """Journal pool and conference pool are percentile-binned
        independently — a conference shouldn't push a journal out of Q1.
        """
        from sqlalchemy import select
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.journal_quality.models import Source

        sources = {
            # Single journal — alone in its pool, lands at the top → Q1
            "J1": {
                "n": "Solo Journal",
                "t": "j",
                "i": "0001-0001",
                "h": 5,
                "cb": 10,
            },
            # Single conference — alone in its pool, also Q1
            "C1": {
                "n": "Solo Conference",
                "t": "c",
                "i": "0002-0002",
                "h": 5,
                "cb": 10000,
            },
        }
        pred = {
            "journals": set(),
            "publishers": set(),
            "hijacked": set(),
            "long_pubs": [],
        }
        engine = self._build_in_memory(sources, {}, pred)
        try:
            with sessionmaker(bind=engine)() as s:
                rows = s.scalars(select(Source)).all()
                by_name = {r.name: r for r in rows}
                # Pool size 1 → percentile 0/1=0.0 → Q4 (lowest bucket).
                # The point of the test is that they're processed in
                # separate pools, not merged: the conference's huge cb does
                # NOT affect the journal's quartile.
                assert by_name["Solo Journal"].quartile is not None
                assert by_name["Solo Conference"].quartile is not None
        finally:
            engine.dispose()

    def test_print_and_electronic_issn_both_survive(self):
        from sqlalchemy import select
        from sqlalchemy.orm import sessionmaker

        from local_deep_research.journal_quality.models import Source

        sources = {
            "S1": {"n": "Journal of X", "i": "1111-1111", "h": 50},
            "S2": {"n": "Journal of X", "i": "2222-2222", "h": 50},
        }
        pred = {
            "journals": set(),
            "publishers": set(),
            "hijacked": set(),
            "long_pubs": [],
        }
        engine = self._build_in_memory(sources, {}, pred)
        try:
            with sessionmaker(bind=engine)() as s:
                rows = s.scalars(
                    select(Source).where(Source.name_lower == "journal of x")
                ).all()
                # ISSNs are stored in canonical 8-char no-dash form (normalize_issn)
                issns = {r.issn for r in rows}
                assert issns == {"11111111", "22222222"}
        finally:
            engine.dispose()


# ---------------------------------------------------------------------------
# Singleton
# ---------------------------------------------------------------------------


class TestSingleton:
    """Singleton pattern."""

    def test_returns_same_instance(self):
        import local_deep_research.journal_quality.db as mod

        mod._db = None
        a = get_journal_reference_db()
        b = get_journal_reference_db()
        assert a is b
        mod._db = None


# ---------------------------------------------------------------------------
# Read-only enforcement
# ---------------------------------------------------------------------------


class TestLookupSourcesBatch:
    """Batch name lookup used by the user-research dashboard endpoint."""

    def test_empty_names_returns_empty_dict(self, ref_db):
        assert ref_db.lookup_sources_batch([]) == {}

    def test_none_and_blank_names_filtered(self, ref_db):
        # None and "" entries are dropped before the SQL query — the
        # result for an input that has nothing real is an empty dict,
        # not a SQL error.
        assert ref_db.lookup_sources_batch([None, "", "   "]) == {}

    def test_returns_matches_keyed_by_name_lower(self, ref_db):
        result = ref_db.lookup_sources_batch(["Nature", "Science"])
        # Both are near-certainly in the ref DB; the keys are the
        # normalized name_lower values, not the input casing.
        assert "nature" in result
        assert "science" in result
        assert result["nature"]["h_index"] > 1000
        assert result["nature"]["quality"] is not None

    def test_unknown_name_simply_absent(self, ref_db):
        result = ref_db.lookup_sources_batch(
            ["Nature", "ZZZ-definitely-not-a-real-journal-XYZ"]
        )
        assert "nature" in result
        assert "zzz-definitely-not-a-real-journal-xyz" not in result

    def test_deduplicates_input(self, ref_db):
        # Repeated names should not cause duplicate key errors or
        # multiply SQL parameter counts — the input is de-duplicated.
        result = ref_db.lookup_sources_batch(["Nature"] * 5 + ["Science"])
        assert set(result.keys()) >= {"nature", "science"}


class TestStaleDataVersionWarning:
    """_ensure_engine must log a WARNING when the on-disk version.json
    is behind the bundled JOURNAL_DATA_VERSION so admins see the
    mismatch in server logs without having to visit the dashboard.
    """

    @pytest.fixture
    def loguru_sink(self):
        """Capture loguru log records — caplog doesn't see loguru output.

        The package's __init__ calls ``logger.disable("local_deep_research")``
        to avoid interfering with downstream users' log setup, so the
        sink alone isn't enough — we also re-enable the package for the
        duration of the test.
        """
        from loguru import logger as loguru_logger

        records = []

        def _sink(message):
            records.append(message.record)

        # diagnose=False keeps the captured sink consistent with the
        # production policy (#4185 / #4384) — exceptions logged through this
        # fixture would otherwise carry frame-local repr() into recorded
        # messages and pytest output.
        handler_id = loguru_logger.add(_sink, level="WARNING", diagnose=False)
        loguru_logger.enable("local_deep_research")
        yield records
        loguru_logger.disable("local_deep_research")
        loguru_logger.remove(handler_id)

    def test_warn_on_stale_version(self, tmp_path, monkeypatch, loguru_sink):
        """A mismatched version.json triggers one WARNING per engine.

        The scenario we're guarding against: an older build is on disk,
        code has been upgraded to JOURNAL_DATA_VERSION="v4", but
        nothing has invalidated the cached sources. Without this
        warning, the filter silently serves stale scores until an
        admin manually visits /metrics/journals.
        """
        import json
        from local_deep_research.journal_quality import db as db_mod
        from local_deep_research.journal_quality import downloader

        instance = db_mod.JournalQualityDB()
        version_file = tmp_path / "version.json"
        version_file.write_text(json.dumps({"version": "v3"}))
        monkeypatch.setattr(downloader, "JOURNAL_DATA_VERSION", "v4")

        instance._warn_on_stale_data_version(tmp_path)

        assert any(
            "stale" in r["message"] and "v3" in r["message"]
            for r in loguru_sink
        )
        assert instance._stale_version_warned is True

        # Second call must not re-emit — one warning per engine lifetime.
        loguru_sink.clear()
        instance._warn_on_stale_data_version(tmp_path)
        assert not loguru_sink

    def test_no_warn_on_matching_version(
        self, tmp_path, monkeypatch, loguru_sink
    ):
        """When version.json matches JOURNAL_DATA_VERSION, stay silent."""
        import json
        from local_deep_research.journal_quality import db as db_mod
        from local_deep_research.journal_quality import downloader

        instance = db_mod.JournalQualityDB()
        version_file = tmp_path / "version.json"
        version_file.write_text(json.dumps({"version": "v4"}))
        monkeypatch.setattr(downloader, "JOURNAL_DATA_VERSION", "v4")

        instance._warn_on_stale_data_version(tmp_path)

        assert not loguru_sink
        assert instance._stale_version_warned is False

    def test_no_warn_on_missing_version_file(self, tmp_path, loguru_sink):
        """Fresh install (no version.json yet) → silent; the dashboard
        banner handles first-run messaging, no server-log spam."""
        from local_deep_research.journal_quality import db as db_mod

        instance = db_mod.JournalQualityDB()
        instance._warn_on_stale_data_version(tmp_path)
        assert not loguru_sink


class TestReadOnlyEnforcement:
    """The runtime accessor must physically refuse writes."""

    def test_write_attempt_raises_operational_error(self, ref_db):
        """sqlalchemy.exc.OperationalError on any write attempt.

        This is the safety net for the SQLite URI mode=ro flag — if
        someone removes it from `_ensure_engine`, this test catches it.
        """
        from sqlalchemy.exc import OperationalError

        from local_deep_research.journal_quality.models import Source

        with ref_db.session() as s:
            s.add(
                Source(
                    name="hack",
                    name_lower="hack",
                    score_source="test",
                )
            )
            with pytest.raises(OperationalError, match="readonly"):
                s.commit()
            s.rollback()


# ---------------------------------------------------------------------------
# Cleanup-path logging
# ---------------------------------------------------------------------------


class TestUnlinkUnusableDbLogs:
    """``_unlink_unusable_db`` does best-effort corruption-recovery
    cleanup. Previously swallowed OSError silently; now logs so a
    chmod/unlink failure surfaces in the ops log instead of quietly
    masking the real underlying problem (read-only mount, Windows
    file-in-use, permissions).
    """

    def test_chmod_failure_logs_warning(self, tmp_path, loguru_caplog):
        from local_deep_research.journal_quality.db import JournalQualityDB

        dead_path = tmp_path / "nonexistent.db"  # chmod will raise FileNotFound
        # 30 == WARNING in the stdlib level numbering that loguru_caplog
        # maps from (see tests/conftest.py::loguru_caplog fixture).
        with loguru_caplog.at_level(30):
            JournalQualityDB._unlink_unusable_db(dead_path)

        # Both chmod and unlink fail on a missing file — we expect a
        # warning mentioning the path, not silent success.
        assert any(
            "chmod" in rec.message.lower() and str(dead_path) in rec.message
            for rec in loguru_caplog.records
        ), "chmod failure on cleanup must be logged, not silenced"
        assert any(
            "unlink" in rec.message.lower() and str(dead_path) in rec.message
            for rec in loguru_caplog.records
        ), "unlink failure on cleanup must be logged, not silenced"

    def test_success_path_is_silent(self, tmp_path, loguru_caplog):
        """Happy path: both chmod and unlink succeed → nothing logged
        above WARNING level. Otherwise we'd pollute logs on every
        schema-drift rebuild.
        """
        from local_deep_research.journal_quality.db import JournalQualityDB

        path = tmp_path / "doomed.db"
        path.write_text("x")
        path.chmod(0o444)  # simulate the real post-build chmod
        with loguru_caplog.at_level(30):  # WARNING
            JournalQualityDB._unlink_unusable_db(path)
        assert not path.exists()
        # No WARNING records for chmod / unlink in success case.
        assert not any(
            "unlink" in rec.message.lower() or "chmod" in rec.message.lower()
            for rec in loguru_caplog.records
            if rec.levelname == "WARNING"
        )