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}