From 94d21af61f28c07250399a89a404f7d8217f977d Mon Sep 17 00:00:00 2001 From: Kori Kuzma Date: Fri, 6 Nov 2026 09:51:16 -0500 Subject: [PATCH] fix(hgvstools): preserve attributes when closing the uta connection close #488 --- src/ga4gh/vrs/utils/hgvs_tools.py | 67 +++++++++++++++++++++++------- tests/extras/test_hgvs_tools.py | 68 +++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+), 14 deletions(-) diff --git a/src/ga4gh/vrs/utils/hgvs_tools.py b/src/ga4gh/vrs/utils/hgvs_tools.py index 85825436..165a38df 100644 --- a/src/ga4gh/vrs/utils/hgvs_tools.py +++ b/src/ga4gh/vrs/utils/hgvs_tools.py @@ -2,7 +2,8 @@ import logging import re -from typing import Literal, TypeGuard +from types import TracebackType +from typing import Literal, Self, TypeGuard import hgvs import hgvs.dataproviders.uta @@ -178,7 +179,10 @@ def _vrs_pos_to_hgvs( class HgvsTools: - """A convenience class that exposes only the tools needed by vrs-python for working with HGVS (Human Genome Variation Society) notation. + """Provide tools for working with HGVS notation. + + Use a context manager or call ``close()`` to close this instance's UTA + connection. The caller owns the supplied data proxy, which stays open. Attributes: parser (hgvs.parser.Parser): The HGVS parser object. @@ -192,26 +196,61 @@ class HgvsTools: hgvs_re = re.compile(r"[^:]+:[cgmnpr]\.") def __init__(self, data_proxy: _DataProxy | None = None) -> None: - """Initialize object. + """Create the HGVS parser, normalizer, variant mapper, and UTA connection. + + Close the UTA connection if normalizer or variant mapper setup fails. + If cleanup also fails, raise its error with the setup error chained. - :param data_proxy: GA4GH data proxy instance + :param data_proxy: Optional caller-owned GA4GH data proxy. """ self.parser = hgvs.parser.Parser() self.uta_conn = hgvs.dataproviders.uta.connect() - self.normalizer = hgvs.normalizer.Normalizer(self.uta_conn, validate=True) - self.variant_mapper = hgvs.variantmapper.VariantMapper(self.uta_conn) + self._closed = False + try: + self.normalizer = hgvs.normalizer.Normalizer(self.uta_conn, validate=True) + self.variant_mapper = hgvs.variantmapper.VariantMapper(self.uta_conn) + except BaseException: + self.close() + raise self.data_proxy = data_proxy def close(self) -> None: - """Close resources. + """Close this instance's UTA connection. - # TODO These should only be closed if they are owned by this instance + Repeated calls after success do nothing. Cleanup errors propagate; + callers can retry. Keep all resource references and leave the supplied + data proxy open. Operations that need UTA are unsupported after closure. """ - self.normalizer = None - self.variant_mapper = None - self.data_proxy = None - if self.uta_conn is not None: + if not self._closed: self.uta_conn.close() + self._closed = True + + def __enter__(self) -> Self: + """Return this instance for use in a context manager. + + :return: This instance. + :raises RuntimeError: If already closed. + """ + if self._closed: + msg = "HgvsTools is closed" + raise RuntimeError(msg) + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + """Close the owned UTA connection without suppressing exceptions. + + If cleanup fails, raise its error with any context body error chained. + + :param exc_type: Exception type from the context body, if any. + :param exc_value: Exception from the context body, if any. + :param traceback: Traceback from the context body, if any. + """ + self.close() # convenience methods for hgvs parsing, normalization, and some mappings def parse(self, hgvs_str: str) -> HgvsSequenceVariant | None: @@ -586,7 +625,7 @@ def c_to_n(self, hgvs: HgvsSequenceVariant) -> HgvsSequenceVariant: vrs_allele = models.Allele.model_validate(allele_dict) dp = create_dataproxy(seqrepo_uri) - hgvs_tools = HgvsTools(dp) - hgvs_expr = hgvs_tools.from_allele(vrs_allele, namespace="refseq") + with HgvsTools(dp) as hgvs_tools: + hgvs_expr = hgvs_tools.from_allele(vrs_allele, namespace="refseq") print(hgvs_expr) # noqa: T201 diff --git a/tests/extras/test_hgvs_tools.py b/tests/extras/test_hgvs_tools.py index fde53e60..84cb0f49 100644 --- a/tests/extras/test_hgvs_tools.py +++ b/tests/extras/test_hgvs_tools.py @@ -1,5 +1,7 @@ """Tests for ga4gh.vrs.utils.hgvs_tools.""" +from unittest.mock import Mock + import hgvs.parser import pytest @@ -40,3 +42,69 @@ def test_is_intronic(hgvs_tools, hgvs_expr, expected): sv = hgvs_tools.parse(hgvs_expr) assert sv is not None assert hgvs_tools.is_intronic(sv) is expected + + +@pytest.fixture +def lifecycle(monkeypatch): + """Mock the UTA connection, normalizer, and variant mapper.""" + connection = Mock() + monkeypatch.setattr("hgvs.dataproviders.uta.connect", Mock(return_value=connection)) + monkeypatch.setattr("hgvs.normalizer.Normalizer", Mock()) + monkeypatch.setattr("hgvs.variantmapper.VariantMapper", Mock()) + return connection + + +def test_close_preserves_borrowed_resources(lifecycle): + """Close UTA once, preserve references, and leave the borrowed proxy open.""" + proxy = Mock() + tools = HgvsTools(proxy) + normalizer, mapper = tools.normalizer, tools.variant_mapper + tools.close() + tools.close() + lifecycle.close.assert_called_once_with() + proxy.close.assert_not_called() + assert tools.data_proxy is proxy + assert tools.normalizer is normalizer + assert tools.variant_mapper is mapper + with pytest.raises(RuntimeError, match="closed"): + tools.__enter__() + + +@pytest.mark.parametrize("fail", [False, True]) +def test_context_cleanup(lifecycle, fail): + """Context exit closes the owned UTA connection and propagates body errors.""" + tools = HgvsTools() + from contextlib import nullcontext + + error = ValueError("body error") + expectation = ( + pytest.raises(ValueError, match="body error") if fail else nullcontext() + ) + with expectation, tools as entered: + assert entered is tools + if fail: + raise error + lifecycle.close.assert_called_once_with() + + +@pytest.mark.parametrize("collaborator", ["Normalizer", "VariantMapper"]) +def test_initialization_failure_closes_connection(lifecycle, monkeypatch, collaborator): + """Close the owned UTA connection if normalizer or variant mapper setup fails.""" + module = "normalizer" if collaborator == "Normalizer" else "variantmapper" + monkeypatch.setattr( + f"hgvs.{module}.{collaborator}", Mock(side_effect=ValueError("init")) + ) + with pytest.raises(ValueError, match="init"): + HgvsTools() + lifecycle.close.assert_called_once_with() + + +def test_failed_cleanup_can_retry(lifecycle): + """Propagate cleanup errors and allow another close attempt.""" + tools = HgvsTools() + lifecycle.close.side_effect = [RuntimeError("cleanup"), None] + with pytest.raises(RuntimeError, match="cleanup"): + tools.close() + tools.close() + tools.close() + assert lifecycle.close.call_count == 2