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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright 2022 Daniel Estevez <daniel@destevez.net>
#
# This file is part of gr-satellites
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
from gnuradio import gr, blocks, gr_unittest
import numpy as np
import pmt
# bootstrap satellites module, even from build dir
try:
import python as satellites
except ImportError:
pass
else:
import sys
sys.modules['satellites'] = satellites
from satellites import crc, crc_append, crc_check
class qa_crc(gr_unittest.TestCase):
def setUp(self):
"""Common part of all CRC tests
Creates a flowgraph, a Message Debug block, and a PDU
containing the numbers 0x00 through 0x0F.
"""
self.tb = gr.top_block()
self.dbg = blocks.message_debug()
self.data = list(range(16))
self.pdu = pmt.cons(pmt.PMT_NIL,
pmt.init_u8vector(len(self.data), self.data))
def run_crc_append(self, crc_params, crc_result):
"""Common part of CRC Append tests
Creates a CRC Append block with the specified crc_params parameters,
connects it to the Message Debug block, sends a test PDU to the
CRC Append block, and checks that the output PDU matches the expected
crc_result.
"""
crc_append_block = crc_append(*crc_params)
self.tb.msg_connect((crc_append_block, 'out'), (self.dbg, 'store'))
crc_append_block.to_basic_block()._post(pmt.intern('in'), self.pdu)
crc_append_block.to_basic_block()._post(
pmt.intern('system'),
pmt.cons(pmt.intern('done'), pmt.from_long(1)))
self.tb.start()
self.tb.wait()
self.assertEqual(self.dbg.num_messages(), 1)
out = pmt.u8vector_elements(pmt.cdr(self.dbg.get_message(0)))
self.assertEqual(out[:len(self.data)], self.data)
self.assertEqual(out[len(self.data):], crc_result)
def common_test_crc_check(self, matching_crc, header_bytes=0):
"""Common part of CRC Check tests
Creates a CRC Append block and a CRC Check block using either the
same CRC or a different one depending on the whether matching_crc
is True or False. Connects CRC Append -> CRC Check -> Message Debug
and sends a PDU through. There are two message debugs to allow
checking whether the PDU ended up in the ok or fail port of the
CRC Check block.
"""
crc_append_block = crc_append(
16, 0x1021, 0x0, 0x0, False, False, False, header_bytes)
x = 0x0 if matching_crc else 0xFFFF
crc_check_block = crc_check(
16, 0x1021, x, x, False, False, False, True, header_bytes)
self.dbg_fail = blocks.message_debug()
self.tb.msg_connect((crc_append_block, 'out'), (crc_check_block, 'in'))
self.tb.msg_connect((crc_check_block, 'ok'), (self.dbg, 'store'))
self.tb.msg_connect((crc_check_block, 'fail'),
(self.dbg_fail, 'store'))
crc_append_block.to_basic_block()._post(pmt.intern('in'), self.pdu)
crc_append_block.to_basic_block()._post(
pmt.intern('system'),
pmt.cons(pmt.intern('done'), pmt.from_long(1)))
self.tb.start()
self.tb.wait()
def test_crc_check(self):
"""Test a successful CRC check
Checks that the PDU ends in the ok port of CRC check
"""
self.common_test_crc_check(matching_crc=True)
self.assertEqual(self.dbg.num_messages(), 1)
out = pmt.u8vector_elements(pmt.cdr(self.dbg.get_message(0)))
self.assertEqual(out, self.data)
self.assertEqual(self.dbg_fail.num_messages(), 0)
def test_crc_check_header_bytes(self):
"""Test a successful CRC check (skipping some header bytes)
Checks that the PDU ends in the ok port of CRC check
"""
self.common_test_crc_check(matching_crc=True, header_bytes=5)
self.assertEqual(self.dbg.num_messages(), 1)
out = pmt.u8vector_elements(pmt.cdr(self.dbg.get_message(0)))
self.assertEqual(out, self.data)
self.assertEqual(self.dbg_fail.num_messages(), 0)
def test_crc_check_wrong_crc(self):
"""Test a failed CRC check
Checks that the PDU ends in the fail port of CRC check
"""
self.common_test_crc_check(matching_crc=False)
self.assertEqual(self.dbg.num_messages(), 0)
self.assertEqual(self.dbg_fail.num_messages(), 1)
out = pmt.u8vector_elements(pmt.cdr(self.dbg_fail.get_message(0)))
self.assertEqual(out, self.data)
def test_crc_append_crc16_ccitt_zero(self):
"""Test CRC-16-CCITT-Zero calculation"""
self.run_crc_append(
(16, 0x1021, 0x0, 0x0,
False, False, False),
[0x51, 0x3D])
def test_crc_append_crc16_ccitt_false(self):
"""Test CRC-16-CCITT-False calculation"""
self.run_crc_append(
(16, 0x1021, 0xFFFF, 0x0,
False, False, False),
[0x3B, 0x37])
def test_crc_append_crc16_ccitt_x25(self):
"""Test CRC-16-CCITT-X.25 calculation"""
self.run_crc_append(
(16, 0x1021, 0xFFFF, 0xFFFF,
True, True, False),
[0x13, 0xE9])
def test_crc_append_crc32(self):
"""Test CRC-32 calculation"""
self.run_crc_append(
(32, 0x4C11DB7, 0xFFFFFFFF, 0xFFFFFFFF,
True, True, False),
[0xCE, 0xCE, 0xE2, 0x88])
def test_crc_append_crc32c(self):
"""Test CRC-32C calculation"""
self.run_crc_append(
(32, 0x1EDC6F41, 0xFFFFFFFF, 0xFFFFFFFF,
True, True, False),
[0xD9, 0xC9, 0x08, 0xEB])
def test_crc_append_crc32c_endianness_swap(self):
"""Test CRC-32C calculation with endianness swapped"""
self.run_crc_append(
(32, 0x1EDC6F41, 0xFFFFFFFF, 0xFFFFFFFF,
True, True, True),
[0xEB, 0x08, 0xC9, 0xD9])
def test_crc_append_crc32c_skip_header_bytes(self):
"""Test CRC-32C calculation skipping some header bytes"""
skip_bytes = 3
self.run_crc_append(
(32, 0x1EDC6F41, 0xFFFFFFFF, 0xFFFFFFFF,
True, True, False, skip_bytes),
[0xE8, 0x62, 0x60, 0x68])
class qa_crc_class(gr_unittest.TestCase):
def test_crc_crc32c(self):
"""Test CRC-32C calculation (using crc class directly)"""
c = crc(32, 0x1EDC6F41, 0xFFFFFFFF, 0xFFFFFFFF, True, True)
out = c.compute(list(range(16)))
self.assertEqual(c.compute(list(range(16))),
0xD9C908EB)
if __name__ == '__main__':
gr_unittest.run(qa_crc)
gr_unittest.run(qa_crc_class)
|
