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

"""Field-level schema drift detection in the OpenAlex fetcher.

A row-count floor catches the case where the whole fetch collapses,
but not the case where every row loads but a key field (``h_index``,
``cited_by_count``) has been renamed upstream. This test feeds
OpenAlex-shaped JSONL with `h_index` silently renamed to `h_idx`
and asserts that ``SchemaDriftError`` is raised instead of a
silently corrupt snapshot.
"""

from __future__ import annotations

import gzip
import io
import json
from unittest.mock import MagicMock, patch

import pytest

from local_deep_research.journal_quality.data_sources.openalex import (
    OpenAlexSource,
    SchemaDriftError,
)


def _make_jsonl_gz(records):
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="wb") as gz:
        for rec in records:
            gz.write((json.dumps(rec) + "\n").encode("utf-8"))
    return buf.getvalue()


def _manifest(num_parts: int):
    return {
        "files": [
            {"url": f"s3://openalex/data/jsonl/sources/part_{i}.gz"}
            for i in range(num_parts)
        ]
    }


def _response(*, status_code=200, content=b"", json_body=None):
    resp = MagicMock()
    resp.status_code = status_code
    resp.raise_for_status = MagicMock()
    resp.content = content
    resp.json = MagicMock(return_value=json_body or {})
    return resp


def test_missing_h_index_raises_schema_drift(tmp_path):
    # 10010 records — enough to clear the row-count floor, but every
    # one of them carries h_idx (renamed) instead of h_index.
    drifted = [
        {
            "id": f"S{i}",
            "display_name": f"J {i}",
            "type": "journal",
            "summary_stats": {
                "h_idx": 5,  # renamed from h_index
                "2yr_mean_citedness": 1.2,
            },
            "cited_by_count": 1000,
            "issn_l": None,
        }
        for i in range(10_010)
    ]
    content = _make_jsonl_gz(drifted)

    with patch(
        "local_deep_research.security.safe_requests.safe_get_with_retries",
    ) as mock_get:
        mock_get.side_effect = [
            _response(json_body=_manifest(1)),  # manifest
            _response(content=content),  # single partition
        ]
        with pytest.raises(SchemaDriftError) as exc_info:
            OpenAlexSource().fetch(tmp_path)

    msg = str(exc_info.value)
    assert "h_index present in journal sample=False" in msg
    # Existing file must not have been overwritten.
    assert not (tmp_path / OpenAlexSource().filename).exists()


def test_missing_cited_by_count_raises_schema_drift(tmp_path):
    drifted = [
        {
            "id": f"S{i}",
            "display_name": f"J {i}",
            "type": "journal",
            "summary_stats": {"h_index": 5, "2yr_mean_citedness": 1.2},
            # cited_by_count silently missing
        }
        for i in range(10_010)
    ]
    content = _make_jsonl_gz(drifted)

    with patch(
        "local_deep_research.security.safe_requests.safe_get_with_retries",
    ) as mock_get:
        mock_get.side_effect = [
            _response(json_body=_manifest(1)),
            _response(content=content),
        ]
        with pytest.raises(SchemaDriftError) as exc_info:
            OpenAlexSource().fetch(tmp_path)

    assert "cited_by_count present in journal sample=False" in str(
        exc_info.value
    )


def test_healthy_snapshot_passes(tmp_path):
    healthy = [
        {
            "id": f"S{i}",
            "display_name": f"J {i}",
            "type": "journal",
            "summary_stats": {"h_index": 5, "2yr_mean_citedness": 1.2},
            "cited_by_count": 1000,
        }
        for i in range(10_010)
    ]
    content = _make_jsonl_gz(healthy)

    with patch(
        "local_deep_research.security.safe_requests.safe_get_with_retries",
    ) as mock_get:
        mock_get.side_effect = [
            _response(json_body=_manifest(1)),
            _response(content=content),
        ]
        count = OpenAlexSource().fetch(tmp_path)

    assert count == 10_010
    assert (tmp_path / OpenAlexSource().filename).exists()


def test_missing_id_raises_schema_drift(tmp_path):
    """If the ``id`` field is renamed (e.g. to ``source_id``), every record
    gets silently skipped at parse time. Without a dedicated drift check
    this surfaces only as the generic row-count floor RuntimeError, hiding
    the real cause. Guard that the ``SchemaDriftError`` path fires first.
    """
    drifted = [
        {
            "source_id": f"S{i}",  # renamed from id
            "display_name": f"J {i}",
            "type": "journal",
            "summary_stats": {"h_index": 5, "2yr_mean_citedness": 1.2},
            "cited_by_count": 1000,
        }
        for i in range(10_010)
    ]
    content = _make_jsonl_gz(drifted)

    with patch(
        "local_deep_research.security.safe_requests.safe_get_with_retries",
    ) as mock_get:
        mock_get.side_effect = [
            _response(json_body=_manifest(1)),
            _response(content=content),
        ]
        with pytest.raises(SchemaDriftError) as exc_info:
            OpenAlexSource().fetch(tmp_path)

    msg = str(exc_info.value)
    assert "none carried an 'id' field" in msg
    assert not (tmp_path / OpenAlexSource().filename).exists()


def test_conference_only_snapshot_does_not_false_trigger(tmp_path):
    """Conferences and other non-journal source types legitimately lack
    ``h_index`` — a sample that's all conferences must not raise drift.
    Journal records with ``h_index`` drive the check; conference records
    are excluded from the sample.
    """
    mixed = []
    # 10,010 conference records without h_index (would false-trigger if
    # the sample were unfiltered).
    mixed.extend(
        {
            "id": f"C{i}",
            "display_name": f"Conf {i}",
            "type": "conference",
            "summary_stats": {"2yr_mean_citedness": 1.2},
            "cited_by_count": 500,
        }
        for i in range(10_010)
    )
    # 100 journals with h_index — enough to fill the drift sample.
    mixed.extend(
        {
            "id": f"J{i}",
            "display_name": f"Journal {i}",
            "type": "journal",
            "summary_stats": {"h_index": 5, "2yr_mean_citedness": 1.2},
            "cited_by_count": 1000,
        }
        for i in range(100)
    )
    content = _make_jsonl_gz(mixed)

    with patch(
        "local_deep_research.security.safe_requests.safe_get_with_retries",
    ) as mock_get:
        mock_get.side_effect = [
            _response(json_body=_manifest(1)),
            _response(content=content),
        ]
        count = OpenAlexSource().fetch(tmp_path)

    assert count == 10_110
    assert (tmp_path / OpenAlexSource().filename).exists()