Newer
Older
TIASDataHub / tests / test_manifest.py
"""manifest.csv の組み立て(SPEC_03「manifest.csv の列」「版の作成」の行の順序).

- 列は path・sha256・kind・subject_id・capture_id のあとに `<所見>.<項目>`
  (定義の `[columns]` の順)
- path は `files/<sha256>.<ext>`(区切りは `/`)
- UTF-8(BOM なし)・改行 LF
- 行は subject_id・capture_id・sha256 の順(同じ構成なら同じバイト列)
- 所見の値は文字列ならそのまま,無ければ空(ユーザー承認済み)
"""

import csv
import hashlib
import io

from tiasdata.manifest import (
    BASE_COLUMNS,
    ManifestEntry,
    build_manifest,
    bundle_file_path,
    manifest_columns,
    manifest_sha256,
)

_COLUMNS = {"tongue_color": ["label", "severity"]}


def _sha(seed: str) -> str:
    return hashlib.sha256(seed.encode()).hexdigest()


def _entry(
    seed: str,
    subject_id: str,
    capture_id: str,
    annotations: dict | None = None,
    *,
    ext: str = "png",
    kind: str = "tongue_crop_masked",
) -> ManifestEntry:
    return ManifestEntry(
        sha256=_sha(seed),
        ext=ext,
        kind=kind,
        subject_id=subject_id,
        capture_id=capture_id,
        annotations=annotations or {},
    )


def _rows(data: bytes) -> list[list[str]]:
    return list(csv.reader(io.StringIO(data.decode("utf-8"), newline="")))


class TestColumns:
    def test_base_columns(self) -> None:
        assert BASE_COLUMNS == ("path", "sha256", "kind", "subject_id", "capture_id")

    def test_annotation_columns_follow_definition_order(self) -> None:
        columns = {"tongue_color": ["severity", "label"], "coating": ["thickness"]}
        assert manifest_columns(columns) == [
            "path",
            "sha256",
            "kind",
            "subject_id",
            "capture_id",
            "tongue_color.severity",
            "tongue_color.label",
            "coating.thickness",
        ]

    def test_no_annotation_columns(self) -> None:
        assert manifest_columns({}) == list(BASE_COLUMNS)

    def test_header_row_matches_columns(self) -> None:
        data = build_manifest([_entry("a", "S000001", "C000001")], _COLUMNS)
        assert _rows(data)[0] == manifest_columns(_COLUMNS)


class TestPath:
    def test_bundle_file_path(self) -> None:
        sha = _sha("x")
        assert bundle_file_path(sha, "png") == f"files/{sha}.png"

    def test_row_path_and_sha256(self) -> None:
        entry = _entry("a", "S000001", "C000001", ext="jpg")
        row = _rows(build_manifest([entry], {}))[1]
        assert row == [
            f"files/{entry.sha256}.jpg",
            entry.sha256,
            "tongue_crop_masked",
            "S000001",
            "C000001",
        ]


class TestEncoding:
    def test_utf8_without_bom_and_lf_only(self) -> None:
        entries = [
            _entry(
                "a", "S000001", "C000001", {"tongue_color": {"label": "red", "severity": "mild"}}
            ),
            _entry(
                "b", "S000002", "C000002", {"tongue_color": {"label": "pale", "severity": "strong"}}
            ),
        ]
        data = build_manifest(entries, _COLUMNS)
        assert not data.startswith(b"\xef\xbb\xbf")
        assert b"\r" not in data
        assert data.endswith(b"\n")
        assert data.count(b"\n") == 3
        data.decode("utf-8")

    def test_non_ascii_value_is_utf8(self) -> None:
        entry = _entry("a", "S000001", "C000001", {"note": {"text": "淡紅"}})
        data = build_manifest([entry], {"note": ["text"]})
        assert "淡紅".encode() in data
        assert not data.startswith(b"\xef\xbb\xbf")


class TestOrder:
    def test_rows_sorted_by_subject_capture_sha256(self) -> None:
        # 入力の順・SHA-256 の順と,症例・撮影の順が食い違うようにする
        e1 = _entry("z1", "S000002", "C000001")
        e2 = _entry("z2", "S000001", "C000009")
        e3 = _entry("z3", "S000001", "C000003")
        e4 = _entry("z4", "S000001", "C000003")
        data = build_manifest([e1, e2, e3, e4], {})
        got = [(r[3], r[4], r[1]) for r in _rows(data)[1:]]
        expected = sorted((e.subject_id, e.capture_id, e.sha256) for e in (e1, e2, e3, e4))
        assert got == expected
        assert got[0][0] == "S000001"
        assert got[-1][0] == "S000002"

    def test_same_entries_in_any_order_give_same_bytes(self) -> None:
        entries = [
            _entry(
                str(i),
                f"S00000{i % 3 + 1}",
                f"C00000{i}",
                {"tongue_color": {"label": "red", "severity": "mild"}},
            )
            for i in range(1, 7)
        ]
        a = build_manifest(entries, _COLUMNS)
        b = build_manifest(list(reversed(entries)), _COLUMNS)
        assert a == b
        assert manifest_sha256(a) == manifest_sha256(b)


class TestValues:
    def test_string_values_as_is(self) -> None:
        entry = _entry(
            "a", "S000001", "C000001", {"tongue_color": {"label": "dark_red", "severity": "strong"}}
        )
        row = _rows(build_manifest([entry], _COLUMNS))[1]
        assert row[5:] == ["dark_red", "strong"]

    def test_missing_annotation_is_empty(self) -> None:
        entry = _entry("a", "S000001", "C000001")
        data = build_manifest([entry], _COLUMNS)
        assert _rows(data)[1][5:] == ["", ""]

    def test_missing_field_and_none_are_empty(self) -> None:
        entry = _entry("a", "S000001", "C000001", {"coating": {"thickness": "thin", "moist": None}})
        row = _rows(build_manifest([entry], {"coating": ["moist", "thickness", "color"]}))[1]
        assert row[5:] == ["", "thin", ""]

    def test_annotation_not_in_columns_is_not_output(self) -> None:
        entry = _entry(
            "a",
            "S000001",
            "C000001",
            {"tongue_color": {"label": "red", "severity": "mild"}, "x": {"y": "secret"}},
        )
        data = build_manifest([entry], _COLUMNS)
        assert b"secret" not in data
        assert len(_rows(data)[1]) == len(BASE_COLUMNS) + 2


class TestSha256:
    def test_manifest_sha256_is_sha256_of_bytes(self) -> None:
        data = build_manifest([_entry("a", "S000001", "C000001")], {})
        digest = manifest_sha256(data)
        assert digest == hashlib.sha256(data).hexdigest()
        assert len(digest) == 64
        assert digest == digest.lower()

    def test_empty_entries_gives_header_only(self) -> None:
        data = build_manifest([], _COLUMNS)
        assert data == (",".join(manifest_columns(_COLUMNS)) + "\n").encode()