From 9d069ff9efb9fffeb6204d060c6bfd147ae3215b Mon Sep 17 00:00:00 2001 From: shuchenliu Date: Mon, 5 Oct 2026 18:12:05 -0700 Subject: [PATCH] Validate stored clique types before normalization Return null for affected inputs when base or conflation source types are malformed, and emit structured diagnostics. Preserve ancestry order while removing duplicate types and every biolink:Entity. Add regression coverage for invalid sources, batch isolation, valid conflation, individual types, and GET/POST all-null responses. Validation: 53 focused tests passed. --- src/nodenorm/handlers/normalized_nodes.py | 102 ++++++---- tests/test_normalized_nodes_lookup.py | 222 ++++++++++++++++++++++ 2 files changed, 291 insertions(+), 33 deletions(-) diff --git a/src/nodenorm/handlers/normalized_nodes.py b/src/nodenorm/handlers/normalized_nodes.py index 64df0e8..18835aa 100644 --- a/src/nodenorm/handlers/normalized_nodes.py +++ b/src/nodenorm/handlers/normalized_nodes.py @@ -2,7 +2,7 @@ import json import logging import time -from typing import Union +from typing import Optional, Union from tornado.web import HTTPError @@ -296,6 +296,42 @@ async def create_normalized_node( return normal_node +def _unresolved_node(curie: str) -> NormalizedNode: + return NormalizedNode( + curie=curie, + canonical_identifier=None, + preferred_label=None, + information_content=-1.0, + identifiers=[], + types=[], + taxa=[], + ) + + +def _has_invalid_source_type(result: dict, input_curie: str, stage: str) -> bool: + biolink_type = result.get("_source", {}).get("type") + # Missing types retain the existing NamedThing fallback. A stored clique + # must otherwise have one scalar type, even when conflation is requested. + if biolink_type is None or isinstance(biolink_type, str): + return False + + logger.error( + "invalid_source_type: input_curie=%s document_id=%s stage=%s source_type=%r; returning null", + input_curie, + result.get("_id"), + stage, + biolink_type, + extra={ + "error_code": "invalid_source_type", + "input_curie": input_curie, + "document_id": result.get("_id"), + "source_type": biolink_type, + "stage": stage, + }, + ) + return True + + async def _lookup_curie_metadata( biothings_metadata: NodeNormalizationAPINamespace, curies: list[str], conflations: dict ) -> list[NormalizedNode]: @@ -311,18 +347,13 @@ async def _lookup_curie_metadata( nodes = [] for input_curie in curies: if input_curie in malformed_curies: - node = NormalizedNode( - curie=input_curie, - canonical_identifier=None, - preferred_label=None, - information_content=-1.0, - identifiers=[], - types=[], - taxa=[], - ) - nodes.append(node) + nodes.append(_unresolved_node(input_curie)) else: result = identifier_result_lookup[input_curie] + if _has_invalid_source_type(result, input_curie, "base"): + nodes.append(_unresolved_node(input_curie)) + continue + result_source = result.get("_source", {}) identifiers = result_source.get("identifiers", []) biolink_type = result_source.get("type", None) @@ -370,6 +401,7 @@ async def _lookup_curie_metadata( replacement_types = [] skipped_conflation_curies = [] conflation_label_discovered = False + invalid_conflation = False for conflation_curie in conflation_identifiers: if conflation_curie in malformed_conflation_curies: skipped_conflation_curies.append(conflation_curie) @@ -380,7 +412,11 @@ async def _lookup_curie_metadata( skipped_conflation_curies.append(conflation_curie) continue - conflation_biolink_type = conflation_result.get("_source", {}).get("type", []) + if _has_invalid_source_type(conflation_result, input_curie, "conflation"): + invalid_conflation = True + break + + conflation_biolink_type = conflation_result.get("_source", {}).get("type") conflation_identifier_lookup = conflation_result.get("_source", {}).get("identifiers", []) if not conflation_identifier_lookup: skipped_conflation_curies.append(conflation_curie) @@ -401,6 +437,12 @@ async def _lookup_curie_metadata( preferred_label = conflation_preferred_label conflation_label_discovered = True + if invalid_conflation: + # Do not return a partial conflation or hide a corrupt + # dependency by falling back to the base clique. + nodes.append(_unresolved_node(input_curie)) + continue + replacement_types = unique_list(replacement_types) if not replacement_identifiers: @@ -451,31 +493,25 @@ async def _lookup_curie_metadata( return nodes -async def _populate_biolink_type_ancestors(biolink_type: Union[str, list[str]], canonical_identifier: str) -> list[str]: - if not isinstance(biolink_type, list): - biolink_type = [biolink_type] +async def _populate_biolink_type_ancestors(biolink_type: Optional[str], canonical_identifier: str) -> list[str]: + if biolink_type is not None and not isinstance(biolink_type, str): + raise TypeError(f"Expected a scalar Biolink type for {canonical_identifier}, got {type(biolink_type).__name__}") - biolink_type_tree = [] - for bltype in biolink_type: - if not bltype: - fallback_type = "biolink:NamedThing" - logging.error( - "No type information found for '%s'. Default type set to -> '%s'", - canonical_identifier, - fallback_type, - ) - biolink_type_tree.append(fallback_type) - else: - for anc in toolkit.get_ancestors(bltype): - biolink_type_tree.append(toolkit.get_element(anc)["class_uri"]) + if not biolink_type: + fallback_type = "biolink:NamedThing" + logging.error( + "No type information found for '%s'. Default type set to -> '%s'", + canonical_identifier, + fallback_type, + ) + return [fallback_type] + + biolink_type_tree = [toolkit.get_element(anc)["class_uri"] for anc in toolkit.get_ancestors(biolink_type)] # We need to remove `biolink:Entity` from the types returned. # (See explanation at https://github.com/TranslatorSRI/NodeNormalization/issues/173) - try: - biolink_type_tree.remove("biolink:Entity") - except ValueError: - pass - return biolink_type_tree + # Preserve first-seen order because clients use the first type as primary. + return unique_list(ancestor for ancestor in biolink_type_tree if ancestor != "biolink:Entity") def unique_list(seq) -> list: diff --git a/tests/test_normalized_nodes_lookup.py b/tests/test_normalized_nodes_lookup.py index 418726f..a386b5e 100644 --- a/tests/test_normalized_nodes_lookup.py +++ b/tests/test_normalized_nodes_lookup.py @@ -1,10 +1,14 @@ import importlib.util +import json import logging import sys +from copy import deepcopy from pathlib import Path from types import ModuleType, SimpleNamespace import pytest +from tornado.testing import AsyncHTTPTestCase +from tornado.web import Application class FakeBiolinkToolkit: @@ -42,6 +46,7 @@ def load_normalized_nodes_module(): _lookup_curie_metadata = normalized_nodes._lookup_curie_metadata _lookup_equivalent_identifiers = normalized_nodes._lookup_equivalent_identifiers create_normalized_node = normalized_nodes.create_normalized_node +get_normalized_nodes = normalized_nodes.get_normalized_nodes NormalizedNode = normalized_nodes.NormalizedNode @@ -208,3 +213,220 @@ async def test_lookup_curie_metadata_logs_skipped_conflation_curies_once(caplog) assert [identifier["i"] for identifier in nodes[0].identifiers] == ["CONF:1"] skip_logs = [record for record in caplog.records if "Skipped 1 conflation CURIEs" in record.message] assert len(skip_logs) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_type", [[], ["biolink:Drug"], ["biolink:Drug", "biolink:Gene"], {}, 0, False]) +async def test_invalid_source_type_returns_null_without_mutating_identifiers(source_type, caplog): + source = { + "type": source_type, + "identifiers": [{"i": "CANONICAL:1", "t": ["existing metadata"]}, {"i": "INPUT:1"}], + } + original_source = deepcopy(source) + namespace = fake_namespace([[hit_response("DOCUMENT:1", source)]]) + + with caplog.at_level(logging.ERROR): + response = await get_normalized_nodes(namespace, ["INPUT:1"], include_individual_types=True) + + assert response == {"INPUT:1": None} + assert source == original_source + errors = [record for record in caplog.records if getattr(record, "error_code", None) == "invalid_source_type"] + assert len(errors) == 1 + assert errors[0].levelno >= logging.ERROR + assert errors[0].input_curie == "INPUT:1" + assert errors[0].document_id == "DOCUMENT:1" + assert errors[0].source_type == source_type + assert errors[0].stage == "base" + + +@pytest.mark.asyncio +async def test_mixed_batch_keeps_valid_results_and_nulls_corrupt_sources(): + corrupt_source = { + "type": ["biolink:OrganismTaxon", "biolink:ChemicalEntity"], + "identifiers": [{"i": "CORRUPT:1"}], + } + namespace = fake_namespace( + [[hit_response("VALID:1"), hit_response("CORRUPT:1", corrupt_source), no_hit_response()]] + ) + + response = await get_normalized_nodes( + namespace, ["VALID:1", "CORRUPT:1", "MISSING:1"], include_individual_types=True + ) + + assert set(response) == {"VALID:1", "CORRUPT:1", "MISSING:1"} + assert response["CORRUPT:1"] is None + assert response["MISSING:1"] is None + assert response["VALID:1"]["id"]["identifier"] == "VALID:1" + assert response["VALID:1"]["type"] == ["biolink:ChemicalEntity"] + assert response["VALID:1"]["equivalent_identifiers"][0]["type"] == "biolink:ChemicalEntity" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("conflation_key,option", [("gp", "conflate_gene_protein"), ("dc", "conflate_chemical_drug")]) +@pytest.mark.parametrize("source_type", [[], ["biolink:Drug"], ["biolink:Drug", "biolink:Gene"]]) +@pytest.mark.parametrize("has_valid_partner", [False, True]) +async def test_corrupt_conflation_partner_nulls_original_input( + conflation_key, option, source_type, has_valid_partner, caplog +): + conflation_curies = ["PARTNER:VALID", "PARTNER:CORRUPT"] if has_valid_partner else ["PARTNER:CORRUPT"] + base_source = { + "type": "biolink:ChemicalEntity", + "identifiers": [{"i": "BASE:CANONICAL", "c": {conflation_key: conflation_curies}}, {"i": "INPUT:1"}], + } + corrupt_source = { + "type": source_type, + "identifiers": [{"i": "PARTNER:CORRUPT", "t": ["existing metadata"]}], + } + original_corrupt_source = deepcopy(corrupt_source) + conflation_responses = [hit_response("CORRUPT:DOCUMENT", corrupt_source)] + if has_valid_partner: + conflation_responses.insert(0, hit_response("PARTNER:VALID")) + namespace = fake_namespace( + [[hit_response("BASE:DOCUMENT", base_source), hit_response("UNRELATED:1")], conflation_responses] + ) + + with caplog.at_level(logging.ERROR): + response = await get_normalized_nodes( + namespace, ["INPUT:1", "UNRELATED:1"], include_individual_types=True, **{option: True} + ) + + assert set(response) == {"INPUT:1", "UNRELATED:1"} + assert response["INPUT:1"] is None + assert response["UNRELATED:1"]["id"]["identifier"] == "UNRELATED:1" + assert corrupt_source == original_corrupt_source + errors = [record for record in caplog.records if getattr(record, "error_code", None) == "invalid_source_type"] + assert len(errors) == 1 + assert errors[0].input_curie == "INPUT:1" + assert errors[0].document_id == "CORRUPT:DOCUMENT" + assert errors[0].source_type == source_type + assert errors[0].stage == "conflation" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "conflation_key,option,source_types", + [ + ("gp", "conflate_gene_protein", ["biolink:Gene", "biolink:Protein"]), + ("dc", "conflate_chemical_drug", ["biolink:ChemicalEntity", "biolink:Drug"]), + ], +) +async def test_scalar_sources_conflate_with_ordered_ancestry_and_scalar_individual_types( + conflation_key, option, source_types, monkeypatch +): + monkeypatch.setattr( + normalized_nodes.toolkit, + "get_ancestors", + lambda source_type: [ + source_type, + "biolink:NamedThing", + "biolink:Entity", + "biolink:NamedThing", + "biolink:Entity", + ], + ) + base_source = { + "type": source_types[0], + "identifiers": [{"i": "INPUT:1", "c": {conflation_key: ["PARTNER:1", "PARTNER:2"]}}], + } + partner_sources = [ + {"type": source_type, "identifiers": [{"i": f"PARTNER:{index}"}]} + for index, source_type in enumerate(source_types, start=1) + ] + namespace = fake_namespace( + [ + [hit_response("INPUT:1", base_source)], + [hit_response(f"PARTNER:{index}", source) for index, source in enumerate(partner_sources, start=1)], + ] + ) + + response = await get_normalized_nodes(namespace, ["INPUT:1"], include_individual_types=True, **{option: True}) + + node = response["INPUT:1"] + assert node["type"] == [source_types[0], "biolink:NamedThing", source_types[1]] + assert node["equivalent_identifiers"] == [ + {"identifier": f"PARTNER:{index}", "type": source_type} + for index, source_type in enumerate(source_types, start=1) + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_type", [[], ["biolink:Gene"], ["biolink:Gene", "biolink:Protein"], {}, 0, False]) +async def test_biolink_ancestor_helper_rejects_non_scalar_types(source_type): + with pytest.raises(TypeError): + await normalized_nodes._populate_biolink_type_ancestors(source_type, "INPUT:1") + + +@pytest.mark.asyncio +async def test_biolink_ancestor_helper_deduplicates_in_order_and_filters_every_entity(monkeypatch): + monkeypatch.setattr( + normalized_nodes.toolkit, + "get_ancestors", + lambda source_type: [ + source_type, + "biolink:NamedThing", + "biolink:Entity", + source_type, + "biolink:PhysicalEssence", + "biolink:Entity", + "biolink:NamedThing", + ], + ) + + result = await normalized_nodes._populate_biolink_type_ancestors("biolink:Drug", "INPUT:1") + + assert result == ["biolink:Drug", "biolink:NamedThing", "biolink:PhysicalEssence"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_type", ["missing", None, ""]) +@pytest.mark.parametrize("conflate", [False, True]) +async def test_missing_source_type_keeps_named_thing_fallback(source_type, conflate): + source = {"identifiers": [{"i": "INPUT:1"}]} + if source_type != "missing": + source["type"] = source_type + response_batches = [[hit_response("INPUT:1", source)]] + if conflate: + base_source = { + "type": "biolink:Gene", + "identifiers": [{"i": "INPUT:1", "c": {"gp": ["PARTNER:1"]}}], + } + response_batches = [[hit_response("INPUT:1", base_source)], [hit_response("PARTNER:1", source)]] + namespace = fake_namespace(response_batches) + + response = await get_normalized_nodes(namespace, ["INPUT:1"], conflate_gene_protein=conflate) + + assert response["INPUT:1"]["type"] == ["biolink:NamedThing"] + + +class TestInvalidSourceTypeHTTP(AsyncHTTPTestCase): + def get_app(self): + app = Application([(r"/get_normalized_nodes", normalized_nodes.NormalizedNodesHandler)]) + app.biothings = fake_namespace( + [ + [ + hit_response("CORRUPT:1", {"type": [], "identifiers": [{"i": "CORRUPT:1"}]}), + hit_response( + "CORRUPT:2", + {"type": ["biolink:Gene", "biolink:Protein"], "identifiers": [{"i": "CORRUPT:2"}]}, + ), + ] + ] + ) + return app + + def test_get_all_corrupt_batch_returns_200_with_nulls(self): + response = self.fetch("/get_normalized_nodes?curie=CORRUPT%3A1&curie=CORRUPT%3A2") + + assert response.code == 200 + assert json.loads(response.body) == {"CORRUPT:1": None, "CORRUPT:2": None} + + def test_post_all_corrupt_batch_returns_200_with_nulls(self): + response = self.fetch( + "/get_normalized_nodes", + method="POST", + headers={"Content-Type": "application/json"}, + body=json.dumps({"curies": ["CORRUPT:1", "CORRUPT:2"]}), + ) + + assert response.code == 200 + assert json.loads(response.body) == {"CORRUPT:1": None, "CORRUPT:2": None}