diff --git a/CHANGELOG.md b/CHANGELOG.md index 76684a7c5..f1fd924a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Enhancements - Optimized `get` within `VectorIndex` to be more efficient when retrieving a dataset of references. @mavaylon1 [#1248](https://github.com/hdmf-dev/hdmf/pull/1248) +- Enhanced warnings about ignoring cached namespaces. @stephprince [#1258](https://github.com/hdmf-dev/hdmf/pull/1258) - Added support in append for same dimensional args for numpy arrays. @mavaylon1 [#1261](https://github.com/hdmf-dev/hdmf/pull/1261) - Improved error messages when optional requirements are not installed. @rly [#1263](https://github.com/hdmf-dev/hdmf/pull/1263) diff --git a/src/hdmf/backends/hdf5/h5tools.py b/src/hdmf/backends/hdf5/h5tools.py index 493c4057f..e1873dd5d 100644 --- a/src/hdmf/backends/hdf5/h5tools.py +++ b/src/hdmf/backends/hdf5/h5tools.py @@ -202,24 +202,13 @@ def __load_namespaces(cls, namespace_catalog, namespaces, file_obj): namespaces = list(spec_group.keys()) readers = dict() - deps = dict() for ns in namespaces: latest_version = namespace_versions[ns] ns_group = spec_group[ns][latest_version] reader = H5SpecReader(ns_group) readers[ns] = reader - # for each namespace in the 'namespace' dataset, track all included namespaces (dependencies) - for spec_ns in reader.read_namespace(cls.__ns_spec_path): - deps[ns] = list() - for s in spec_ns['schema']: - dep = s.get('namespace') - if dep is not None: - deps[ns].append(dep) - - order = cls._order_deps(deps) - for ns in order: - reader = readers[ns] - d.update(namespace_catalog.load_namespaces(cls.__ns_spec_path, reader=reader)) + + d.update(namespace_catalog.load_namespaces(cls.__ns_spec_path, reader=readers)) return d @@ -285,37 +274,6 @@ def __get_namespaces(cls, file_obj): return used_version_names - @classmethod - def _order_deps(cls, deps): - """ - Order namespaces according to dependency for loading into a NamespaceCatalog - - Args: - deps (dict): a dictionary that maps a namespace name to a list of name of - the namespaces on which the namespace is directly dependent - Example: {'a': ['b', 'c'], 'b': ['d'], 'c': ['d'], 'd': []} - Expected output: ['d', 'b', 'c', 'a'] - """ - order = list() - keys = list(deps.keys()) - deps = dict(deps) - for k in keys: - if k in deps: - cls.__order_deps_aux(order, deps, k) - return order - - @classmethod - def __order_deps_aux(cls, order, deps, key): - """ - A recursive helper function for _order_deps - """ - if key not in deps: - return - subdeps = deps.pop(key) - for subk in subdeps: - cls.__order_deps_aux(order, deps, subk) - order.append(key) - @classmethod @docval({'name': 'source_filename', 'type': str, 'doc': 'the path to the HDF5 file to copy'}, {'name': 'dest_filename', 'type': str, 'doc': 'the name of the destination file'}, diff --git a/src/hdmf/common/__init__.py b/src/hdmf/common/__init__.py index 6b36e29cd..5f6b6534a 100644 --- a/src/hdmf/common/__init__.py +++ b/src/hdmf/common/__init__.py @@ -233,7 +233,13 @@ def get_hdf5io(**kwargs): # load the hdmf-common namespace __resources = __get_resources() if os.path.exists(__resources['namespace_path']): - __TYPE_MAP = TypeMap(NamespaceCatalog()) + # NOTE: even though HDMF does not guarantee backwards compatibility with schema + # using an older version of the experimental namespace, in practice, this has not been + # an issue, and it is costly to determine whether there is an incompatibility before issuing + # a warning. so, we ignore the experimental namespace warning by default by specifying it + # as a "core_namespace" in the NamespaceCatalog. + # see https://github.com/hdmf-dev/hdmf/pull/1258 + __TYPE_MAP = TypeMap(NamespaceCatalog(core_namespaces=[CORE_NAMESPACE, EXP_NAMESPACE],)) load_namespaces(__resources['namespace_path']) diff --git a/src/hdmf/spec/namespace.py b/src/hdmf/spec/namespace.py index 57232bd25..e7f5262a6 100644 --- a/src/hdmf/spec/namespace.py +++ b/src/hdmf/spec/namespace.py @@ -9,7 +9,7 @@ from .catalog import SpecCatalog from .spec import DatasetSpec, GroupSpec -from ..utils import docval, getargs, popargs, get_docval +from ..utils import docval, getargs, popargs, get_docval, is_newer_version _namespace_args = [ {'name': 'doc', 'type': str, 'doc': 'a description about what this namespace represents'}, @@ -229,13 +229,19 @@ class NamespaceCatalog: {'name': 'dataset_spec_cls', 'type': type, 'doc': 'the class to use for dataset specifications', 'default': DatasetSpec}, {'name': 'spec_namespace_cls', 'type': type, - 'doc': 'the class to use for specification namespaces', 'default': SpecNamespace}) + 'doc': 'the class to use for specification namespaces', 'default': SpecNamespace}, + {'name': 'core_namespaces', 'type': list, + 'doc': 'the names of the core namespaces', 'default': list()}) def __init__(self, **kwargs): """Create a catalog for storing multiple Namespaces""" self.__namespaces = OrderedDict() self.__dataset_spec_cls = getargs('dataset_spec_cls', kwargs) self.__group_spec_cls = getargs('group_spec_cls', kwargs) self.__spec_namespace_cls = getargs('spec_namespace_cls', kwargs) + + core_namespaces = getargs('core_namespaces', kwargs) + self.__core_namespaces = core_namespaces + # keep track of all spec objects ever loaded, so we don't have # multiple object instances of a spec self.__loaded_specs = dict() @@ -248,6 +254,7 @@ def __copy__(self): ret = NamespaceCatalog(self.__group_spec_cls, self.__dataset_spec_cls, self.__spec_namespace_cls) + ret.__core_namespaces = copy(self.__core_namespaces) ret.__namespaces = copy(self.__namespaces) ret.__loaded_specs = copy(self.__loaded_specs) ret.__included_specs = copy(self.__included_specs) @@ -258,6 +265,8 @@ def merge(self, ns_catalog): for name, namespace in ns_catalog.__namespaces.items(): self.add_namespace(name, namespace) + self.__core_namespaces.extend(ns_catalog.__core_namespaces) + @property @docval(returns='a tuple of the available namespaces', rtype=tuple) def namespaces(self): @@ -279,6 +288,11 @@ def spec_namespace_cls(self): """The SpecNamespace class used in this NamespaceCatalog""" return self.__spec_namespace_cls + @property + def core_namespaces(self): + """The core namespaces used in this NamespaceCatalog""" + return self.__core_namespaces + @docval({'name': 'name', 'type': str, 'doc': 'the name of this namespace'}, {'name': 'namespace', 'type': SpecNamespace, 'doc': 'the SpecNamespace object'}) def add_namespace(self, **kwargs): @@ -508,36 +522,122 @@ def __register_dependent_types_helper(spec, inc_ns, catalog, registered_types): 'type': bool, 'doc': 'whether or not to include objects from included/parent spec objects', 'default': True}, {'name': 'reader', - 'type': SpecReader, - 'doc': 'the class to user for reading specifications', 'default': None}, + 'type': (SpecReader, dict), + 'doc': 'the SpecReader or dict of SpecReader classes to use for reading specifications', + 'default': None}, returns='a dictionary describing the dependencies of loaded namespaces', rtype=dict) def load_namespaces(self, **kwargs): """Load the namespaces in the given file""" namespace_path, resolve, reader = getargs('namespace_path', 'resolve', 'reader', kwargs) + + # determine which readers and order of readers to use for loading specs if reader is None: # load namespace definition from file if not os.path.exists(namespace_path): msg = "namespace file '%s' not found" % namespace_path raise IOError(msg) - reader = YAMLSpecReader(indir=os.path.dirname(namespace_path)) - ns_path_key = os.path.join(reader.source, os.path.basename(namespace_path)) - ret = self.__included_specs.get(ns_path_key) - if ret is None: - ret = dict() + ordered_readers = [YAMLSpecReader(indir=os.path.dirname(namespace_path))] + elif isinstance(reader, SpecReader): + ordered_readers = [reader] # only one reader else: - return ret - namespaces = reader.read_namespace(namespace_path) - to_load = list() - for ns in namespaces: - if ns['name'] in self.__namespaces: - if ns['version'] != self.__namespaces.get(ns['name'])['version']: - # warn if the cached namespace differs from the already loaded namespace - warn("Ignoring cached namespace '%s' version %s because version %s is already loaded." - % (ns['name'], ns['version'], self.__namespaces.get(ns['name'])['version'])) - else: - to_load.append(ns) - # now load specs into namespace - for ns in to_load: - ret[ns['name']] = self.__load_namespace(ns, reader, resolve=resolve) - self.__included_specs[ns_path_key] = ret + deps = dict() # for each namespace, track all included namespaces (dependencies) + for ns, r in reader.items(): + for spec_ns in r.read_namespace(namespace_path): + deps[ns] = list() + for s in spec_ns['schema']: + dep = s.get('namespace') + if dep is not None: + deps[ns].append(dep) + order = self._order_deps(deps) + ordered_readers = [reader[ns] for ns in order] + + # determine which namespaces to load and which to ignore + ignored_namespaces = list() + ret = dict() + for r in ordered_readers: + # continue to next reader if spec is already included + ns_path_key = os.path.join(r.source, os.path.basename(namespace_path)) + included_specs = self.__included_specs.get(ns_path_key) + if included_specs is not None: + ret.update(included_specs) + continue # continue to next reader if spec is already included + + to_load = list() + namespaces = r.read_namespace(namespace_path) + for ns in namespaces: + if ns['name'] in self.__namespaces: + if ns['version'] != self.__namespaces.get(ns['name'])['version']: + cached_version = ns['version'] + loaded_version = self.__namespaces.get(ns['name'])['version'] + ignored_namespaces.append((ns['name'], cached_version, loaded_version)) + else: + to_load.append(ns) + + # now load specs into namespace + for ns in to_load: + ret[ns['name']] = self.__load_namespace(ns, r, resolve=resolve) + self.__included_specs[ns_path_key] = ret + + # warn if there are any ignored namespaces + if ignored_namespaces: + self.warn_for_ignored_namespaces(ignored_namespaces) + return ret + + def warn_for_ignored_namespaces(self, ignored_namespaces): + """Warning if namespaces were ignored where a different version was already loaded + + Args: + ignored_namespaces (list): name, cached version, and loaded version of the namespace + """ + core_warnings = list() + other_warnings = list() + warning_msg = list() + for name, cached_version, loaded_version in ignored_namespaces: + version_info = f"{name} - cached version: {cached_version}, loaded version: {loaded_version}" + if name in self.__core_namespaces and is_newer_version(cached_version, loaded_version): + core_warnings.append(version_info) # for core namespaces, warn if the cached version is newer + elif name not in self.__core_namespaces: + other_warnings.append(version_info) # for all other namespaces, issue a warning for compatibility + + if core_warnings: + joined_warnings = "\n".join(core_warnings) + warning_msg.append(f'{joined_warnings}\nPlease update to the latest package versions.') + if other_warnings: + joined_warnings = "\n".join(other_warnings) + warning_msg.append(f'{joined_warnings}\nThe loaded extension(s) may not be compatible with the cached ' + 'extension(s) in the file. Please check the extension documentation and ignore this ' + 'warning if these versions are compatible.') + if warning_msg: + joined_warnings = "\n".join(warning_msg) + warn(f'Ignoring the following cached namespace(s) because another version is already loaded:\n' + f'{joined_warnings}', category=UserWarning, stacklevel=2) + + def _order_deps(self, deps): + """ + Order namespaces according to dependency for loading into a NamespaceCatalog + + Args: + deps (dict): a dictionary that maps a namespace name to a list of name of + the namespaces on which the namespace is directly dependent + Example: {'a': ['b', 'c'], 'b': ['d'], 'c': ['d'], 'd': []} + Expected output: ['d', 'b', 'c', 'a'] + """ + order = list() + keys = list(deps.keys()) + deps = dict(deps) + for k in keys: + if k in deps: + self.__order_deps_aux(order, deps, k) + return order + + def __order_deps_aux(self, order, deps, key): + """ + A recursive helper function for _order_deps + """ + if key not in deps: + return + subdeps = deps.pop(key) + for subk in subdeps: + self.__order_deps_aux(order, deps, subk) + order.append(key) diff --git a/src/hdmf/utils.py b/src/hdmf/utils.py index c21382a2a..4216d7fde 100644 --- a/src/hdmf/utils.py +++ b/src/hdmf/utils.py @@ -1,5 +1,6 @@ import collections import copy as _copy +import re import types import warnings from abc import ABCMeta @@ -876,6 +877,22 @@ def is_ragged(data): return False +def is_newer_version(version_a: str, version_b: str) -> bool: + # this method could be replaced by packaging.version if packaging is added as a dependency + version_a_match = re.match(r"(\d+\.\d+\.\d+)", version_a)[0] # trim off any non-numeric symbols at end + version_a_list = [int(i) for i in version_a_match.split(".")] + + version_b_match = re.match(r"(\d+\.\d+\.\d+)", version_b)[0] # trim off any non-numeric symbols at end + version_b_list = [int(i) for i in version_b_match.split(".")] + + for a, b in zip(version_a_list, version_b_list): + if a > b: + return True + elif a < b: + return False + + return False + def get_basic_array_info(array): def convert_bytes_to_str(bytes_size): suffixes = ['bytes', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB'] diff --git a/tests/unit/spec_tests/test_load_namespace.py b/tests/unit/spec_tests/test_load_namespace.py index 5d7e6573c..b97da3f07 100644 --- a/tests/unit/spec_tests/test_load_namespace.py +++ b/tests/unit/spec_tests/test_load_namespace.py @@ -247,11 +247,17 @@ def setUp(self): self.ext_source2 = 'extension2.yaml' self.ns_path2 = 'namespace2.yaml' + # get core namespace + hdmf_typemap = get_type_map() + self.ns_catalog = hdmf_typemap.namespace_catalog + self.core_ns = self.ns_catalog.core_namespaces[0] + self.core_ns_version = self.ns_catalog.get_namespace(self.core_ns)['version'] + def tearDown(self): for f in (self.ext_source1, self.ns_path1, self.ext_source2, self.ns_path2): remove_test_file(os.path.join(self.tempdir, f)) - def test_catch_dup_name(self): + def test_catch_dup_name_extension_different(self): ns_builder1 = NamespaceBuilder('Extension doc', "test_ext", version='0.1.0') ns_builder1.add_spec(self.ext_source1, GroupSpec('doc', data_type_def='MyType')) ns_builder1.export(self.ns_path1, outdir=self.tempdir) @@ -262,11 +268,15 @@ def test_catch_dup_name(self): ns_catalog = NamespaceCatalog() ns_catalog.load_namespaces(os.path.join(self.tempdir, self.ns_path1)) - msg = "Ignoring cached namespace 'test_ext' version 0.2.0 because version 0.1.0 is already loaded." - with self.assertWarnsRegex(UserWarning, msg): + msg = ("Ignoring the following cached namespace(s) because another version is already loaded:\n" + "test_ext - cached version: 0.2.0, loaded version: 0.1.0\n" + "The loaded extension(s) may not be compatible with the cached extension(s) in the file. " + "Please check the extension documentation and ignore this warning if these versions are " + "compatible.") + with self.assertWarnsWith(UserWarning, msg): ns_catalog.load_namespaces(os.path.join(self.tempdir, self.ns_path2)) - def test_catch_dup_name_same_version(self): + def test_catch_dup_name_extension_same_version(self): ns_builder1 = NamespaceBuilder('Extension doc', "test_ext", version='0.1.0') ns_builder1.add_spec(self.ext_source1, GroupSpec('doc', data_type_def='MyType')) ns_builder1.export(self.ns_path1, outdir=self.tempdir) @@ -278,12 +288,72 @@ def test_catch_dup_name_same_version(self): ns_catalog.load_namespaces(os.path.join(self.tempdir, self.ns_path1)) # no warning should be raised (but don't just check for 0 warnings -- warnings can come from other sources) - msg = "Ignoring cached namespace 'test_ext' version 0.1.0 because version 0.1.0 is already loaded." + msg = ("Ignoring the following cached namespace(s) because another version is already loaded:\n" + "test_ext - cached version: 0.1.0, loaded version: 0.1.0\n" + "The loaded extension(s) may not be compatible with the cached extension(s) in the file. " + "Please check the extension documentation and ignore this warning if these versions are " + "compatible.") with warnings.catch_warnings(record=True) as ws: ns_catalog.load_namespaces(os.path.join(self.tempdir, self.ns_path2)) for w in ws: - self.assertTrue(str(w) != msg) + self.assertTrue(str(w.message) != msg) + warnings.warn(str(w.message), w.category) + + def test_catch_dup_name_core_newer(self): + new_ns_version = '100.0.0' + ns_builder1 = NamespaceBuilder('Extension doc', self.core_ns, version=new_ns_version) + ns_builder1.add_spec(self.ext_source1, GroupSpec('doc', data_type_def='MyType')) + ns_builder1.export(self.ns_path1, outdir=self.tempdir) + + # create new catalog and merge the loaded core namespace catalog + ns_catalog = NamespaceCatalog() + ns_catalog.merge(self.ns_catalog) + + # test loading newer namespace than one already loaded will warn + msg = (f'Ignoring the following cached namespace(s) because another version is already loaded:\n' + f'{self.core_ns} - cached version: {new_ns_version}, loaded version: {self.core_ns_version}\n' + f'Please update to the latest package versions.') + with self.assertWarnsWith(UserWarning, msg): + ns_catalog.load_namespaces(os.path.join(self.tempdir, self.ns_path1)) + + def test_catch_dup_name_core_older(self): + new_ns_version = '0.0.0' + ns_builder1 = NamespaceBuilder('Extension doc', self.core_ns, version=new_ns_version) + ns_builder1.add_spec(self.ext_source1, GroupSpec('doc', data_type_def='MyType')) + ns_builder1.export(self.ns_path1, outdir=self.tempdir) + # create new catalog and merge the loaded core namespace catalog + ns_catalog = NamespaceCatalog() + ns_catalog.merge(self.ns_catalog) + + # test no warning if loading older namespace than one already loaded + msg = (f'Ignoring the following cached namespace(s) because another version is already loaded:\n' + f'{self.core_ns} - cached version: {new_ns_version}, loaded version: {self.core_ns_version}\n' + f'Please update to the latest package versions.') + with warnings.catch_warnings(record=True) as ws: + ns_catalog.load_namespaces(os.path.join(self.tempdir, self.ns_path1)) + for w in ws: + self.assertTrue(str(w.message) != msg) + warnings.warn(str(w.message), w.category) + + def test_catch_dup_name_core_same(self): + new_ns_version = self.core_ns_version + ns_builder1 = NamespaceBuilder('Extension doc', self.core_ns, version=new_ns_version) + ns_builder1.add_spec(self.ext_source1, GroupSpec('doc', data_type_def='MyType')) + ns_builder1.export(self.ns_path1, outdir=self.tempdir) + + # create new catalog and merge the loaded core namespace catalog + ns_catalog = NamespaceCatalog() + ns_catalog.merge(self.ns_catalog) + + msg = (f'Ignoring the following cached namespace(s) because another version is already loaded:\n' + f'{self.core_ns} - cached version: {new_ns_version}, loaded version: {self.core_ns_version}\n' + f'Please update to the latest package versions.') + with warnings.catch_warnings(record=True) as ws: + ns_catalog.load_namespaces(os.path.join(self.tempdir, self.ns_path1)) + for w in ws: + self.assertTrue(str(w.message) != msg) + warnings.warn(str(w.message), w.category) class TestCustomSpecClasses(TestCase): diff --git a/tests/unit/test_io_hdf5_h5tools.py b/tests/unit/test_io_hdf5_h5tools.py index 6c679bb49..a06e89738 100644 --- a/tests/unit/test_io_hdf5_h5tools.py +++ b/tests/unit/test_io_hdf5_h5tools.py @@ -2140,6 +2140,40 @@ def tearDown(self): if os.path.exists(self.path): os.remove(self.path) + def create_test_namespace(self, name, version, source): + file_spec = GroupSpec(doc="A FooFile", data_type_def='FooFile') + spec_catalog = SpecCatalog() + namespace = SpecNamespace( + doc='a test namespace', + name=name, + schema=[{'source': source}], + version=version, + catalog=spec_catalog + ) + spec_catalog.register_spec(file_spec, source) + return namespace + + def write_test_file(self, path, ns_name, ns_source, version, mode): + namespace = self.create_test_namespace(ns_name, version, ns_source) + namespace_catalog = NamespaceCatalog() + namespace_catalog.add_namespace(ns_name, namespace) + type_map = TypeMap(namespace_catalog) + type_map.register_container_type(ns_name, 'FooFile', FooFile) + manager = BuildManager(type_map) + container = FooFile() + with HDF5IO(path, manager=manager, mode=mode) as io: + io.write(container) + + return namespace_catalog + + def replace_cached_ns_version(self, path, namespace, cached_version, new_version): + with h5py.File(path, mode='r+') as f: + old_dict = f[f'/specifications/{namespace}/{cached_version}/namespace'][()].decode('utf-8') + new_dict = old_dict.replace(f'"version":"{cached_version}"', f'"version":"{new_version}"') + + f[f'/specifications/{namespace}/{cached_version}/namespace'][()] = new_dict + f.move(f'/specifications/{namespace}/{cached_version}', f'/specifications/{namespace}/{new_version}') + def test_load_namespaces_none_version(self): """Test that reading a file with a cached namespace and None version works but raises a warning.""" # make the file have group name "None" instead of "0.1.0" (namespace version is used as group name) @@ -2178,6 +2212,45 @@ def test_load_namespaces_unversioned(self): with self.assertWarnsWith(UserWarning, msg): HDF5IO.load_namespaces(ns_catalog, self.path) + def test_load_namespaces_multiple_duplicate_versions(self): + """Test that loading a file with a cached namespace will warn if newer version than already loaded namespace""" + + # write file with two different namespaces + namespace_catalog = NamespaceCatalog() + ns1 = self.create_test_namespace('test_ext1', '0.1.0', 'test1.yaml') + ns2 = self.create_test_namespace('test_ext2', '0.1.0', 'test2.yaml') + namespace_catalog.add_namespace('test_ext1', ns1) + namespace_catalog.add_namespace('test_ext2', ns2) + type_map = TypeMap(namespace_catalog) + type_map.register_container_type('test_ext1', 'FooFile', FooFile) + type_map.register_container_type('test_ext2', 'FooFile', FooFile) + manager = BuildManager(type_map) + container = FooFile() + + path = get_temp_filepath() + with HDF5IO(path, manager=manager, mode='w') as io: + io.write(container) + + # modify the versions of the example namespaces + self.replace_cached_ns_version(path=path, + namespace='test_ext1', + cached_version='0.1.0', + new_version='100.0.0') + self.replace_cached_ns_version(path=path, + namespace='test_ext2', + cached_version='0.1.0', + new_version='100.0.0') + + # test warning is raised when loading the file and a newer version is cached + msg = ("Ignoring the following cached namespace(s) because another version is already loaded:\n" + "test_ext1 - cached version: 100.0.0, loaded version: 0.1.0\n" + "test_ext2 - cached version: 100.0.0, loaded version: 0.1.0\n" + "The loaded extension(s) may not be compatible with the cached extension(s) in the file. " + "Please check the extension documentation and ignore this warning if these versions are " + "compatible.") + with self.assertWarnsWith(UserWarning, msg): + HDF5IO.load_namespaces(namespace_catalog, path) + def test_load_namespaces_path(self): """Test that loading namespaces given a path is OK and returns the correct dictionary.""" ns_catalog = NamespaceCatalog() diff --git a/tests/unit/test_io_hdf5_streaming.py b/tests/unit/test_io_hdf5_streaming.py index d82c9c5c3..1a487b939 100644 --- a/tests/unit/test_io_hdf5_streaming.py +++ b/tests/unit/test_io_hdf5_streaming.py @@ -2,7 +2,6 @@ import os import urllib.request import h5py -import warnings from hdmf.backends.hdf5.h5tools import HDF5IO from hdmf.build import TypeMap, BuildManager @@ -80,8 +79,6 @@ def setUp(self): self.manager = BuildManager(type_map) - warnings.filterwarnings(action="ignore", message="Ignoring cached namespace .*") - def tearDown(self): if os.path.exists(self.ns_filename): os.remove(self.ns_filename) diff --git a/tests/unit/utils_test/test_utils.py b/tests/unit/utils_test/test_utils.py index b0cd05e9d..6a6e0bc0d 100644 --- a/tests/unit/utils_test/test_utils.py +++ b/tests/unit/utils_test/test_utils.py @@ -4,7 +4,7 @@ import numpy as np from hdmf.data_utils import DataChunkIterator, DataIO from hdmf.testing import TestCase -from hdmf.utils import get_data_shape, to_uint_array +from hdmf.utils import get_data_shape, to_uint_array, is_newer_version class TestGetDataShape(TestCase): @@ -204,3 +204,21 @@ def test_list_float(self): arr = [0., 1., 2.] with self.assertRaisesWith(ValueError, 'Cannot convert array of dtype float64 to uint.'): to_uint_array(arr) + +class TestVersionComparison(TestCase): + """Test the version comparison functionality in NamespaceCatalog.""" + + def test_is_newer_version(self): + """Test basic version comparison scenarios.""" + # test when first version is newer + self.assertTrue(is_newer_version("10.0.0", "2.0.0")) + self.assertTrue(is_newer_version("1.1.0", "1.0.0")) + self.assertTrue(is_newer_version("1.0.1", "1.0.0")) + + # test when second version is newer + self.assertFalse(is_newer_version("2.0.0", "10.0.0")) + self.assertFalse(is_newer_version("1.0.0", "1.1.0")) + self.assertFalse(is_newer_version("1.0.0", "1.0.1")) + + # test when versions are equal + self.assertFalse(is_newer_version("1.0.0", "1.0.0"))