← Back to branch summary

~ubuntu-branches/debian/sid/pyrlp/sid

« back to all changes in this revision

Viewing changes to tests/test_benchmark.py

  • Committer: Package Import Robot
  • Author(s): Ben Finney
  • Date: 2017-07-15 05:25:42 UTC
  • Revision ID: package-import@ubuntu.com-20170715052542-a20zzsypt1qfecq1
Tags: upstream-0.5.1
Import upstream version 0.5.1

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_benchmark.py
 
1
from __future__ import unicode_literals
 
2
 
 
3
from itertools import repeat, chain
 
4
import sys
 
5
import pytest
 
6
import rlp
 
7
from rlp.sedes import binary, CountableList
 
8
from rlp.exceptions import DecodingError, DeserializationError
 
9
 
 
10
 
 
11
try:
 
12
    import pytest_benchmark
 
13
except ImportError:
 
14
    do_benchmark = False
 
15
else:
 
16
    do_benchmark = True
 
17
 
 
18
 
 
19
# speed up setup in case tests aren't run anyway
 
20
if do_benchmark:
 
21
    SIZE = int(1e6)
 
22
else:
 
23
    SIZE = 1
 
24
 
 
25
 
 
26
class Message(rlp.Serializable):
 
27
 
 
28
    fields = [
 
29
        ('field1', binary),
 
30
        ('field2', binary),
 
31
        ('field3', CountableList(binary, max_length=100))
 
32
    ]
 
33
 
 
34
 
 
35
def lazy_test_factory(s, valid):
 
36
    @pytest.mark.benchmark(group='lazy')
 
37
    def f(benchmark):
 
38
        @benchmark
 
39
        def result():
 
40
            try:
 
41
                Message.deserialize(rlp.decode_lazy(s))
 
42
            except (DecodingError, DeserializationError):
 
43
                return not valid
 
44
            else:
 
45
                return valid
 
46
        assert result
 
47
    return f
 
48
 
 
49
 
 
50
def eager_test_factory(s, valid):
 
51
    @pytest.mark.benchmark(group='eager')
 
52
    def f(benchmark):
 
53
        @benchmark
 
54
        def result():
 
55
            try:
 
56
                rlp.decode(s, Message)
 
57
            except (DecodingError, DeserializationError):
 
58
                return not valid
 
59
            else:
 
60
                return valid
 
61
        assert result
 
62
    return f
 
63
 
 
64
 
 
65
def generate_test_functions():
 
66
    valid = {}
 
67
    invalid = {}
 
68
    long_string = bytes(bytearray((i % 256 for i in range(SIZE))))
 
69
    long_list = rlp.encode([c for c in long_string])
 
70
    invalid['long_string'] = long_string
 
71
    invalid['long_list'] = long_list
 
72
 
 
73
    nested_list = rlp.encode('\x00')
 
74
    for _ in repeat(None, SIZE):
 
75
        nested_list += rlp.codec.length_prefix(len(nested_list), 0xc0)
 
76
    invalid['nested_list'] = nested_list
 
77
 
 
78
    valid['long_string_object'] = rlp.encode([b'\x00', long_string, []])
 
79
 
 
80
    prefix = rlp.codec.length_prefix(1 + 1 + len(long_list), 0xc0)
 
81
    invalid['long_list_object'] = prefix + rlp.encode(b'\x00') + rlp.encode(b'\x00') + long_list
 
82
 
 
83
    valid['friendly'] = rlp.encode(Message('hello', 'I\'m friendly', ['not', 'many', 'elements']))
 
84
 
 
85
    invalid = invalid.items()
 
86
    valid = valid.items()
 
87
    rlp_strings = [i[1] for i in chain(valid, invalid)]
 
88
    valids = [True] * len(valid) + [False] * len(invalid)
 
89
    names = [i[0] for i in chain(valid, invalid)]
 
90
 
 
91
    current_module = sys.modules[__name__]
 
92
    for rlp_string, valid, name in zip(rlp_strings, valids, names):
 
93
        f_eager = pytest.mark.skipif('not do_benchmark')(eager_test_factory(rlp_string, valid))
 
94
        f_lazy = pytest.mark.skipif('not do_benchmark')(lazy_test_factory(rlp_string, valid))
 
95
        setattr(current_module, 'test_eager_' + name, f_eager)
 
96
        setattr(current_module, 'test_lazy_' + name, f_lazy)
 
97
 
 
98
 
 
99
generate_test_functions()