tests/test_redundancy.py

from __future__ import annotations

from pathlib import Path

from gcmd_classifier.classification import (
    ClassificationCandidate,
    remove_redundant_classifications,
    validate_candidate,
)
from gcmd_classifier.models import ClassificationRecord, SupportType
from gcmd_classifier.vocabulary import load_vocabulary

FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json")


def _index():
    return load_vocabulary(FIXTURE_PATH)


def _candidate(uuid: str, *, branch_id: str | None) -> ClassificationCandidate:
    index = _index()
    record = index.get(uuid)
    topic = record if record.level == "Topic" else index.get(index.ancestors_of(uuid)[-1])
    term = None
    if record.level == "Term":
        term = record
    elif record.level.startswith("Variable_Level_"):
        term = next(
            index.get(ancestor)
            for ancestor in index.ancestors_of(uuid)
            if index.get(ancestor).level == "Term"
        )
    return ClassificationCandidate(
        final_uuid=record.UUID,
        final_name=record.name,
        final_level=record.level,
        final_canonical_path=record.canonical_path,
        path_components=record.path_components,
        topic_uuid=topic.UUID,
        topic_name=topic.name,
        term_uuid=None if term is None else term.UUID,
        term_name=None if term is None else term.name,
        branch_id=branch_id,
        evidence=f"Evidence for {record.name}.",
        support_type=SupportType.EXPLICIT,
        confidence=0.7,
    )


def _record(uuid: str, *, branch_id: str | None) -> ClassificationRecord:
    result = validate_candidate(_candidate(uuid, branch_id=branch_id), _index())
    assert result.classification is not None
    return result.classification


def _warning_codes(result) -> list[str]:
    return [warning.code for warning in result.warnings]


def test_exact_duplicate_uuid_removed() -> None:
    first = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a")
    duplicate = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a-copy")

    result = remove_redundant_classifications([first, duplicate], _index())

    assert [record.UUID for record in result.classifications] == ["vl2-carbon-dioxide"]
    assert [record.UUID for record in result.removed] == ["vl2-carbon-dioxide"]
    assert "DUPLICATE_UUID_REMOVED" in _warning_codes(result)


def test_exact_duplicate_canonical_path_removed() -> None:
    first = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a")
    duplicate_path = _record("vl2-methane", branch_id="topic:t/term:x/variable:b").model_copy(
        update={"canonical_path": first.canonical_path}
    )

    result = remove_redundant_classifications([first, duplicate_path], _index())

    assert [record.UUID for record in result.classifications] == ["vl2-carbon-dioxide"]
    assert [record.UUID for record in result.removed] == ["vl2-methane"]
    assert "DUPLICATE_CANONICAL_PATH_REMOVED" in _warning_codes(result)


def test_ancestor_removed_when_descendant_exists_in_same_branch_lineage() -> None:
    ancestor = _record("term-atmospheric-chemistry", branch_id="topic:t/term:x")
    descendant = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a")

    result = remove_redundant_classifications([ancestor, descendant], _index())

    assert [record.UUID for record in result.classifications] == ["vl2-carbon-dioxide"]
    assert [record.UUID for record in result.removed] == ["term-atmospheric-chemistry"]
    assert "ANCESTOR_REMOVED_SAME_BRANCH" in _warning_codes(result)


def test_deepest_descendant_retained() -> None:
    term = _record("term-atmospheric-chemistry", branch_id="topic:t/term:x")
    variable_level_1 = _record("vl1-atmosphere-carbon", branch_id="topic:t/term:x/variable:a")
    variable_level_3 = _record(
        "vl3-carbon-dioxide-profiles",
        branch_id="topic:t/term:x/variable:a/variable:b/variable:c",
    )

    result = remove_redundant_classifications([term, variable_level_1, variable_level_3], _index())

    assert [record.UUID for record in result.classifications] == ["vl3-carbon-dioxide-profiles"]
    assert {record.UUID for record in result.removed} == {
        "term-atmospheric-chemistry",
        "vl1-atmosphere-carbon",
    }


def test_ancestor_preserved_when_independent_branch() -> None:
    ancestor = _record("term-atmospheric-chemistry", branch_id="topic:t/term:independent")
    descendant = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a")

    result = remove_redundant_classifications([ancestor, descendant], _index())

    assert [record.UUID for record in result.classifications] == [
        "term-atmospheric-chemistry",
        "vl2-carbon-dioxide",
    ]
    assert result.removed == ()
    assert "ANCESTOR_PRESERVED_INDEPENDENT_BRANCH" in _warning_codes(result)


def test_sibling_classifications_preserved() -> None:
    dioxide = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a")
    methane = _record("vl2-methane", branch_id="topic:t/term:x/variable:b")

    result = remove_redundant_classifications([dioxide, methane], _index())

    assert [record.UUID for record in result.classifications] == [
        "vl2-carbon-dioxide",
        "vl2-methane",
    ]
    assert result.removed == ()


def test_same_concept_name_in_different_branches_is_preserved_when_uuid_and_path_differ() -> None:
    atmosphere_carbon = _record("vl1-atmosphere-carbon", branch_id="topic:a/term:a/variable:a")
    ocean_carbon = _record("vl1-ocean-carbon", branch_id="topic:o/term:o/variable:o")

    result = remove_redundant_classifications([atmosphere_carbon, ocean_carbon], _index())

    assert [record.name for record in result.classifications] == ["CARBON", "CARBON"]
    assert [record.UUID for record in result.classifications] == [
        "vl1-atmosphere-carbon",
        "vl1-ocean-carbon",
    ]
    assert result.removed == ()


def test_redundancy_uses_uuid_ancestry_not_string_prefix_matching() -> None:
    atmosphere_carbon = _record("vl1-atmosphere-carbon", branch_id="topic:a/term:a/variable:a")
    ocean_carbon = _record("vl1-ocean-carbon", branch_id="topic:o/term:o/variable:o")

    assert atmosphere_carbon.name == ocean_carbon.name
    result = remove_redundant_classifications([atmosphere_carbon, ocean_carbon], _index())

    assert result.retained_count == 2
    assert result.removed_count == 0


def test_insufficient_branch_provenance_preserves_ancestor_with_warning() -> None:
    ancestor = _record("term-atmospheric-chemistry", branch_id=None)
    descendant = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a")

    result = remove_redundant_classifications([ancestor, descendant], _index())

    assert [record.UUID for record in result.classifications] == [
        "term-atmospheric-chemistry",
        "vl2-carbon-dioxide",
    ]
    assert "ANCESTOR_PRESERVED_INSUFFICIENT_PROVENANCE" in _warning_codes(result)


def test_removal_and_preservation_diagnostics_include_branch_provenance() -> None:
    ancestor = _record("term-atmospheric-chemistry", branch_id="topic:t/term:x")
    descendant = _record("vl2-carbon-dioxide", branch_id="topic:t/term:x/variable:a")

    result = remove_redundant_classifications([ancestor, descendant], _index())

    removal_warning = next(
        warning for warning in result.warnings if warning.code == "ANCESTOR_REMOVED_SAME_BRANCH"
    )
    assert removal_warning.details["branch_id"] == "topic:t/term:x"
    assert removal_warning.details["descendant_branch_id"] == "topic:t/term:x/variable:a"