stack/tests/lead/test_release_bagit.py

"""BagIt 1.0 writer and validator for LEAD review releases (RFC 8493)."""

from __future__ import annotations

import hashlib
from pathlib import Path

from lead.release.bagit import (
    decode_path,
    encode_path,
    read_bag_info,
    validate_bag,
    write_bag,
)

INFO = [
    ("External-Identifier", "doi:10.17605/OSF.IO/8WQ3M"),
    ("Bag-Group-Identifier", "lead-living-review"),
    ("Bagging-Date", "2026-11-11"),
    ("Version", "0.1"),
]


def _payload(bag: Path) -> None:
    data = bag / "data"
    (data / "ledger").mkdir(parents=True)
    (data / "a.txt").write_text("alpha\n")
    (data / "ledger" / "claim.parquet").write_bytes(b"\x00\x01\x02")


def test_write_bag_writes_the_four_tag_files(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    octets, streams = write_bag(bag, INFO)
    assert (octets, streams) == (9, 2)
    assert (bag / "bagit.txt").read_text() == (
        "BagIt-Version: 1.0\nTag-File-Character-Encoding: UTF-8\n"
    )
    manifest = (bag / "manifest-sha512.txt").read_text().splitlines()
    digest = hashlib.sha512(b"alpha\n").hexdigest()
    assert manifest == [
        f"{digest}  data/a.txt",
        f"{hashlib.sha512(bytes([0, 1, 2])).hexdigest()}  data/ledger/claim.parquet",
    ]
    tags = [
        line.split()[1]
        for line in (bag / "tagmanifest-sha512.txt").read_text().splitlines()
    ]
    assert tags == ["bag-info.txt", "bagit.txt", "manifest-sha512.txt"]


def test_bag_info_adds_size_and_oxum_after_the_bagging_date(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    write_bag(bag, INFO)
    labels = [label for label, _value in read_bag_info(bag)]
    assert labels == [
        "External-Identifier",
        "Bag-Group-Identifier",
        "Bagging-Date",
        "Bag-Size",
        "Payload-Oxum",
        "Version",
    ]
    assert dict(read_bag_info(bag))["Payload-Oxum"] == "9.2"
    assert dict(read_bag_info(bag))["Bag-Size"] == "9 B"


def test_a_long_bag_info_value_folds_and_reads_back_whole(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    note = " ".join(["the ledger gained claims and the vocabulary moved"] * 4)
    write_bag(bag, [*INFO, ("Internal-Sender-Description", note)])
    text = (bag / "bag-info.txt").read_text()
    assert all(len(line) <= 79 for line in text.splitlines())
    assert dict(read_bag_info(bag))["Internal-Sender-Description"] == note
    assert validate_bag(bag) == []


def test_a_fresh_bag_is_valid(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    write_bag(bag, INFO)
    assert validate_bag(bag) == []


def test_validate_reports_a_changed_payload_file(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    write_bag(bag, INFO)
    (bag / "data" / "a.txt").write_text("alpha!\n")
    problems = validate_bag(bag)
    assert any("data/a.txt" in p and "sha512" in p for p in problems)
    assert any("Payload-Oxum" in p for p in problems)


def test_validate_reports_unlisted_and_missing_payload_files(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    write_bag(bag, INFO)
    (bag / "data" / "extra.txt").write_text("x")
    (bag / "data" / "ledger" / "claim.parquet").unlink()
    problems = validate_bag(bag)
    assert any(
        "data/extra.txt" in p and "not in manifest-sha512.txt" in p for p in problems
    )
    assert any("data/ledger/claim.parquet" in p and "missing" in p for p in problems)


def test_validate_reports_a_bad_bagit_txt(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    write_bag(bag, INFO)
    (bag / "bagit.txt").write_text("BagIt-Version: one\n")
    problems = validate_bag(bag)
    assert any(p.startswith("bagit.txt") for p in problems)


def test_validate_reports_a_missing_bagit_txt_and_manifest(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    problems = validate_bag(bag)
    assert any("bagit.txt is missing" in p for p in problems)
    assert any("manifest-sha512.txt is missing" in p for p in problems)


def test_validate_reports_an_edited_tag_file(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    write_bag(bag, INFO)
    info = bag / "bag-info.txt"
    info.write_text(info.read_text().replace("0.1", "0.2"))
    problems = validate_bag(bag)
    assert problems == ["bag-info.txt: sha512 in tagmanifest-sha512.txt does not match"]


def test_validate_reports_a_manifest_path_outside_data(tmp_path):
    bag = tmp_path / "bag"
    _payload(bag)
    write_bag(bag, INFO)
    manifest = bag / "manifest-sha512.txt"
    manifest.write_text(manifest.read_text() + f"{'0' * 128}  ../escape.txt\n")
    problems = validate_bag(bag)
    assert any("../escape.txt" in p for p in problems)


def test_path_encoding_round_trips_percent_and_newlines():
    raw = "data/100%\nodd\r.txt"
    assert encode_path(raw) == "data/100%25%0Aodd%0D.txt"
    assert decode_path(encode_path(raw)) == raw