learningcircuit--local-deep-research/tests/integration/test_journal_quality_release_gate.py

"""Release-gate integration test for the journal-quality data pipeline.

Why this test exists
====================

The journal-quality system depends on five external data sources:

- OpenAlex Sources (S3 bulk dump)     ~280K journals/conferences
- DOAJ public CSV dump                 ~22K  open-access journals
- Stop Predatory Journals              ~2.5K predatory entries
- JabRef abbreviation list             ~66K abbreviations
- OpenAlex Institutions (S3 bulk dump) ~120K institutions

If any of these upstreams change their schema (rename a field, drop a
column, restructure the JSON layout), the bundled-data tier silently
breaks: every academic search result starts coming back unscored. This
test catches that BEFORE we cut a release.

Structure
---------

A session-scoped fixture downloads ALL FIVE sources **in parallel** via
``ThreadPoolExecutor``. Each per-source test then asserts file presence
and record-shape against the already-downloaded data (fast). A separate
test runs ``build_db()`` against the freshly-downloaded files and a
final test verifies the runtime accessor can score a real journal.

Parallelism is essential: the OpenAlex Institutions API alone takes
~10 minutes (550 paginated requests). Sequentially, all five would
exceed 25 minutes; in parallel the wall-clock is bounded by the
slowest single source (~10 min for institutions).

This test is intentionally **not** part of the regular suite — it pulls
~30 MB from third-party APIs and runs ~10–15 minutes wall-clock. It's
marked with ``@pytest.mark.integration`` and ``@pytest.mark.slow`` so
it's skipped by default; the dedicated workflow opts in via ``-m``
selection.

Run locally with::

    pytest tests/integration/test_journal_quality_release_gate.py \\
        -m "integration and slow" --no-header -v --timeout=2700

Or via the dedicated CI workflow::

    .github/workflows/journal-data-integration.yml
"""

from __future__ import annotations

import gzip
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import pytest
import requests.exceptions
from loguru import logger


pytestmark = [
    pytest.mark.integration,
    pytest.mark.slow,
]


# Lower bounds — deliberately loose so a small upstream fluctuation
# doesn't break the gate, but tight enough to catch a catastrophic
# regression (an API returning an empty result set).
MIN_OPENALEX_SOURCES = 100_000  # actual ~217K
MIN_DOAJ_JOURNALS = 5_000  # actual ~35K
MIN_PREDATORY_JOURNALS = 500  # actual ~1.3K
MIN_INSTITUTIONS = 50_000  # actual ~110K
MIN_ABBREVIATIONS = 10_000  # actual ~66K


@pytest.fixture(scope="module")
def downloaded_data_dir(tmp_path_factory) -> Path:
    """Download every external data source in parallel into a tmp dir.

    Uses ``ThreadPoolExecutor`` so the slowest source (institutions)
    sets the wall-clock floor instead of the sum of all five. Each
    source is fetched directly via its ``DataSource.fetch()`` method —
    we deliberately bypass ``download_journal_data()`` here so a single
    source's failure doesn't abort the others. We want the test to
    report ALL broken sources, not just the first one we hit.

    Module-scoped so the per-source tests + the build test + the
    lookup test all share one download.
    """
    from local_deep_research.journal_quality.data_sources import (
        ALL_SOURCES,
    )

    tmp_dir: Path = tmp_path_factory.mktemp("journal_quality_release_gate")

    errors: dict[str, str] = {}
    counts: dict[str, int] = {}

    # Transient network errors that warrant a full-source retry. The
    # per-partition download in ``iter_partitions`` already retries
    # individual partitions with a 2-5-10-20-40 s backoff (5 attempts).
    # If even that budget is exhausted (e.g. a sustained S3 outage >
    # ~75 s), we retry the *entire* source from scratch — fresh TCP
    # connections, fresh manifest, fresh partition list — up to 3 times.
    _TRANSIENT_ERRORS = (
        requests.exceptions.ChunkedEncodingError,
        requests.exceptions.ConnectionError,
        requests.exceptions.Timeout,
        ConnectionResetError,
        ConnectionAbortedError,
        BrokenPipeError,
    )
    _FETCH_MAX_RETRIES = 3
    _FETCH_BACKOFF_SECONDS = (30, 60, 120)
    assert len(_FETCH_BACKOFF_SECONDS) == _FETCH_MAX_RETRIES, (
        "backoff tuple length must match retry count"
    )

    def _fetch_one(src):
        for attempt in range(1 + _FETCH_MAX_RETRIES):
            try:
                n = src.fetch(tmp_dir)
                return src.key, n, None
            except _TRANSIENT_ERRORS as e:
                if attempt < _FETCH_MAX_RETRIES:
                    wait = _FETCH_BACKOFF_SECONDS[attempt]
                    logger.warning(
                        f"Source {src.key} failed (attempt {attempt + 1}/"
                        f"{1 + _FETCH_MAX_RETRIES}): {e!r}  — retrying in {wait}s"
                    )
                    time.sleep(wait)
                    continue
                logger.exception(
                    f"Source {src.key} exhausted all "
                    f"{1 + _FETCH_MAX_RETRIES} retries"
                )
                return src.key, 0, repr(e)
            except Exception as e:  # noqa: BLE001 — surface every failure
                return src.key, 0, repr(e)

    # All five in parallel. max_workers=5 lets each source own a thread
    # and run end-to-end without blocking on its peers.
    with ThreadPoolExecutor(max_workers=len(ALL_SOURCES)) as pool:
        futures = [pool.submit(_fetch_one, src) for src in ALL_SOURCES]
        for fut in as_completed(futures):
            key, n, err = fut.result()
            counts[key] = n
            if err:
                errors[key] = err

    # If the REQUIRED source (OpenAlex) failed, abort the whole module
    # — there's nothing meaningful to assert against. Optional sources
    # are reported as test failures by their own per-source tests so
    # the report still tells us which one broke.
    if "openalex" in errors:
        pytest.fail(
            f"Required OpenAlex source failed to download: {errors['openalex']}"
        )

    # Stash the count + error info on the dir for the per-source tests
    # to assert against. tmp_path is otherwise a plain Path so we use a
    # sidecar JSON file.
    (tmp_dir / "_release_gate_meta.json").write_text(
        json.dumps({"counts": counts, "errors": errors})
    )

    return tmp_dir


def _meta(data_dir: Path) -> dict:
    return json.loads((data_dir / "_release_gate_meta.json").read_text())


# ---------------------------------------------------------------------------
# Per-source download tests
# Each one asserts that ONE source downloaded successfully, the file is
# present, and the field names we depend on at build time are still
# there. They run instantly because the download already happened in
# the parallel fixture.
# ---------------------------------------------------------------------------


def test_openalex_sources(downloaded_data_dir: Path):
    """OpenAlex sources file is gzipped JSON with the compact-record
    field names we read in db.py::_populate_sources."""
    meta = _meta(downloaded_data_dir)
    assert "openalex" not in meta["errors"], (
        f"OpenAlex fetch failed: {meta['errors'].get('openalex')}"
    )
    assert meta["counts"]["openalex"] >= MIN_OPENALEX_SOURCES

    f = downloaded_data_dir / "openalex_sources.json.gz"
    assert f.exists()
    assert f.stat().st_size > 1_000_000

    with gzip.open(f, "rt", encoding="utf-8") as fh:
        payload = json.load(fh)
    assert "s" in payload
    sources = payload["s"]
    assert len(sources) >= MIN_OPENALEX_SOURCES

    # Spot-check a sample — the field names are the wire contract
    # between OpenAlex's API and our build pipeline. ``cb`` is the new
    # cited_by_count field added for quartile derivation; if OpenAlex
    # ever drops that field this assertion fires.
    sample = next(iter(sources.values()))
    expected = {"n", "t", "h", "if", "cb", "p", "i"}
    missing = expected - set(sample.keys())
    assert len(missing) < len(expected) / 2, (
        f"OpenAlex compact record missing too many expected keys: "
        f"{missing} (sample={sample!r})"
    )


def test_doaj_journals(downloaded_data_dir: Path):
    """DOAJ dump downloaded with the field names we read at build time."""
    meta = _meta(downloaded_data_dir)
    assert "doaj" not in meta["errors"], (
        f"DOAJ fetch failed: {meta['errors'].get('doaj')}"
    )
    assert meta["counts"]["doaj"] >= MIN_DOAJ_JOURNALS

    f = downloaded_data_dir / "doaj_journals.json"
    assert f.exists()
    data = json.loads(f.read_text())
    assert isinstance(data, dict)
    assert len(data) >= MIN_DOAJ_JOURNALS

    sample = next(iter(data.values()))
    # Field names consumed by the DOAJ pass in db.py::_populate_sources.
    assert "name" in sample
    assert "publisher" in sample


def test_predatory_list(downloaded_data_dir: Path):
    """Stop-predatory-journals lists downloaded and shaped correctly."""
    meta = _meta(downloaded_data_dir)
    assert "predatory" not in meta["errors"], (
        f"Predatory fetch failed: {meta['errors'].get('predatory')}"
    )
    assert meta["counts"]["predatory"] >= MIN_PREDATORY_JOURNALS

    f = downloaded_data_dir / "predatory.json"
    assert f.exists()
    data = json.loads(f.read_text())
    assert "journals" in data
    assert "publishers" in data
    assert "hijacked" in data
    assert len(data["journals"]) >= MIN_PREDATORY_JOURNALS


def test_jabref_abbreviations(downloaded_data_dir: Path):
    """JabRef abbreviation list downloaded with sane row counts."""
    meta = _meta(downloaded_data_dir)
    assert "jabref" not in meta["errors"], (
        f"JabRef fetch failed: {meta['errors'].get('jabref')}"
    )
    assert meta["counts"]["jabref"] >= MIN_ABBREVIATIONS

    f = downloaded_data_dir / "jabref_abbreviations.json.gz"
    assert f.exists()
    with gzip.open(f, "rt", encoding="utf-8") as fh:
        data = json.load(fh)
    assert len(data) >= MIN_ABBREVIATIONS


def test_openalex_institutions(downloaded_data_dir: Path):
    """OpenAlex institutions API still returns compact records.

    This is the slowest source (~10 min for ~110K institutions via
    cursor pagination). It's not strictly required for the journal
    scoring tier — it powers the Tier 3.5 affiliation salvage path —
    but a regression here means arxiv preprints lose their
    institution-tier scoring fallback, which is a real quality drop.
    """
    meta = _meta(downloaded_data_dir)
    assert "institutions" not in meta["errors"], (
        f"Institutions fetch failed: {meta['errors'].get('institutions')}"
    )
    assert meta["counts"]["institutions"] >= MIN_INSTITUTIONS

    f = downloaded_data_dir / "openalex_institutions.json.gz"
    assert f.exists()
    with gzip.open(f, "rt", encoding="utf-8") as fh:
        data = json.load(fh)
    # Same wrapper convention as the sources file.
    institutions = data.get("i") or data.get("institutions") or data
    if isinstance(institutions, dict):
        assert len(institutions) >= MIN_INSTITUTIONS
    else:
        assert len(institutions) >= MIN_INSTITUTIONS


# ---------------------------------------------------------------------------
# Build + lookup tests — these run AFTER all five downloads have
# completed (the fixture is module-scoped so the build test sees a
# fully-populated data directory).
# ---------------------------------------------------------------------------


def test_build_journal_quality_db(downloaded_data_dir: Path):
    """``build_db()`` runs end-to-end against the freshly-downloaded
    files and produces a queryable database with all the columns from
    this PR (cited_by_count + quartile + the existing schema).
    """
    from sqlalchemy import create_engine, func, select

    from local_deep_research.journal_quality.db import build_db
    from local_deep_research.journal_quality.models import (
        Institution,
        PredatoryJournal,
        Source,
    )

    db_file = downloaded_data_dir / "journal_quality.db"
    if db_file.exists():
        # Build is the only writer; the file is chmod 0o444 by default.
        db_file.chmod(0o644)
        db_file.unlink()

    build_db(data_dir=downloaded_data_dir, output_path=db_file)

    assert db_file.exists()

    engine = create_engine(f"sqlite:///{db_file}")
    try:
        with engine.connect() as conn:
            # 1. Source table populated.
            n_sources = conn.execute(
                select(func.count()).select_from(Source)
            ).scalar()
            assert n_sources >= MIN_OPENALEX_SOURCES, (
                f"Source row count below minimum: {n_sources}"
            )

            # 2. cited_by_count populated for at least some rows (it's
            #    NULL on DOAJ-only entries by design).
            n_with_citations = conn.execute(
                select(func.count())
                .select_from(Source)
                .where(Source.cited_by_count.is_not(None))
            ).scalar()
            assert n_with_citations > 0, (
                "cited_by_count is NULL for every source — either the "
                "OpenAlex API stopped exposing the field or the openalex.py "
                "data source loader regressed."
            )

            # 3. Quartile post-pass ran and assigned every bucket.
            quartiles = {
                row[0]
                for row in conn.execute(
                    select(Source.quartile)
                    .where(Source.quartile.is_not(None))
                    .distinct()
                ).all()
            }
            assert quartiles == {"Q1", "Q2", "Q3", "Q4"}, (
                f"Quartile buckets not all populated: got {quartiles}"
            )

            # 4. Predatory list loaded into its dedicated table.
            n_pred = conn.execute(
                select(func.count()).select_from(PredatoryJournal)
            ).scalar()
            assert n_pred >= MIN_PREDATORY_JOURNALS

            # 5. Institutions loaded (used by Tier 3.5 affiliation salvage).
            n_inst = conn.execute(
                select(func.count()).select_from(Institution)
            ).scalar()
            assert n_inst >= MIN_INSTITUTIONS
    finally:
        engine.dispose()


def test_runtime_accessor_can_score_real_journal(
    downloaded_data_dir: Path,
):
    """End-to-end smoke: bind the runtime ``JournalQualityDB`` to the
    freshly-built file and score a real journal. This is the same code
    path the ``JournalReputationFilter`` uses in production.
    """
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

    from local_deep_research.journal_quality.db import JournalQualityDB

    db_file = downloaded_data_dir / "journal_quality.db"
    assert db_file.exists(), (
        "test_build_journal_quality_db must run before this test"
    )

    db = JournalQualityDB()
    db._engine = create_engine(f"sqlite:///{db_file}")
    db._SessionLocal = sessionmaker(bind=db._engine, expire_on_commit=False)

    try:
        nature = db.lookup_openalex(name="Nature")
        assert nature is not None, "Nature not found in built DB"
        assert nature["h_index"] is not None and nature["h_index"] > 1000
        # Field shape contract used by the filter (`is_in_doaj`,
        # `publisher`, `issn_l`, `openalex_source_id`).
        assert "is_in_doaj" in nature
        assert "publisher" in nature
        assert "issn_l" in nature
        assert "openalex_source_id" in nature
    finally:
        db.reset()


def test_dashboard_queries_against_real_db(downloaded_data_dir: Path):
    """Exercise the dashboard query methods against the freshly built
    DB. Same code path that ``/api/journals`` (the journal-quality
    dashboard) hits in production — if the schema or query helpers
    regress, the dashboard goes blank.
    """
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

    from local_deep_research.journal_quality.db import JournalQualityDB

    db_file = downloaded_data_dir / "journal_quality.db"
    assert db_file.exists(), (
        "test_build_journal_quality_db must run before this test"
    )

    db = JournalQualityDB()
    db._engine = create_engine(f"sqlite:///{db_file}")
    db._SessionLocal = sessionmaker(bind=db._engine, expire_on_commit=False)

    try:
        # 1. Summary card on the dashboard top.
        summary = db.get_summary()
        assert summary["total"] >= MIN_OPENALEX_SOURCES
        assert summary["avg_quality"] is not None
        assert summary["doaj_count"] >= MIN_DOAJ_JOURNALS // 2
        assert summary["predatory_count"] >= MIN_PREDATORY_JOURNALS

        # 2. Quality histogram (powers the bar chart).
        qdist = db.get_quality_distribution()
        assert qdist, "quality distribution is empty"
        assert all(int(k) >= 1 and int(k) <= 10 for k in qdist.keys())
        assert sum(qdist.values()) >= MIN_OPENALEX_SOURCES // 2

        # 3. Source breakdown (openalex / doaj / predatory / llm).
        sdist = db.get_source_distribution()
        assert "openalex" in sdist
        assert sdist["openalex"] >= MIN_OPENALEX_SOURCES // 2

        # 4. Default first page of the journals table.
        journals, total = db.get_journals_page(page=1, per_page=50)
        assert total >= MIN_OPENALEX_SOURCES
        assert len(journals) == 50
        # Default sort=quality desc — first page should be Q1 / elite.
        assert journals[0]["quality"] >= journals[-1]["quality"]
        # Field shape consumed by the dashboard JS.
        j0 = journals[0]
        for key in ("name", "quality", "h_index", "score_source"):
            assert key in j0, f"dashboard row missing field: {key}"

        # 5. Search filter — "nature" should always match a real journal.
        journals, total = db.get_journals_page(
            page=1, per_page=10, search="nature"
        )
        assert total > 0
        assert any("nature" in (j["name"] or "").lower() for j in journals)

        # 6. Tier filter — elite tier should always have entries given the
        # ~280K-row corpus.
        _, total_elite = db.get_journals_page(page=1, per_page=10, tier="elite")
        assert total_elite > 0
    finally:
        db.reset()