+ def test_iterator_pickling(self):
+ class SeqOf(SequenceOf):
+ schema = Integer()
+ register_class(SeqOf)
+ seqof = SeqOf()
+ pickle_dumps(seqof)
+ seqof = seqof(iter(range(10)))
+ with self.assertRaisesRegex(ValueError, "iterator"):
+ pickle_dumps(seqof)
+
+ def test_iterator_bounds(self):
+ class SeqOf(SequenceOf):
+ schema = Integer()
+ bounds = (10, 20)
+ seqof = None
+
+ def gen(n):
+ for i in range(n):
+ yield Integer(i)
+ for n in (9, 21):
+ seqof = SeqOf(gen(n))
+ self.assertTrue(seqof.ready)
+ with self.assertRaises(BoundsError):
+ seqof.encode()
+ self.assertFalse(seqof.ready)
+ seqof = seqof(gen(n))
+ self.assertTrue(seqof.ready)
+ with self.assertRaises(BoundsError):
+ encode_cer(seqof)
+ self.assertFalse(seqof.ready)
+
+ def test_iterator_twice(self):
+ class SeqOf(SequenceOf):
+ schema = Integer()
+ bounds = (1, float("+inf"))
+
+ def gen():
+ for i in range(10):
+ yield Integer(i)
+ seqof = SeqOf(gen())
+ self.assertTrue(seqof.ready)
+ seqof.encode()
+ self.assertFalse(seqof.ready)
+ register_class(SeqOf)
+ pickle_dumps(seqof)
+
+ def test_iterator_2pass(self):
+ class SeqOf(SequenceOf):
+ schema = Integer()
+ bounds = (1, float("+inf"))
+
+ def gen():
+ for i in range(10):
+ yield Integer(i)
+ seqof = SeqOf(gen())
+ self.assertTrue(seqof.ready)
+ _, state = seqof.encode1st()
+ self.assertFalse(seqof.ready)
+ seqof = seqof(gen())
+ self.assertTrue(seqof.ready)
+ buf = BytesIO()
+ seqof.encode2nd(buf.write, iter(state))
+ self.assertSequenceEqual(
+ [int(i) for i in seqof.decod(buf.getvalue())],
+ list(gen()),
+ )