File size: 7,581 Bytes
57b6fe2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | 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"
|