"""Per-line resilience in the OpenAlex institutions fetcher.

Mirrors the openalex.py pattern: a single corrupted line in any
partition must be skipped (with a warning) instead of aborting the
whole rebuild. Without this guard a single bad gzip line bubbles up
through the outer try/finally as an unhandled JSONDecodeError, deleting
the temp file but leaving the caller with no context about which
partition failed.
"""

from __future__ import annotations

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

import pytest

from local_deep_research.journal_quality.data_sources import (
    institutions as inst_mod,
)
from local_deep_research.journal_quality.data_sources.institutions import (
    InstitutionSource,
)


def _gz_lines(lines: list[bytes]) -> bytes:
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="wb") as gz:
        for line in lines:
            gz.write(line + b"\n")
    return buf.getvalue()


def _manifest(num_parts: int) -> dict:
    return {
        "files": [
            {"url": f"s3://openalex/data/jsonl/institutions/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_malformed_line_does_not_abort_fetch(tmp_path, monkeypatch, caplog):
    """A bad JSON line is skipped + logged, valid lines still land."""
    # Lower the floor so we don't need to construct 50K valid records
    # just to exercise the per-line resilience path.
    monkeypatch.setattr(inst_mod, "_MIN_INSTITUTIONS", 2)

    valid = (
        b'{"id": "https://openalex.org/I1", "display_name": "MIT", '
        b'"summary_stats": {"h_index": 500}, "country_code": "US", '
        b'"type": "education"}'
    )
    bad = b"this is not valid json {{{"
    valid2 = (
        b'{"id": "https://openalex.org/I2", "display_name": "Stanford", '
        b'"summary_stats": {"h_index": 480}, "country_code": "US", '
        b'"type": "education"}'
    )

    content = _gz_lines([valid, bad, valid2])

    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
        ]
        result = InstitutionSource().fetch(tmp_path)

    # Both valid records survived; the malformed line was skipped, not
    # treated as fatal.
    assert result == 2

    # Snapshot was written.
    assert (tmp_path / InstitutionSource().filename).exists()


def test_malformed_line_floor_still_protects(tmp_path, monkeypatch):
    """If the surviving records fall below the floor, fetch still aborts."""
    monkeypatch.setattr(inst_mod, "_MIN_INSTITUTIONS", 5)

    content = _gz_lines([b"garbage 1", b"garbage 2", b"garbage 3"])

    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(RuntimeError, match="suspiciously few records"):
            InstitutionSource().fetch(tmp_path)
