Repository: cogent3/cogent3
Pull request #2295 (2025-04-30 01:57:26)

Add tests for storage driver plugins
## Summary by Sourcery

Add tests for storage driver plugins in cogent3, focusing on testing different storage backends for sequence collections

New Features:
- Added support for testing HDF5 sequence storage drivers

Bug Fixes:
- Ensured storage backend can be explicitly set or defaulted

Tests:
- Added parametrized tests for selecting storage drivers
- Added tests for degapping sequences with different storage backends
- Added tests to verify storage backend propagation

Chores:
- Updated noxfile to install cogent3-h5seqs for testing
- Updated CI workflow to install cogent3-h5seqs


Patch:
diff --git a/noxfile.py b/noxfile.py
index 1983023eb..bc48d6fbb 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -18,7 +18,13 @@ def test_slow(session):
 
 @nox.session(python=[f"3.{v}" for v in _py_versions])
 def test(session):
+    # the following line is temporary until we have both
+    # cogent3-h5seqs and cogent3 with driver support on pypi
+    session.install(
+        "cogent3-h5seqs @ git+https://github.com/GavinHuttley/cogent3-h5seqs.git@develop",
+    )
     session.install("-e.[test]")
+    session.run("pip", "list")
     # doctest modules within cogent3/app
     session.chdir("src/cogent3/app")
     session.run(
diff --git a/src/cogent3/_plugin.py b/src/cogent3/_plugin.py
index f37702b71..6d5fceb4a 100644
--- a/src/cogent3/_plugin.py
+++ b/src/cogent3/_plugin.py
@@ -149,6 +149,19 @@ def get_seq_format_writer_plugin(
 UNALIGNED_SEQ_STORAGE_ENTRY_POINT = "cogent3.storage.unaligned_seqs"
 
 
+def _get_driver(namespace: str, storage_backend: str) -> stevedore.driver.DriverManager:
+    try:
+        mgr = stevedore.driver.DriverManager(
+            namespace=namespace,
+            name=storage_backend,
+            invoke_on_load=False,
+        )
+    except stevedore.exception.NoMatches as err:
+        msg = f"Invalid storage backend {storage_backend!r}"
+        raise ValueError(msg) from err
+    return mgr
+
+
 def get_unaligned_storage_driver(
     storage_backend: str,
 ) -> typing.Optional["SeqsDataABC"]:
@@ -162,11 +175,7 @@ def get_unaligned_storage_driver(
     if not storage_backend:
         return _STORAGE_DEFAULT.unaligned
 
-    mgr = stevedore.driver.DriverManager(
-        namespace=UNALIGNED_SEQ_STORAGE_ENTRY_POINT,
-        name=storage_backend,
-        invoke_on_load=False,
-    )
+    mgr = _get_driver(UNALIGNED_SEQ_STORAGE_ENTRY_POINT, storage_backend)
     return mgr.extensions[0].plugin if mgr.extensions else _STORAGE_DEFAULT.unaligned
 
 
@@ -186,11 +195,7 @@ def get_aligned_storage_driver(
     if not storage_backend:
         return _STORAGE_DEFAULT.aligned
 
-    mgr = stevedore.driver.DriverManager(
-        namespace=ALIGNED_SEQ_STORAGE_ENTRY_POINT,
-        name=storage_backend,
-        invoke_on_load=False,
-    )
+    mgr = _get_driver(ALIGNED_SEQ_STORAGE_ENTRY_POINT, storage_backend)
     return mgr.extensions[0].plugin if mgr.extensions else _STORAGE_DEFAULT.aligned
 
 
diff --git a/src/cogent3/core/new_alignment.py b/src/cogent3/core/new_alignment.py
index 6b7375334..8e78deabb 100644
--- a/src/cogent3/core/new_alignment.py
+++ b/src/cogent3/core/new_alignment.py
@@ -1060,7 +1060,12 @@ class SequenceCollection:
         -----
         The returned collection will not retain an annotation_db if present.
         """
-        klass = cogent3._plugin.get_unaligned_storage_driver(storage_backend)  # noqa: SLF001
+        if storage_backend:
+            make_storage = cogent3._plugin.get_unaligned_storage_driver(
+                storage_backend,
+            ).from_seqs
+        else:
+            make_storage = self._seqs_data.from_seqs
         data = {}
         for name in self.names:
             # because we are in a SequenceCollection, which cannot be sliced, so
@@ -1069,7 +1074,7 @@ class SequenceCollection:
             data[name] = self.moltype.degap(seq)
 
         init_kwargs = self._get_init_kwargs()
-        init_kwargs["seqs_data"] = klass.from_seqs(
+        init_kwargs["seqs_data"] = make_storage(
             data=data,
             alphabet=self._seqs_data.alphabet,
             check=False,


Test patch:
diff --git a/tests/test_core/test_new_alignment.py b/tests/test_core/test_new_alignment.py
index a858b9dcb..0148f76c9 100644
--- a/tests/test_core/test_new_alignment.py
+++ b/tests/test_core/test_new_alignment.py
@@ -7,7 +7,13 @@ from warnings import catch_warnings, filterwarnings
 import numpy
 import pytest
 
-from cogent3 import get_app, load_aligned_seqs, load_unaligned_seqs, open_
+from cogent3 import (
+    get_app,
+    load_aligned_seqs,
+    load_unaligned_seqs,
+    open_,
+    set_storage_defaults,
+)
 from cogent3._version import __version__
 from cogent3.core import new_alignment, new_alphabet, new_moltype, new_sequence
 from cogent3.core.annotation import Feature
@@ -16,6 +22,13 @@ from cogent3.core.location import FeatureMap, LostSpan, Span
 from cogent3.util.deserialise import deserialise_object
 from cogent3.util.misc import get_object_provenance
 
+try:
+    import cogent3_h5seqs  # noqa: F401
+
+    has_hf_seqs = True
+except ImportError:
+    has_hf_seqs = False
+
 
 @pytest.fixture(scope="session")
 def tmp_path(tmpdir_factory):
@@ -5781,3 +5794,109 @@ def test_seqcoll_storage_immutable(mk_cls):
     seqcoll = mk_cls(data, moltype="dna")
     with pytest.raises(TypeError):
         seqcoll.storage = "new_storage"
+
+
+@pytest.mark.parametrize(
+    "mk_cls",
+    [new_alignment.make_unaligned_seqs, new_alignment.make_aligned_seqs],
+)
+def test_invalid_storage_backend(mk_cls):
+    data = {
+        "seq1": "ATCG",
+        "seq2": "TAGC",
+    }
+    with pytest.raises(ValueError):
+        mk_cls(data, moltype="dna", storage_backend="invalid_backend")
+
+
+@pytest.mark.skipif(not has_hf_seqs, reason="hdf5 seqs plugin not available")
+@pytest.mark.parametrize(
+    "mk_cls",
+    [new_alignment.make_unaligned_seqs, new_alignment.make_aligned_seqs],
+)
+def test_coll_storage_select_driver(mk_cls):
+    data = {
+        "seq1": "ATCG",
+        "seq2": "TAGC",
+    }
+    storage_backend = (
+        "h5seqs_unaligned"
+        if mk_cls == new_alignment.make_unaligned_seqs
+        else "h5seqs_aligned"
+    )
+    seqcoll = mk_cls(data, moltype="dna", storage_backend=storage_backend)
+    builtin = (
+        new_alignment.SeqsData
+        if mk_cls is new_alignment.make_unaligned_seqs
+        else new_alignment.AlignedSeqsData
+    )
+    assert not isinstance(seqcoll.storage, builtin)
+    seqcoll = mk_cls(data, moltype="dna", storage_backend=None)
+    assert isinstance(seqcoll.storage, builtin)
+
+
+@pytest.mark.skipif(not has_hf_seqs, reason="hdf5 seqs plugin not available")
+@pytest.mark.parametrize(
+    "mk_cls",
+    [new_alignment.make_unaligned_seqs, new_alignment.make_aligned_seqs],
+)
+def test_coll_storage_set_default_driver(mk_cls):
+    data = {
+        "seq1": "ATCG",
+        "seq2": "TAGC",
+    }
+    aligned = mk_cls == new_alignment.make_aligned_seqs
+    storage_backend = "h5seqs_aligned" if aligned else "h5seqs_unaligned"
+    defaults_arg = "aligned_seqs" if aligned else "unaligned_seqs"
+    set_storage_defaults(**{defaults_arg: storage_backend})
+    seqcoll = mk_cls(data, moltype="dna")
+    builtin = (
+        new_alignment.SeqsData
+        if mk_cls is new_alignment.make_unaligned_seqs
+        else new_alignment.AlignedSeqsData
+    )
+    assert not isinstance(seqcoll.storage, builtin)
+    # resetting to defaults restores builtin's
+    set_storage_defaults(reset=True)
+    seqcoll = mk_cls(data, moltype="dna")
+    assert isinstance(seqcoll.storage, builtin)
+
+
+@pytest.mark.skipif(not has_hf_seqs, reason="hdf5 seqs plugin not available")
+@pytest.mark.parametrize(
+    "mk_cls",
+    [new_alignment.make_unaligned_seqs, new_alignment.make_aligned_seqs],
+)
+def test_coll_storage_degap_explicit(mk_cls):
+    data = {
+        "seq1": "ATC-G",
+        "seq2": "TAGCC",
+    }
+    aligned = mk_cls == new_alignment.make_aligned_seqs
+    seqcoll = mk_cls(
+        data,
+        moltype="dna",
+    )
+    builtin = new_alignment.AlignedSeqsData if aligned else new_alignment.SeqsData
+    assert isinstance(seqcoll.storage, builtin)
+    # we use the storage specified by the command
+    dg = seqcoll.degap(storage_backend="h5seqs_unaligned")
+    assert not isinstance(dg.storage, builtin)
+
+
+@pytest.mark.skipif(not has_hf_seqs, reason="hdf5 seqs plugin not available")
+def test_coll_storage_degap_propagates_type():
+    data = {
+        "seq1": "ATC-G",
+        "seq2": "TAGCC",
+    }
+    seqcoll = new_alignment.make_unaligned_seqs(
+        data,
+        moltype="dna",
+        storage_backend="h5seqs_unaligned",
+    )
+    builtin = new_alignment.SeqsData
+    assert not isinstance(seqcoll.storage, builtin)
+    dg = seqcoll.degap(storage_backend="h5seqs_unaligned")
+    assert not isinstance(dg.storage, builtin)
+    assert isinstance(dg.storage, seqcoll.storage.__class__)