forked from petertodd/python-bitcoinlib
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_messages.py
More file actions
137 lines (93 loc) · 4.12 KB
/
Copy pathtest_messages.py
File metadata and controls
137 lines (93 loc) · 4.12 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
# Copyright (C) 2013-2014 The python-bitcoinlib developers
#
# This file is part of python-bitcoinlib.
#
# It is subject to the license terms in the LICENSE file found in the top-level
# directory of this distribution.
#
# No part of python-bitcoinlib, including this file, may be copied, modified,
# propagated, or distributed except according to the terms contained in the
# LICENSE file.
import unittest
from bitcoin.messages import msg_version, msg_verack, msg_addr, msg_alert, \
msg_inv, msg_getdata, msg_getblocks, msg_getheaders, msg_headers, msg_tx, \
msg_block, msg_getaddr, msg_ping, msg_pong, msg_mempool, MsgSerializable, \
msg_notfound, msg_reject
import sys
if sys.version > '3':
from io import BytesIO
else:
from cStringIO import StringIO as BytesIO
class MessageTestCase(unittest.TestCase):
def serialization_test(self, cls):
m = cls()
mSerialized = m.to_bytes()
mDeserialzed = cls.from_bytes(mSerialized)
mSerialzedTwice = mDeserialzed.to_bytes()
self.assertEqual(mSerialized, mSerialzedTwice)
class Test_msg_version(MessageTestCase):
def test_serialization(self):
super(Test_msg_version, self).serialization_test(msg_version)
class Test_msg_verack(MessageTestCase):
def test_serialization(self):
super(Test_msg_verack, self).serialization_test(msg_verack)
class Test_msg_addr(MessageTestCase):
def test_serialization(self):
super(Test_msg_addr, self).serialization_test(msg_addr)
class Test_msg_alert(MessageTestCase):
def test_serialization(self):
super(Test_msg_alert, self).serialization_test(msg_alert)
class Test_msg_inv(MessageTestCase):
def test_serialization(self):
super(Test_msg_inv, self).serialization_test(msg_inv)
class Test_msg_getdata(MessageTestCase):
def test_serialization(self):
super(Test_msg_getdata, self).serialization_test(msg_getdata)
class Test_msg_getblocks(MessageTestCase):
def test_serialization(self):
super(Test_msg_getblocks, self).serialization_test(msg_getblocks)
class Test_msg_notfound(MessageTestCase):
def test_serialization(self):
super(Test_msg_notfound, self).serialization_test(msg_notfound)
class Test_msg_getheaders(MessageTestCase):
def test_serialization(self):
super(Test_msg_getheaders, self).serialization_test(msg_getheaders)
class Test_msg_headers(MessageTestCase):
def test_serialization(self):
super(Test_msg_headers, self).serialization_test(msg_headers)
class Test_msg_tx(MessageTestCase):
def test_serialization(self):
super(Test_msg_tx, self).serialization_test(msg_tx)
class Test_msg_block(MessageTestCase):
def test_serialization(self):
super(Test_msg_block, self).serialization_test(msg_block)
class Test_msg_getaddr(MessageTestCase):
def test_serialization(self):
super(Test_msg_getaddr, self).serialization_test(msg_getaddr)
class Test_msg_ping(MessageTestCase):
def test_serialization(self):
super(Test_msg_ping, self).serialization_test(msg_ping)
class Test_msg_pong(MessageTestCase):
def test_serialization(self):
super(Test_msg_pong, self).serialization_test(msg_pong)
class Test_msg_reject(MessageTestCase):
def test_serialization(self):
super(Test_msg_reject, self).serialization_test(msg_reject)
class Test_msg_mempool(MessageTestCase):
def test_serialization(self):
super(Test_msg_mempool, self).serialization_test(msg_mempool)
class Test_messages(unittest.TestCase):
verackbytes = b'\xf9\xbe\xb4\xd9verack\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00]\xf6\xe0\xe2'
def test_read_msg_verack(self):
f = BytesIO(self.verackbytes)
m = MsgSerializable.stream_deserialize(f)
self.assertEqual(m.command, msg_verack.command)
def test_fail_invalid_message(self):
bad_verack_bytes = b'\xf8' + self.verackbytes[1:]
f = BytesIO(bad_verack_bytes)
with self.assertRaises(ValueError):
MsgSerializable.stream_deserialize(f)
def test_msg_verack_to_bytes(self):
m = msg_verack()
b = m.to_bytes()
self.assertEqual(self.verackbytes, b)