From dcc73a0230e0f6b4a841cfa1fa208478407e191f Mon Sep 17 00:00:00 2001 From: Andrew Davison Date: Tue, 27 Oct 2026 10:17:11 +0100 Subject: [PATCH 1/3] Fix multiple bugs in ObjectList (#1935) extend() and += now convert their input to a list, then check every object before changing anything. This rejects duplicates within the iterable (1), handles generators (2), and leaves the list and the objects' parent links unchanged if any object is rejected (5). Membership tests (in, index(), count(), remove()) now compare by identity, as the duplicate check in append() already does (3). reverse() no longer raises after reversing (4). __setitem__ now does the same type, duplicate and parent checks as append(), for both integer and slice keys (6). Container._set_object_list() now raises the TypeError it builds (7). SpikeTrainList's append(), extend(), __setitem__ and += now delegate to ObjectList after converting a multiplexed list to SpikeTrain objects, so they get the same fixes. Fixes #1935. See also #1927. --- neo/core/container.py | 2 +- neo/core/objectlist.py | 64 ++++++--- neo/core/spiketrainlist.py | 15 +-- neo/test/coretest/test_container.py | 15 +++ neo/test/coretest/test_objectlist.py | 162 +++++++++++++++++++++++ neo/test/coretest/test_spiketrainlist.py | 24 ++++ 6 files changed, 252 insertions(+), 30 deletions(-) create mode 100644 neo/test/coretest/test_objectlist.py diff --git a/neo/core/container.py b/neo/core/container.py index 8f9980ad1..77d12a360 100644 --- a/neo/core/container.py +++ b/neo/core/container.py @@ -226,7 +226,7 @@ def _set_object_list(self, name, value): elif isinstance(value, ObjectList): # from __iadd__ setattr(self, name, value) else: - TypeError("value must be a list or an ObjectList") + raise TypeError("value must be a list or an ObjectList") @property def _child_objects(self): diff --git a/neo/core/objectlist.py b/neo/core/objectlist.py index f3209cda5..51f256587 100644 --- a/neo/core/objectlist.py +++ b/neo/core/objectlist.py @@ -27,6 +27,12 @@ def __init__(self, allowed_contents, parent=None): self.parent = parent def _handle_append(self, obj): + self._check_allowed(obj) + if self._contains(obj): + raise ValueError("Cannot add this object because it is already contained within the list") + self._set_parent(obj) + + def _check_allowed(self, obj): if not ( isinstance(obj, self.allowed_contents) or ( # also allow proxy objects of the correct type @@ -35,9 +41,7 @@ def _handle_append(self, obj): ): raise TypeError(f"Object is a {type(obj)}. It should be one of {self.allowed_contents}.") - if self._contains(obj): - raise ValueError("Cannot add this object because it is already contained within the list") - + def _set_parent(self, obj): # set the child-parent relationship if self.parent: relationship_name = self.parent.__class__.__name__.lower() @@ -49,11 +53,8 @@ def _handle_append(self, obj): setattr(obj, relationship_name, self.parent) def _contains(self, obj): - if self._items is None: - obj_ids = [] - else: - obj_ids = [id(item) for item in self._items] - return id(obj) in obj_ids + # membership is by identity + return self._items is not None and any(item is obj for item in self._items) def __str__(self): return str(self._items) @@ -75,12 +76,10 @@ def __radd__(self, objects): return objects + self._items def __contains__(self, key): - return key in self._items + return self._contains(key) def __iadd__(self, objects): - for obj in objects: - self._handle_append(obj) - self._items.extend(objects) + self.extend(objects) return self def __iter__(self): @@ -93,25 +92,54 @@ def __len__(self): return len(self._items) def __setitem__(self, key, value): - self._items[key] = value + if isinstance(key, slice): + values = list(value) + for obj in values: + self._check_allowed(obj) + # assign to a copy first, so that nothing changes if the assignment or the checks fail + new_items = self._items[:] + new_items[key] = values + if len({id(obj) for obj in new_items}) != len(new_items): + raise ValueError("Cannot add this object because it is already contained within the list") + for obj in values: + self._set_parent(obj) + self._items = new_items + else: + if value is self._items[key]: + return + self._handle_append(value) + self._items[key] = value def append(self, obj): self._handle_append(obj) self._items.append(obj) def extend(self, objects): + # objects may be an iterator that can only be consumed once (e.g., a generator), and we need to iterate twice + objects = list(objects) + # check everything before changing anything, so that a failure leaves the list unchanged + seen = set() + for obj in objects: + self._check_allowed(obj) + if id(obj) in seen or self._contains(obj): + raise ValueError("Cannot add this object because it is already contained within the list") + seen.add(id(obj)) for obj in objects: - self._handle_append(obj) + self._set_parent(obj) self._items.extend(objects) def clear(self): self._items = [] def count(self, value): - return self._items.count(value) + return int(self._contains(value)) def index(self, value, start=0, stop=sys.maxsize): - return self._items.index(value, start, stop) + items = self._items or [] + for i in range(*slice(start, stop).indices(len(items))): + if items[i] is value: + return i + raise ValueError("object is not in list") def insert(self, index, obj): self._handle_append(obj) @@ -121,10 +149,10 @@ def pop(self, index=-1): return self._items.pop(index) def remove(self, value): - return self._items.remove(value) + del self._items[self.index(value)] def reverse(self): - raise self._items.reverse() + self._items.reverse() def sort(self, *args, key=None, reverse=False): self._items.sort(*args, key=key, reverse=reverse) diff --git a/neo/core/spiketrainlist.py b/neo/core/spiketrainlist.py index 6b9859df3..f7564f232 100644 --- a/neo/core/spiketrainlist.py +++ b/neo/core/spiketrainlist.py @@ -144,7 +144,7 @@ def __getitem__(self, i): def __setitem__(self, i, value): if self._items is None: self._spiketrains_from_array() - self._items[i] = value + super().__setitem__(i, value) def __str__(self): """Return str(self)""" @@ -232,11 +232,7 @@ def __iadd__(self, other): if isinstance(other, self.__class__): return self._add_spiketrainlists(other, in_place=True) elif other and is_spiketrain_or_proxy(other[0]): - for obj in other: - self._handle_append(obj) - if self._items is None: - self._spiketrains_from_array() - self._items.extend(other) + self.extend(other) return self else: raise TypeError("Can only add a SpikeTrainList or a list of SpikeTrains in place") @@ -281,8 +277,7 @@ def append(self, obj): raise ValueError("Can only append SpikeTrain objects") if self._items is None: self._spiketrains_from_array() - self._handle_append(obj) - self._items.append(obj) + super().append(obj) def extend(self, iterable): """Extends the SpikeTrainList with additional SpikeTrain's from an iterable @@ -306,9 +301,7 @@ def extend(self, iterable): if self._items is None: self._spiketrains_from_array() - for obj in iterable: - self._handle_append(obj) - self._items.extend(iterable) + super().extend(iterable) @classmethod def from_spike_time_array( diff --git a/neo/test/coretest/test_container.py b/neo/test/coretest/test_container.py index 3ba8b2f5c..05a278709 100644 --- a/neo/test/coretest/test_container.py +++ b/neo/test/coretest/test_container.py @@ -268,6 +268,21 @@ def test_merge__unmergable_unequal_AssertionError(self): ) +class Test_Container_set_object_list(unittest.TestCase): + def test_set_list(self): + seg = neo.core.Segment() + sig = neo.core.AnalogSignal([1, 2, 3] * pq.mV, sampling_rate=1 * pq.kHz) + seg.analogsignals = [sig] + self.assertEqual(len(seg.analogsignals), 1) + self.assertIs(sig.segment, seg) + + def test_set_other_type_TypeError(self): + seg = neo.core.Segment() + sig = neo.core.AnalogSignal([1, 2, 3] * pq.mV, sampling_rate=1 * pq.kHz) + with self.assertRaises(TypeError): + seg.analogsignals = (sig,) + + @unittest.skipUnless(HAVE_IPYTHON, "requires IPython") class Test_pprint(unittest.TestCase): def test__pretty(self): diff --git a/neo/test/coretest/test_objectlist.py b/neo/test/coretest/test_objectlist.py new file mode 100644 index 000000000..847e3df4e --- /dev/null +++ b/neo/test/coretest/test_objectlist.py @@ -0,0 +1,162 @@ +""" +Tests of the neo.core.objectlist.ObjectList class +""" + +import unittest + +import quantities as pq + +from neo.core import AnalogSignal, Event, Group, Segment +from neo.core.objectlist import ObjectList + + +def make_signal(value=1.0): + return AnalogSignal([[value], [value + 1], [value + 2]] * pq.mV, sampling_rate=1 * pq.kHz) + + +class TestObjectList(unittest.TestCase): + def setUp(self): + self.segment = Segment() + self.objlist = ObjectList(AnalogSignal, parent=self.segment) + self.a, self.b, self.c = make_signal(1.0), make_signal(10.0), make_signal(100.0) + + def test_append_sets_parent(self): + self.objlist.append(self.a) + self.assertEqual(len(self.objlist), 1) + self.assertIs(self.a.segment, self.segment) + + def test_append_duplicate(self): + self.objlist.append(self.a) + self.assertRaises(ValueError, self.objlist.append, self.a) + self.assertEqual(len(self.objlist), 1) + + def test_append_equal_but_distinct_object(self): + # membership is by identity, so an object with the same data is a different object + self.objlist.append(self.a) + self.objlist.append(make_signal(1.0)) + self.assertEqual(len(self.objlist), 2) + + def test_append_wrong_type(self): + self.assertRaises(TypeError, self.objlist.append, Event()) + + def test_extend_from_generator(self): + self.objlist.extend(sig for sig in (self.a, self.b)) + self.assertEqual(len(self.objlist), 2) + self.assertIs(self.objlist[1], self.b) + + def test_iadd_from_generator(self): + self.objlist += (sig for sig in (self.a, self.b)) + self.assertEqual(len(self.objlist), 2) + + def test_extend_duplicate_within_iterable(self): + self.assertRaises(ValueError, self.objlist.extend, [self.a, self.a]) + self.assertEqual(len(self.objlist), 0) + self.assertIsNone(self.a.segment) + + def test_extend_duplicate_of_existing_item(self): + self.objlist.append(self.a) + self.assertRaises(ValueError, self.objlist.extend, [self.b, self.a]) + self.assertEqual(len(self.objlist), 1) + self.assertIsNone(self.b.segment) + + def test_extend_is_atomic(self): + self.assertRaises(TypeError, self.objlist.extend, [self.a, Event()]) + self.assertEqual(len(self.objlist), 0) + self.assertIsNone(self.a.segment) + + def test_contains(self): + self.objlist.extend([self.a, self.b]) + self.assertIn(self.a, self.objlist) + self.assertIn(self.b, self.objlist) + self.assertNotIn(self.c, self.objlist) + self.assertNotIn(make_signal(10.0), self.objlist) + + def test_index(self): + self.objlist.extend([self.a, self.b, self.c]) + self.assertEqual(self.objlist.index(self.c), 2) + self.assertEqual(self.objlist.index(self.c, 1), 2) + self.assertEqual(self.objlist.index(self.c, -1), 2) + self.assertRaises(ValueError, self.objlist.index, self.a, 1) + self.assertRaises(ValueError, self.objlist.index, make_signal(10.0)) + + def test_count(self): + self.objlist.extend([self.a, self.b]) + self.assertEqual(self.objlist.count(self.b), 1) + self.assertEqual(self.objlist.count(make_signal(10.0)), 0) + + def test_remove(self): + self.objlist.extend([self.a, self.b, self.c]) + self.objlist.remove(self.b) + self.assertEqual(list(self.objlist), [self.a, self.c]) + self.assertRaises(ValueError, self.objlist.remove, self.b) + + def test_append_after_removal(self): + self.objlist.extend([self.a, self.b]) + self.objlist.remove(self.a) + self.objlist.append(self.a) + self.assertIs(self.objlist.pop(), self.a) + self.objlist.append(self.a) + self.objlist.clear() + self.objlist.append(self.a) + self.assertEqual(len(self.objlist), 1) + + def test_setitem(self): + self.objlist.extend([self.a, self.b]) + self.objlist[1] = self.c + self.assertIs(self.objlist[1], self.c) + self.assertIs(self.c.segment, self.segment) + + def test_setitem_same_object(self): + self.objlist.extend([self.a, self.b]) + self.objlist[1] = self.b + self.assertEqual(list(self.objlist), [self.a, self.b]) + + def test_setitem_duplicate(self): + self.objlist.extend([self.a, self.b]) + with self.assertRaises(ValueError): + self.objlist[1] = self.a + self.assertEqual(list(self.objlist), [self.a, self.b]) + + def test_setitem_wrong_type(self): + self.objlist.append(self.a) + with self.assertRaises(TypeError): + self.objlist[0] = Event() + + def test_setitem_slice(self): + self.objlist.extend([self.a, self.b]) + self.objlist[1:] = [self.b, self.c] + self.assertEqual(list(self.objlist), [self.a, self.b, self.c]) + self.assertIs(self.c.segment, self.segment) + + def test_setitem_slice_duplicate(self): + self.objlist.extend([self.a, self.b]) + with self.assertRaises(ValueError): + self.objlist[1:] = [self.c, self.a] + with self.assertRaises(ValueError): + self.objlist[1:] = [self.c, self.c] + self.assertEqual(list(self.objlist), [self.a, self.b]) + self.assertIsNone(self.c.segment) + + def test_setitem_slice_wrong_type(self): + self.objlist.append(self.a) + with self.assertRaises(TypeError): + self.objlist[:] = [self.b, Event()] + self.assertEqual(list(self.objlist), [self.a]) + + def test_reverse(self): + self.objlist.extend([self.a, self.b, self.c]) + self.objlist.reverse() + self.assertEqual(list(self.objlist), [self.c, self.b, self.a]) + + def test_without_parent(self): + group = Group() + objlist = ObjectList(AnalogSignal) + objlist.extend([self.a, self.b]) + self.assertEqual(len(objlist), 2) + self.assertIsNone(self.a.segment) + group.analogsignals.append(self.c) + self.assertIn(self.c, group.analogsignals) + + +if __name__ == "__main__": + unittest.main() diff --git a/neo/test/coretest/test_spiketrainlist.py b/neo/test/coretest/test_spiketrainlist.py index b1a9d517d..980530852 100644 --- a/neo/test/coretest/test_spiketrainlist.py +++ b/neo/test/coretest/test_spiketrainlist.py @@ -268,6 +268,30 @@ def test_append_something_else(self): for stl in (self.stl_from_obj_list, self.stl_from_array): self.assertRaises(ValueError, stl.append, None) + def test_append_to_multiplexed(self): + """Appending to a SpikeTrainList created from an array should convert it to a list""" + new_st = SpikeTrain([22.2, 33.3], units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) + self.stl_from_array.append(new_st) + self.assertEqual(len(self.stl_from_array), 5) + self.assertIn(new_st, self.stl_from_array) + self.assertIs(self.stl_from_array[-1], new_st) + + def test_extend_from_generator(self): + """Extending from a generator should add all the SpikeTrains""" + for stl in (self.stl_from_obj_list, self.stl_from_array): + stl.extend( + SpikeTrain(times, units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) for times in ([22.2], [44.4]) + ) + self.assertEqual(len(stl), 6) + + def test_extend_duplicate(self): + """Extending with an iterable containing the same SpikeTrain twice should fail""" + new_st = SpikeTrain([22.2, 33.3], units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) + self.assertRaises(ValueError, self.stl_from_obj_list.extend, [new_st, new_st]) + self.assertEqual(len(self.stl_from_obj_list), 4) + self.assertRaises(ValueError, self.stl_from_obj_list.extend, [self.stl_from_obj_list[0]]) + self.assertEqual(len(self.stl_from_obj_list), 4) + def test_multiplexed(self): """The multiplexed property should return a pair of arrays""" channel_id_array, spike_time_array = self.stl_from_array.multiplexed From 533d7c26e770502570d5fa2b7c063d9d7549a9ce Mon Sep 17 00:00:00 2001 From: Andrew Davison Date: Thu, 29 Oct 2026 10:44:08 +0100 Subject: [PATCH 2/3] Check for duplicates in all SpikeTrainList paths that add spike trains SpikeTrainList's constructor, the in-place `stl += other_stl` path and the list branch of __radd__ add spike trains without the duplicate check, so the same SpikeTrain can be stored twice, after which count() and the list disagree. These paths now go through extend(), or in the case of the constructor perform the same check, which also stops the constructor from silently dropping the contents of a generator. See #1935. --- neo/core/spiketrainlist.py | 13 ++++++------- neo/test/coretest/test_spiketrainlist.py | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/neo/core/spiketrainlist.py b/neo/core/spiketrainlist.py index f7564f232..887b028c4 100644 --- a/neo/core/spiketrainlist.py +++ b/neo/core/spiketrainlist.py @@ -105,10 +105,13 @@ def __init__(self, items=None, parent=None): if items is None: self._items = items else: + items = list(items) for item in items: if not is_spiketrain_or_proxy(item): raise ValueError("`items` can only contain SpikeTrain objects or proxy objects") - self._items = list(items) + if len({id(item) for item in items}) != len(items): + raise ValueError("`items` contains the same object more than once") + self._items = items self._spike_time_array = None self._channel_id_array = None self._all_channel_ids = None @@ -183,7 +186,7 @@ def _add_spiketrainlists(self, other, in_place=False): if other._items is None: other._spiketrains_from_array() if in_place: - self._items.extend(other._items) + self.extend(other._items) return self else: return self.__class__(items=self._items[:] + other._items) @@ -242,11 +245,7 @@ def __radd__(self, other): if isinstance(other, self.__class__): return other._add_spiketrainlists(self) elif other and is_spiketrain_or_proxy(other[0]): - for obj in other: - obj.segment = self.segment - if self._items is None: - self._spiketrains_from_array() - self._items.extend(other) + self.extend(other) return self elif len(other) == 0: return self diff --git a/neo/test/coretest/test_spiketrainlist.py b/neo/test/coretest/test_spiketrainlist.py index 980530852..e3202a17c 100644 --- a/neo/test/coretest/test_spiketrainlist.py +++ b/neo/test/coretest/test_spiketrainlist.py @@ -292,6 +292,26 @@ def test_extend_duplicate(self): self.assertRaises(ValueError, self.stl_from_obj_list.extend, [self.stl_from_obj_list[0]]) self.assertEqual(len(self.stl_from_obj_list), 4) + def test_create_with_duplicate(self): + """Creating a SpikeTrainList containing the same SpikeTrain twice should fail""" + st = SpikeTrain([22.2, 33.3], units="ms", t_start=0 * pq.ms, t_stop=100.0 * pq.ms) + self.assertRaises(ValueError, SpikeTrainList, items=[st, st]) + + def test_create_from_generator(self): + stl = SpikeTrainList(items=(st for st in self.stl_from_obj_list)) + self.assertEqual(len(stl), 4) + + def test_add_duplicate(self): + """Combining SpikeTrainLists that share a SpikeTrain should fail, whichever operator is used""" + st = self.stl_from_obj_list[0] + self.assertRaises(ValueError, lambda: self.stl_from_obj_list + SpikeTrainList(items=[st])) + self.assertRaises(ValueError, lambda: self.stl_from_obj_list + [st]) + self.assertRaises(ValueError, lambda: [st] + self.stl_from_obj_list) + with self.assertRaises(ValueError): + self.stl_from_obj_list += SpikeTrainList(items=[st]) + self.assertEqual(len(self.stl_from_obj_list), 4) + self.assertEqual(self.stl_from_obj_list.count(st), 1) + def test_multiplexed(self): """The multiplexed property should return a pair of arrays""" channel_id_array, spike_time_array = self.stl_from_array.multiplexed From 66984c3805d31ad33e76718aef20c8c6a4cf37c5 Mon Sep 17 00:00:00 2001 From: Andrew Davison Date: Thu, 29 Oct 2026 10:48:05 +0100 Subject: [PATCH 3/3] Make ObjectList.extend() O(n + m) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit extend() checks each incoming object against the existing items with _contains(), which scans the whole list, so extending a list of n objects by m objects is O(n·m). Since the previous commit routes `stl += other_stl` through extend(), this makes combining two lists of 10,000 SpikeTrains take over a second, where it previously took under a millisecond. extend() now builds the set of existing ids once per call. --- neo/core/objectlist.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/neo/core/objectlist.py b/neo/core/objectlist.py index 51f256587..2169ab936 100644 --- a/neo/core/objectlist.py +++ b/neo/core/objectlist.py @@ -117,11 +117,12 @@ def append(self, obj): def extend(self, objects): # objects may be an iterator that can only be consumed once (e.g., a generator), and we need to iterate twice objects = list(objects) - # check everything before changing anything, so that a failure leaves the list unchanged - seen = set() + # check everything before changing anything, so that a failure leaves the list unchanged. + # Building the set of ids once, rather than calling _contains() for each object, keeps this O(n + m) + seen = {id(item) for item in self._items or ()} for obj in objects: self._check_allowed(obj) - if id(obj) in seen or self._contains(obj): + if id(obj) in seen: raise ValueError("Cannot add this object because it is already contained within the list") seen.add(id(obj)) for obj in objects: