PEP8

[pyderasn.git] / tests / test_pyderasn.py

diff --git a/tests/test_pyderasn.py b/tests/test_pyderasn.py
index 3a9e3036f0d1490fe930870d950b2bdf5d50a4e4..d112a60c9a2ae7302f8dbf83afaeb5537a47e159 100644 (file)
--- a/tests/test_pyderasn.py
+++ b/tests/test_pyderasn.py
@@ -1,5 +1,5 @@
 # coding: utf-8
-# PyDERASN -- Python ASN.1 DER/BER codec with abstract structures
+# PyDERASN -- Python ASN.1 DER/CER/BER codec with abstract structures
 # Copyright (C) 2017-2020 Sergey Matveev <stargrave@stargrave.org>
 #
 # This program is free software: you can redistribute it and/or modify
@@ -20,6 +20,7 @@ from copy import deepcopy
 from datetime import datetime
 from datetime import timedelta
 from importlib import import_module
+from io import BytesIO
 from operator import attrgetter
 from os import environ
 from os import urandom
@@ -75,6 +76,7 @@ from pyderasn import BoundsError
 from pyderasn import Choice
 from pyderasn import DecodeError
 from pyderasn import DecodePathDefBy
+from pyderasn import encode2pass
 from pyderasn import encode_cer
 from pyderasn import Enumerated
 from pyderasn import EOC
@@ -422,6 +424,8 @@ class TestBoolean(CommonMixin, TestCase):
         pprint(obj, big_blobs=True, with_decode_path=True)
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         repr(err.exception)
         obj = Boolean(value)
         self.assertTrue(obj.ready)
@@ -506,6 +510,8 @@ class TestBoolean(CommonMixin, TestCase):
         obj = Boolean(value, impl=tag_impl)
         with self.assertRaises(NotEnoughData):
             obj.decode(obj.encode()[:-1])
+        with self.assertRaises(NotEnoughData):
+            obj.decode(encode2pass(obj)[:-1])
 
     @given(
         booleans(),
@@ -515,6 +521,8 @@ class TestBoolean(CommonMixin, TestCase):
         obj = Boolean(value, expl=tag_expl)
         with self.assertRaises(NotEnoughData):
             obj.decode(obj.encode()[:-1])
+        with self.assertRaises(NotEnoughData):
+            obj.decode(encode2pass(obj)[:-1])
 
     @given(
         integers(min_value=31),
@@ -603,6 +611,7 @@ class TestBoolean(CommonMixin, TestCase):
             pprint(obj, big_blobs=True, with_decode_path=True)
             self.assertFalse(obj.expled)
             obj_encoded = obj.encode()
+            self.assertEqual(encode2pass(obj), obj_encoded)
             self.assertSequenceEqual(encode_cer(obj), obj_encoded)
             obj_expled = obj(value, expl=tag_expl)
             self.assertTrue(obj_expled.expled)
@@ -863,6 +872,8 @@ class TestInteger(CommonMixin, TestCase):
         pprint(obj, big_blobs=True, with_decode_path=True)
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         repr(err.exception)
         obj = Integer(value)
         self.assertTrue(obj.ready)
@@ -925,6 +936,10 @@ class TestInteger(CommonMixin, TestCase):
                 Integer(values[0]).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            Integer(bounds=(values[1], values[2])).decode(
+                encode2pass(Integer(values[0]))
+            )
         with self.assertRaises(BoundsError) as err:
             Integer(value=values[2], bounds=(values[0], values[1]))
         repr(err.exception)
@@ -933,6 +948,10 @@ class TestInteger(CommonMixin, TestCase):
                 Integer(values[2]).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            Integer(bounds=(values[0], values[1])).decode(
+                encode2pass(Integer(values[2]))
+            )
 
     @given(data_strategy())
     def test_call(self, d):
@@ -1124,6 +1143,7 @@ class TestInteger(CommonMixin, TestCase):
             pprint(obj, big_blobs=True, with_decode_path=True)
             self.assertFalse(obj.expled)
             obj_encoded = obj.encode()
+            self.assertEqual(encode2pass(obj), obj_encoded)
             self.assertSequenceEqual(encode_cer(obj), obj_encoded)
             obj_expled = obj(value, expl=tag_expl)
             self.assertTrue(obj_expled.expled)
@@ -1362,6 +1382,8 @@ class TestBitString(CommonMixin, TestCase):
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         obj = BitString(value)
         self.assertTrue(obj.ready)
         repr(obj)
@@ -1540,6 +1562,7 @@ class TestBitString(CommonMixin, TestCase):
             pprint(obj, big_blobs=True, with_decode_path=True)
             self.assertFalse(obj.expled)
             obj_encoded = obj.encode()
+            self.assertEqual(encode2pass(obj), obj_encoded)
             self.assertSequenceEqual(encode_cer(obj), obj_encoded)
             obj_expled = obj(value, expl=tag_expl)
             self.assertTrue(obj_expled.expled)
@@ -1965,6 +1988,8 @@ class TestOctetString(CommonMixin, TestCase):
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         obj = OctetString(value)
         self.assertTrue(obj.ready)
         repr(obj)
@@ -2011,6 +2036,10 @@ class TestOctetString(CommonMixin, TestCase):
                 OctetString(value).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            OctetString(bounds=(bound_min, bound_max)).decode(
+                encode2pass(OctetString(value))
+            )
         value = d.draw(binary(min_size=bound_max + 1))
         with self.assertRaises(BoundsError) as err:
             OctetString(value=value, bounds=(bound_min, bound_max))
@@ -2020,6 +2049,10 @@ class TestOctetString(CommonMixin, TestCase):
                 OctetString(value).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            OctetString(bounds=(bound_min, bound_max)).decode(
+                encode2pass(OctetString(value))
+            )
 
     @given(data_strategy())
     def test_call(self, d):
@@ -2196,6 +2229,7 @@ class TestOctetString(CommonMixin, TestCase):
             pprint(obj, big_blobs=True, with_decode_path=True)
             self.assertFalse(obj.expled)
             obj_encoded = obj.encode()
+            self.assertEqual(encode2pass(obj), obj_encoded)
             self.assertSequenceEqual(encode_cer(obj), obj_encoded)
             obj_expled = obj(value, expl=tag_expl)
             self.assertTrue(obj_expled.expled)
@@ -2566,6 +2600,7 @@ class TestNull(CommonMixin, TestCase):
             pprint(obj, big_blobs=True, with_decode_path=True)
             self.assertFalse(obj.expled)
             obj_encoded = obj.encode()
+            self.assertEqual(encode2pass(obj), obj_encoded)
             self.assertSequenceEqual(encode_cer(obj), obj_encoded)
             obj_expled = obj(expl=tag_expl)
             self.assertTrue(obj_expled.expled)
@@ -2627,7 +2662,6 @@ class TestNull(CommonMixin, TestCase):
             repr(obj)
             list(obj.pps())
 
-
     @given(integers(min_value=1))
     def test_invalid_len(self, l):
         with self.assertRaises(InvalidLength):
@@ -2695,6 +2729,8 @@ class TestObjectIdentifier(CommonMixin, TestCase):
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         obj = ObjectIdentifier(value)
         self.assertTrue(obj.ready)
         self.assertFalse(obj.ber_encoded)
@@ -2932,6 +2968,7 @@ class TestObjectIdentifier(CommonMixin, TestCase):
             pprint(obj, big_blobs=True, with_decode_path=True)
             self.assertFalse(obj.expled)
             obj_encoded = obj.encode()
+            self.assertEqual(encode2pass(obj), obj_encoded)
             self.assertSequenceEqual(encode_cer(obj), obj_encoded)
             obj_expled = obj(value, expl=tag_expl)
             self.assertTrue(obj_expled.expled)
@@ -3327,6 +3364,7 @@ class TestEnumerated(CommonMixin, TestCase):
         pprint(obj, big_blobs=True, with_decode_path=True)
         self.assertFalse(obj.expled)
         obj_encoded = obj.encode()
+        self.assertEqual(encode2pass(obj), obj_encoded)
         obj_expled = obj(value, expl=tag_expl)
         self.assertTrue(obj_expled.expled)
         repr(obj_expled)
@@ -3444,6 +3482,8 @@ class StringMixin(object):
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         value = d.draw(text(alphabet=self.text_alphabet()))
         obj = self.base_klass(value)
         self.assertTrue(obj.ready)
@@ -3493,6 +3533,10 @@ class StringMixin(object):
                 self.base_klass(value).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            self.base_klass(bounds=(bound_min, bound_max)).decode(
+                encode2pass(self.base_klass(value))
+            )
         value = d.draw(text(alphabet=self.text_alphabet(), min_size=bound_max + 1))
         with self.assertRaises(BoundsError) as err:
             self.base_klass(value=value, bounds=(bound_min, bound_max))
@@ -3502,6 +3546,10 @@ class StringMixin(object):
                 self.base_klass(value).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            self.base_klass(bounds=(bound_min, bound_max)).decode(
+                encode2pass(self.base_klass(value))
+            )
 
     @given(data_strategy())
     def test_call(self, d):
@@ -3677,6 +3725,7 @@ class StringMixin(object):
         pprint(obj, big_blobs=True, with_decode_path=True)
         self.assertFalse(obj.expled)
         obj_encoded = obj.encode()
+        self.assertEqual(encode2pass(obj), obj_encoded)
         obj_expled = obj(value, expl=tag_expl)
         self.assertTrue(obj_expled.expled)
         repr(obj_expled)
@@ -4033,6 +4082,8 @@ class TimeMixin(object):
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         value = d.draw(datetimes(
             min_value=self.min_datetime,
             max_value=self.max_datetime,
@@ -4191,6 +4242,7 @@ class TimeMixin(object):
         pprint(obj, big_blobs=True, with_decode_path=True)
         self.assertFalse(obj.expled)
         obj_encoded = obj.encode()
+        self.assertEqual(encode2pass(obj), obj_encoded)
         self.additional_symmetric_check(value, obj_encoded)
         obj_expled = obj(value, expl=tag_expl)
         self.assertTrue(obj_expled.expled)
@@ -5000,6 +5052,8 @@ class TestAny(CommonMixin, TestCase):
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         obj = Any(value)
         self.assertTrue(obj.ready)
         repr(obj)
@@ -5148,6 +5202,7 @@ class TestAny(CommonMixin, TestCase):
             tag_class, _, tag_num = tag_decode(tag_strip(value)[0])
             self.assertEqual(obj.tag_order, (tag_class, tag_num))
             obj_encoded = obj.encode()
+            self.assertEqual(encode2pass(obj), obj_encoded)
             obj_expled = obj(value, expl=tag_expl)
             self.assertTrue(obj_expled.expled)
             tag_class, _, tag_num = tag_decode(tag_expl)
@@ -5384,6 +5439,8 @@ class TestChoice(CommonMixin, TestCase):
         with self.assertRaises(ObjNotReady) as err:
             obj.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(obj)
         obj["whatever"] = Boolean()
         self.assertFalse(obj.ready)
         repr(obj)
@@ -5532,6 +5589,7 @@ class TestChoice(CommonMixin, TestCase):
         self.assertFalse(obj.expled)
         self.assertEqual(obj.tag_order, obj.value.tag_order)
         obj_encoded = obj.encode()
+        self.assertEqual(encode2pass(obj), obj_encoded)
         obj_expled = obj(value, expl=tag_expl)
         self.assertTrue(obj_expled.expled)
         tag_class, _, tag_num = tag_decode(tag_expl)
@@ -5879,6 +5937,8 @@ class SeqMixing(object):
         with self.assertRaises(ObjNotReady) as err:
             seq.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(seq)
         for name, value in non_ready.items():
             seq[name] = Boolean(value)
         self.assertTrue(seq.ready)
@@ -6067,6 +6127,7 @@ class SeqMixing(object):
         pprint(seq, big_blobs=True, with_decode_path=True)
         self.assertTrue(seq.ready)
         seq_encoded = seq.encode()
+        self.assertEqual(encode2pass(seq), seq_encoded)
         seq_encoded_cer = encode_cer(seq)
         self.assertNotEqual(seq_encoded_cer, seq_encoded)
         self.assertSequenceEqual(
@@ -6155,6 +6216,7 @@ class SeqMixing(object):
         seq, expect_outers = d.draw(sequences_strategy(seq_klass=self.base_klass))
         self.assertTrue(seq.ready)
         seq_encoded = seq.encode()
+        self.assertEqual(encode2pass(seq), seq_encoded)
         seq_decoded, tail = seq.decode(seq_encoded)
         self.assertEqual(tail, b"")
         self.assertTrue(seq.ready)
@@ -6212,24 +6274,34 @@ class SeqMixing(object):
             max_size=len(_schema),
         ))]
 
+        class Wahl(Choice):
+            schema = (("int", Integer()),)
+
         class SeqWithoutDefault(self.base_klass):
             schema = [
-                (n, Integer(impl=t))
+                (n, Wahl(expl=t))
                 for (n, _), t in zip(_schema, tags)
             ]
         seq_without_default = SeqWithoutDefault()
         for name, value in _schema:
-            seq_without_default[name] = Integer(value)
+            seq_without_default[name] = Wahl(("int", Integer(value)))
         seq_encoded = seq_without_default.encode()
+        seq_without_default.decode(seq_encoded)
+        self.assertEqual(
+            len(list(seq_without_default.decode_evgen(seq_encoded))),
+            len(_schema) * 2 + 1,
+        )
 
         class SeqWithDefault(self.base_klass):
             schema = [
-                (n, Integer(default=v, impl=t))
+                (n, Wahl(default=Wahl(("int", Integer(v))), expl=t))
                 for (n, v), t in zip(_schema, tags)
             ]
         seq_with_default = SeqWithDefault()
         with assertRaisesRegex(self, DecodeError, "DEFAULT value met"):
             seq_with_default.decode(seq_encoded)
+        with assertRaisesRegex(self, DecodeError, "DEFAULT value met"):
+            list(seq_with_default.decode_evgen(seq_encoded))
         for ctx in ({"bered": True}, {"allow_default_values": True}):
             seq_decoded, _ = seq_with_default.decode(seq_encoded, ctx=ctx)
             self.assertTrue(seq_decoded.ber_encoded)
@@ -6239,7 +6311,21 @@ class SeqMixing(object):
             self.assertTrue(seq_decoded.bered)
             for name, value in _schema:
                 self.assertEqual(seq_decoded[name], seq_with_default[name])
-                self.assertEqual(seq_decoded[name], value)
+                self.assertEqual(seq_decoded[name].value, value)
+            self.assertEqual(
+                len(list(seq_with_default.decode_evgen(seq_encoded, ctx=ctx))),
+                len(_schema) + 1,
+            )
+
+        seq_without_default = SeqWithoutDefault()
+        for name, value in _schema:
+            seq_without_default[name] = Wahl(("int", Integer(value + 1)))
+        seq_encoded = seq_without_default.encode()
+        seq_with_default.decode(seq_encoded)
+        self.assertEqual(
+            len(list(seq_with_default.decode_evgen(seq_encoded))),
+            len(_schema) + 1,
+        )
 
     @given(data_strategy())
     def test_missing_from_spec(self, d):
@@ -6264,6 +6350,8 @@ class SeqMixing(object):
         seq_missing = SeqMissing()
         with self.assertRaises(TagMismatch):
             seq_missing.decode(seq_encoded)
+        with self.assertRaises(TagMismatch):
+            list(seq_missing.decode_evgen(seq_encoded))
 
     def test_bered(self):
         class Seq(self.base_klass):
@@ -6290,6 +6378,9 @@ class SeqMixing(object):
         encoded = Seq.tag_default + len_encode(len(encoded)) + encoded
         with self.assertRaises(DecodeError):
             Seq().decode(encoded)
+        with self.assertRaises(DecodeError):
+            list(Seq().decode_evgen(encoded))
+        list(Seq().decode_evgen(encoded, ctx={"bered": True}))
         decoded, _ = Seq().decode(encoded, ctx={"bered": True})
         self.assertFalse(decoded.ber_encoded)
         self.assertFalse(decoded.lenindef)
@@ -6527,6 +6618,8 @@ class SeqOfMixing(object):
         with self.assertRaises(ObjNotReady) as err:
             seqof.encode()
         repr(err.exception)
+        with self.assertRaises(ObjNotReady) as err:
+            encode2pass(seqof)
         for i, value in enumerate(values):
             self.assertEqual(seqof[i], value)
             if not seqof[i].ready:
@@ -6570,6 +6663,10 @@ class SeqOfMixing(object):
                 SeqOf(value).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            SeqOf(bounds=(bound_min, bound_max)).decode(
+                encode2pass(SeqOf(value))
+            )
         value = [Boolean(True)] * d.draw(integers(
             min_value=bound_max + 1,
             max_value=bound_max + 10,
@@ -6582,6 +6679,10 @@ class SeqOfMixing(object):
                 SeqOf(value).encode()
             )
         repr(err.exception)
+        with assertRaisesRegex(self, DecodeError, "bounds") as err:
+            SeqOf(bounds=(bound_min, bound_max)).decode(
+                encode2pass(SeqOf(value))
+            )
 
     @given(integers(min_value=1, max_value=10))
     def test_out_of_bounds(self, bound_max):
@@ -6788,6 +6889,7 @@ class SeqOfMixing(object):
         pprint(obj, big_blobs=True, with_decode_path=True)
         self.assertFalse(obj.expled)
         obj_encoded = obj.encode()
+        self.assertEqual(encode2pass(obj), obj_encoded)
         obj_encoded_cer = encode_cer(obj)
         self.assertNotEqual(obj_encoded_cer, obj_encoded)
         self.assertSequenceEqual(
@@ -6944,6 +7046,7 @@ class TestSequenceOf(SeqOfMixing, CommonMixin, TestCase):
             schema = Integer()
             bounds = (10, 20)
         seqof = None
+
         def gen(n):
             for i in six_xrange(n):
                 yield Integer(i)
@@ -6963,6 +7066,7 @@ class TestSequenceOf(SeqOfMixing, CommonMixin, TestCase):
         class SeqOf(SequenceOf):
             schema = Integer()
             bounds = (1, float("+inf"))
+
         def gen():
             for i in six_xrange(10):
                 yield Integer(i)
@@ -6973,6 +7077,27 @@ class TestSequenceOf(SeqOfMixing, CommonMixin, TestCase):
         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 six_xrange(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()),
+        )
+
     def test_non_ready_bound_min(self):
         class SeqOf(SequenceOf):
             schema = Integer()
@@ -7491,6 +7616,7 @@ class TestDefinesByPath(TestCase):
 
     def test_remaining_data(self):
         oid = ObjectIdentifier("1.2.3")
+
         class Seq(Sequence):
             schema = (
                 ("oid", ObjectIdentifier(defines=((("tgt",), {
@@ -7508,6 +7634,7 @@ class TestDefinesByPath(TestCase):
 
     def test_remaining_data_seqof(self):
         oid = ObjectIdentifier("1.2.3")
+
         class SeqOf(SetOf):
             schema = OctetString()