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

"""
Tests for journal_quality/db.py accessor methods and utility functions
that lack direct coverage in test_db.py.

Uses the real bundled journal_quality.db where available (via the shared
``ref_db`` fixture in conftest.py, same skip-if-missing pattern as
test_db.py) and temp-file fixtures for filesystem utilities.
"""

import os
import sqlite3
import time

import pytest

from local_deep_research.journal_quality.db import (
    JOURNAL_QUALITY_SCHEMA_VERSION,
    JournalQualityDB,
    _sweep_stale_tmp_files,
)

# PLoS ONE — open access, in DOAJ since the registry's early days, so a
# stable positive example. Nature is a subscription journal and therefore
# absent from DOAJ — a stable negative example that exists in the sources
# table (unlike a made-up ISSN, it exercises the is_in_doaj filter).
DOAJ_ISSN = "1932-6203"  # PLoS ONE
NON_DOAJ_ISSN = "0028-0836"  # Nature

# _sweep_stale_tmp_files removes tmp files older than 1 hour; backdate
# twice that so the test is comfortably past the cutoff.
SWEEP_CUTOFF_SECONDS = 3600
STALE_MTIME_AGE_SECONDS = 2 * SWEEP_CUTOFF_SECONDS


def _uninitialized_db() -> JournalQualityDB:
    """Instance that skips ``__init__`` (no engine, no lock).

    Valid for ``_validate_existing_db`` and the early-return guard of
    ``expand_abbreviation`` because neither reads instance state before
    the code under test runs. If those methods ever grow instance-state
    dependencies, switch this to a properly constructed instance.
    """
    return JournalQualityDB.__new__(JournalQualityDB)


def _openalex_id_for_ror(ref_db, ror_id):
    """Fetch an institution's OpenAlex ID straight from the SQLite file.

    The public ``lookup_institution()`` dict deliberately omits
    ``openalex_id`` (see ``_institution_to_dict``), so tests that
    exercise the openalex_id lookup path have to read it from the
    table directly.
    """
    conn = sqlite3.connect(
        f"file:{ref_db._resolve_db_path()}?mode=ro", uri=True
    )
    try:
        row = conn.execute(
            "SELECT openalex_id FROM institutions WHERE ror_id = ?",
            (ror_id,),
        ).fetchone()
    finally:
        conn.close()
    return row[0] if row else None


# ---------------------------------------------------------------------------
# lookup_doaj / is_in_doaj
# ---------------------------------------------------------------------------


@pytest.fixture()
def doaj_db():
    """A ``JournalQualityDB`` backed by an in-memory DB with controlled rows.

    The DOAJ accessor tests must not use the shared ``ref_db`` (real bundled
    file): that fixture *skips* in CI where the file is absent, and worse it
    makes these tests *fail* (not skip) on a machine whose local
    ``journal_quality.db`` was built without ``doaj_journals.json`` — in that
    case ``is_in_doaj`` is 0 for every row, so the positive assertions can
    never hold. Wiring a small fixed dataset makes ``lookup_doaj`` /
    ``is_in_doaj`` deterministic and lets them actually run in CI.
    """
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

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

    # StaticPool + a single shared connection keeps the in-memory DB alive for
    # the whole fixture (a plain sqlite:// would get a fresh empty DB per
    # connection).
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    JournalQualityBase.metadata.create_all(engine)
    session_local = sessionmaker(bind=engine, expire_on_commit=False)
    with session_local() as s:
        s.add(
            Source(
                name="PLoS ONE",
                name_lower="plos one",
                issn=normalize_issn(DOAJ_ISSN),
                publisher="PLOS",
                is_in_doaj=True,
                score_source="doaj",
            )
        )
        # Nature: present in sources but NOT in DOAJ — the negative example
        # that exercises the is_in_doaj filter rather than a missing row.
        s.add(
            Source(
                name="Nature",
                name_lower="nature",
                issn=normalize_issn(NON_DOAJ_ISSN),
                publisher="Springer Nature",
                is_in_doaj=False,
                score_source="openalex",
            )
        )
        s.commit()

    db = JournalQualityDB()
    # Pre-wire the engine/session so _ensure_engine() short-circuits and the
    # accessor logic runs against our dataset, never the real DB path.
    db._engine = engine
    db._SessionLocal = session_local
    try:
        yield db
    finally:
        engine.dispose()


class TestLookupDoaj:
    """DOAJ lookup by ISSN (query against is_in_doaj flag on Source rows)."""

    def test_lookup_doaj_journal_returns_dict(self, doaj_db):
        result = doaj_db.lookup_doaj(issn=DOAJ_ISSN)
        assert result is not None
        assert result["name"] == "PLoS ONE"
        assert result["publisher"] == "PLOS"

    def test_lookup_non_doaj_journal_returns_none(self, doaj_db):
        # Nature exists in the sources table but is not in DOAJ, so the
        # DOAJ-only lookup must filter it out.
        assert doaj_db.lookup_doaj(issn=NON_DOAJ_ISSN) is None

    def test_lookup_nonexistent_issn_returns_none(self, doaj_db):
        result = doaj_db.lookup_doaj(issn="0000-0000")
        assert result is None

    def test_lookup_none_issn_returns_none(self, doaj_db):
        result = doaj_db.lookup_doaj(issn=None)
        assert result is None

    def test_lookup_empty_issn_returns_none(self, doaj_db):
        result = doaj_db.lookup_doaj(issn="")
        assert result is None


class TestIsInDoaj:
    def test_doaj_journal_is_true(self, doaj_db):
        assert doaj_db.is_in_doaj(DOAJ_ISSN) is True

    def test_non_doaj_journal_is_false(self, doaj_db):
        assert doaj_db.is_in_doaj(NON_DOAJ_ISSN) is False

    def test_nonexistent_issn(self, doaj_db):
        assert doaj_db.is_in_doaj("0000-0000") is False


# ---------------------------------------------------------------------------
# count_predatory_by_names
# ---------------------------------------------------------------------------


class TestCountPredatoryByNames:
    def test_empty_iterable_returns_zero(self, ref_db):
        assert ref_db.count_predatory_by_names([]) == 0

    def test_none_and_blanks_filtered(self, ref_db):
        assert ref_db.count_predatory_by_names([None, "", "   "]) == 0

    def test_legitimate_journals_count_zero(self, ref_db):
        count = ref_db.count_predatory_by_names(["Nature", "Science"])
        assert count == 0

    def test_returns_int(self, ref_db):
        result = ref_db.count_predatory_by_names(["Nature"])
        assert isinstance(result, int)


# ---------------------------------------------------------------------------
# lookup_institution
# ---------------------------------------------------------------------------


class TestLookupInstitution:
    def test_by_name_returns_dict(self, ref_db):
        result = ref_db.lookup_institution(name="Harvard University")
        assert result is not None
        assert result["name"] == "Harvard University"
        assert result["h_index"] is not None

    def test_nonexistent_returns_none(self, ref_db):
        result = ref_db.lookup_institution(name="ZZZ Nonexistent 99999")
        assert result is None

    def test_no_args_returns_none(self, ref_db):
        result = ref_db.lookup_institution()
        assert result is None

    def test_by_openalex_id(self, ref_db):
        # The accessor dict has no openalex_id key, so resolve it from
        # the table via the ror_id and re-lookup through the public API.
        by_name = ref_db.lookup_institution(name="Harvard University")
        assert by_name is not None
        openalex_id = _openalex_id_for_ror(ref_db, by_name["ror_id"])
        assert openalex_id is not None
        by_id = ref_db.lookup_institution(openalex_id=openalex_id)
        assert by_id is not None
        assert by_id["name"] == "Harvard University"

    def test_by_ror_id(self, ref_db):
        by_name = ref_db.lookup_institution(name="Harvard University")
        assert by_name is not None
        assert by_name["ror_id"]
        by_ror = ref_db.lookup_institution(ror_id=by_name["ror_id"])
        assert by_ror is not None
        assert by_ror["name"] == "Harvard University"


# ---------------------------------------------------------------------------
# score_from_affiliations
# ---------------------------------------------------------------------------


class TestScoreFromAffiliations:
    def test_empty_list_returns_none(self, ref_db):
        assert ref_db.score_from_affiliations([]) is None

    def test_string_affiliation(self, ref_db):
        result = ref_db.score_from_affiliations(["Harvard University"])
        assert isinstance(result, int)

    def test_dict_affiliation_with_name(self, ref_db):
        result = ref_db.score_from_affiliations(
            [{"name": "Harvard University"}]
        )
        assert isinstance(result, int)

    def test_dict_affiliation_with_openalex_id(self, ref_db):
        by_name = ref_db.lookup_institution(name="Harvard University")
        assert by_name is not None
        openalex_id = _openalex_id_for_ror(ref_db, by_name["ror_id"])
        assert openalex_id is not None
        result = ref_db.score_from_affiliations([{"openalex_id": openalex_id}])
        assert isinstance(result, int)

    def test_nonexistent_returns_none(self, ref_db):
        result = ref_db.score_from_affiliations(
            ["ZZZ Nonexistent Institution 99999"]
        )
        assert result is None

    def test_dict_with_no_relevant_keys_returns_none(self, ref_db):
        result = ref_db.score_from_affiliations([{"foo": "bar"}])
        assert result is None


# ---------------------------------------------------------------------------
# expand_abbreviation
# ---------------------------------------------------------------------------


class TestExpandAbbreviation:
    def test_empty_string_returns_none(self, ref_db):
        assert ref_db.expand_abbreviation("") is None

    def test_none_returns_none(self):
        assert _uninitialized_db().expand_abbreviation(None) is None

    def test_nonexistent_returns_none(self, ref_db):
        result = ref_db.expand_abbreviation("ZZZNOTANABBREVIATION12345")
        assert result is None

    def test_known_abbreviation(self, ref_db):
        # JACS comes from the bundled JabRef abbreviation list.
        result = ref_db.expand_abbreviation("JACS")
        assert result is not None
        assert "american chemical society" in result.lower()


# ---------------------------------------------------------------------------
# is_predatory expanded branches
# ---------------------------------------------------------------------------


class TestIsPredatoryExpanded:
    """Expand TestIsPredatory with publisher and hijacked branches."""

    def test_publisher_name_arg(self, ref_db):
        # A legitimate publisher should not be flagged
        is_pred, source = ref_db.is_predatory(publisher_name="Elsevier")
        assert is_pred is False

    def test_both_journal_and_publisher(self, ref_db):
        is_pred, source = ref_db.is_predatory(
            journal_name="Nature", publisher_name="Springer Nature"
        )
        assert is_pred is False

    def test_long_publisher_name_handled(self, ref_db):
        # The implementation checks substrings for long publisher names
        is_pred, source = ref_db.is_predatory(publisher_name="A" * 500)
        assert is_pred is False


# ---------------------------------------------------------------------------
# _validate_existing_db (temp-file tests, no real DB needed)
# ---------------------------------------------------------------------------


class TestValidateExistingDb:
    def test_valid_db_with_matching_schema(self, tmp_path):
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        conn.execute(
            "PRAGMA user_version = %d" % JOURNAL_QUALITY_SCHEMA_VERSION
        )
        conn.execute("CREATE TABLE t (x)")
        conn.commit()
        conn.close()

        assert _uninitialized_db()._validate_existing_db(db_path) is True

    def test_stale_schema_triggers_rebuild(self, tmp_path):
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        conn.execute("PRAGMA user_version = 999")
        conn.execute("CREATE TABLE t (x)")
        conn.commit()
        conn.close()

        result = _uninitialized_db()._validate_existing_db(db_path)
        assert result is False
        # File should have been removed by _unlink_unusable_db
        assert not db_path.exists()

    def test_grandfathered_zero_version(self, tmp_path):
        db_path = tmp_path / "test.db"
        conn = sqlite3.connect(str(db_path))
        conn.execute("CREATE TABLE t (x)")
        conn.commit()
        conn.close()

        assert _uninitialized_db()._validate_existing_db(db_path) is True

    def test_corrupted_file_returns_false(self, tmp_path):
        db_path = tmp_path / "test.db"
        db_path.write_text("this is not a database")

        assert _uninitialized_db()._validate_existing_db(db_path) is False

    def test_missing_file_returns_false(self, tmp_path):
        db_path = tmp_path / "nonexistent.db"

        assert _uninitialized_db()._validate_existing_db(db_path) is False


# ---------------------------------------------------------------------------
# _sweep_stale_tmp_files (temp-file tests)
# ---------------------------------------------------------------------------


class TestSweepStaleTmpFiles:
    def test_removes_old_tmp_files(self, tmp_path):
        stale = tmp_path / "journal_quality.db.tmp-old"
        stale.write_text("stale")
        old_mtime = time.time() - STALE_MTIME_AGE_SECONDS
        os.utime(str(stale), (old_mtime, old_mtime))

        fresh = tmp_path / "journal_quality.db.tmp-recent"
        fresh.write_text("fresh")

        _sweep_stale_tmp_files(tmp_path, "journal_quality.db")

        assert not stale.exists()
        assert fresh.exists()

    def test_nonexistent_directory_is_noop(self, tmp_path):
        missing = tmp_path / "does_not_exist"
        _sweep_stale_tmp_files(missing, "journal_quality.db")
        # No exception raised

    def test_no_matching_files_is_noop(self, tmp_path):
        other = tmp_path / "other.txt"
        other.write_text("unrelated")
        _sweep_stale_tmp_files(tmp_path, "journal_quality.db")
        assert other.exists()


# _escape_like was consolidated into the shared ``sql_utils.escape_like``
# (see tests/utilities/test_sql_utils.py::TestEscapeLike for its coverage).