stack/tests/lead/test_release_crate.py

"""RO-Crate 1.2 metadata and the OSF protocol record for LEAD releases."""

from __future__ import annotations

import ast
import hashlib
import json
from pathlib import Path

import httpx

from lead.release.crate import (
    CONTEXT,
    KEYWORDS,
    LICENSE,
    METADATA_FILE,
    build_crate,
    check_crate,
    encoding_format,
    write_crate,
)
from lead.release.protocol import (
    DOI_URL,
    USER_AGENT,
    fetch_registration,
    load_registration,
    normalize,
    render_protocol,
)

REPO = Path(__file__).resolve().parents[2]

BASED_ON = {"@id": DOI_URL, "@type": "CreativeWork", "name": "LEAD protocol"}


def _root(tmp_path: Path) -> Path:
    root = tmp_path / "data"
    (root / "ledger").mkdir(parents=True)
    (root / "ledger" / "claim.parquet").write_bytes(b"PAR1")
    (root / "protocol").mkdir()
    (root / "protocol" / "protocol.md").write_text("# Protocol\n")
    return root


def _crate(root: Path) -> dict:
    crate = build_crate(
        root,
        name="LEAD living review, release 0.1",
        description="Release 0.1.",
        version="0.1",
        date_published="2026-11-11",
        based_on=BASED_ON,
        entities=[{"@id": "#build", "@type": "CreateAction", "name": "build"}],
    )
    write_crate(root, crate)
    return crate


def _entity(crate: dict, entity_id: str) -> dict:
    return next(e for e in crate["@graph"] if e["@id"] == entity_id)


def test_keywords_are_the_osf_tags():
    script = REPO / "dev" / "scripts" / "osf_metadata.py"
    tree = ast.parse(script.read_text())
    tags = next(
        ast.literal_eval(node.value)
        for node in tree.body
        if isinstance(node, ast.Assign)
        and any(getattr(t, "id", "") == "TAGS" for t in node.targets)
    )
    assert list(KEYWORDS) == tags


def test_root_dataset_carries_the_release_metadata(tmp_path):
    crate = _crate(_root(tmp_path))
    assert crate["@context"] == CONTEXT
    descriptor = _entity(crate, METADATA_FILE)
    assert descriptor["conformsTo"] == {"@id": "https://w3id.org/ro/crate/1.2"}
    assert descriptor["about"] == {"@id": "./"}
    root = _entity(crate, "./")
    assert root["@type"] == "Dataset"
    assert root["version"] == "0.1"
    assert root["datePublished"] == "2026-11-11"
    assert root["license"] == {"@id": LICENSE}
    assert LICENSE == "https://creativecommons.org/licenses/by/4.0/"
    assert root["keywords"] == list(KEYWORDS)
    assert root["isBasedOn"] == {"@id": DOI_URL}
    assert root["hasPart"] == [
        {"@id": "ledger/claim.parquet"},
        {"@id": "protocol/protocol.md"},
    ]
    assert _entity(crate, DOI_URL)["name"] == "LEAD protocol"
    assert _entity(crate, "#build")["@type"] == "CreateAction"


def test_file_entities_carry_format_size_and_sha256(tmp_path):
    crate = _crate(_root(tmp_path))
    entity = _entity(crate, "protocol/protocol.md")
    assert entity["@type"] == "File"
    assert entity["encodingFormat"] == "text/markdown"
    assert entity["contentSize"] == str(len(b"# Protocol\n"))
    assert entity["sha256"] == hashlib.sha256(b"# Protocol\n").hexdigest()


def test_encoding_formats():
    assert encoding_format("ledger/claim.parquet") == "application/vnd.apache.parquet"
    assert (
        encoding_format("bibliography/included.csl.json")
        == "application/vnd.citationstyles.csl+json"
    )
    assert encoding_format("protocol/registration-8wq3m.json") == "application/json"
    assert encoding_format("tags/vocab-v2.yaml") == "application/yaml"
    assert encoding_format("code/stack-abc.tar.gz") == "application/gzip"


def test_check_crate_passes_a_fresh_crate(tmp_path):
    root = _root(tmp_path)
    _crate(root)
    assert check_crate(root) == []


def test_check_crate_reports_undescribed_missing_and_changed_files(tmp_path):
    root = _root(tmp_path)
    _crate(root)
    (root / "extra.json").write_text("{}")
    (root / "ledger" / "claim.parquet").unlink()
    (root / "protocol" / "protocol.md").write_text("# Changed\n")
    problems = check_crate(root)
    assert any("extra.json" in p and "not described" in p for p in problems)
    assert any("ledger/claim.parquet" in p and "missing" in p for p in problems)
    assert any("protocol/protocol.md" in p and "sha256" in p for p in problems)


def test_check_crate_reports_a_missing_or_broken_metadata_file(tmp_path):
    root = _root(tmp_path)
    assert check_crate(root) == [f"{METADATA_FILE} is missing"]
    (root / METADATA_FILE).write_text("{not json")
    assert check_crate(root)[0].startswith(f"{METADATA_FILE} is not JSON")


def test_check_crate_reports_a_root_without_required_properties(tmp_path):
    root = _root(tmp_path)
    crate = _crate(root)
    dataset = _entity(crate, "./")
    del dataset["license"]
    write_crate(root, crate)
    assert check_crate(root) == ["root Dataset has no license"]


# ── OSF registration ─────────────────────────────────────────────


RAW = {
    "data": {
        "id": "8wq3m",
        "attributes": {
            "title": "LEAD evidence chassis",
            "description": "Protocol for a living review.",
            "date_registered": "2026-11-01T03:36:24.703072",
            "registration_supplement": "Generalized Systematic Review Registration",
            "registration_responses": {
                "376-8": "Living review.",
                "376-10": "Preparation; search.",
                "376-76": [],
                "376-99": "Answer with no label.",
            },
            "tags": [],
        },
        "relationships": {
            "registration_schema": {"data": {"id": "schema1", "type": "x"}}
        },
    }
}

BLOCKS = [
    {
        "attributes": {
            "index": 0,
            "block_type": "page-heading",
            "display_text": "Landing Page",
        }
    },
    {"attributes": {"index": 1, "block_type": "paragraph", "display_text": "Intro."}},
    {
        "attributes": {
            "index": 2,
            "block_type": "page-heading",
            "display_text": "Review Methods",
        }
    },
    {
        "attributes": {
            "index": 3,
            "block_type": "question-label",
            "display_text": "Type of review ",
            "schema_block_group_key": "g8",
        }
    },
    {
        "attributes": {
            "index": 4,
            "block_type": "long-text-input",
            "registration_response_key": "376-8",
            "schema_block_group_key": "g8",
        }
    },
    {
        "attributes": {
            "index": 5,
            "block_type": "question-label",
            "display_text": "Review stages",
            "schema_block_group_key": "g10",
        }
    },
    {
        "attributes": {
            "index": 6,
            "block_type": "long-text-input",
            "registration_response_key": "376-10",
            "schema_block_group_key": "g10",
        }
    },
    {"attributes": {"index": 7, "block_type": "page-heading", "display_text": "Data"}},
    {
        "attributes": {
            "index": 8,
            "block_type": "question-label",
            "display_text": "Attachments",
            "schema_block_group_key": "g76",
        }
    },
    {
        "attributes": {
            "index": 9,
            "block_type": "file-input",
            "registration_response_key": "376-76",
            "schema_block_group_key": "g76",
        }
    },
]


def test_normalize_keeps_responses_verbatim_and_labels_questions():
    reg = normalize(RAW["data"], BLOCKS)
    assert reg["guid"] == "8wq3m"
    assert reg["doi"] == "10.17605/OSF.IO/8WQ3M"
    assert reg["title"] == "LEAD evidence chassis"
    assert reg["date_registered"] == "2026-11-01T03:36:24.703072"
    assert (
        reg["registration_responses"]
        == RAW["data"]["attributes"]["registration_responses"]
    )
    assert reg["questions"] == [
        {"key": "376-8", "page": "Review Methods", "label": "Type of review"},
        {"key": "376-10", "page": "Review Methods", "label": "Review stages"},
        {"key": "376-76", "page": "Data", "label": "Attachments"},
    ]


def test_render_protocol_groups_answers_by_page():
    text = render_protocol(normalize(RAW["data"], BLOCKS))
    assert text.startswith("# LEAD evidence chassis\n")
    assert "https://doi.org/10.17605/OSF.IO/8WQ3M" in text
    assert "## Review Methods\n\n### Type of review\n\nLiving review.\n" in text
    assert "### Attachments\n\n(no response)\n" in text
    assert "### 376-99\n\nAnswer with no label.\n" in text
    assert "Landing Page" not in text


def test_load_registration_reads_a_normalized_or_raw_file(tmp_path):
    reg = normalize(RAW["data"], BLOCKS)
    cached = tmp_path / "registration-8wq3m.json"
    cached.write_text(json.dumps(reg))
    assert load_registration(cached) == reg
    raw = tmp_path / "raw.json"
    raw.write_text(json.dumps(RAW))
    loaded = load_registration(raw)
    assert loaded["registration_responses"] == reg["registration_responses"]
    assert loaded["questions"] == []


def test_fetch_registration_reads_the_record_and_the_schema_blocks():
    seen: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request)
        if request.url.path == "/v2/registrations/8wq3m/":
            return httpx.Response(200, json=RAW)
        if request.url.path == "/v2/schemas/registrations/schema1/schema_blocks/":
            if request.url.params.get("page") == "2":
                return httpx.Response(
                    200, json={"data": BLOCKS[5:], "links": {"next": None}}
                )
            nxt = str(request.url.copy_merge_params({"page": "2"}))
            return httpx.Response(
                200, json={"data": BLOCKS[:5], "links": {"next": nxt}}
            )
        return httpx.Response(404)

    transport = httpx.MockTransport(handler)
    with httpx.Client(
        transport=transport, headers={"User-Agent": USER_AGENT}
    ) as client:
        reg = fetch_registration(client)
    assert reg == normalize(RAW["data"], BLOCKS)
    assert all(r.headers["User-Agent"] == "stack-lead-release/1.0" for r in seen)