Fix multiple bugs in ObjectList

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):

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)

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(

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):

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()

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