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
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2004,2010,2013,2016 Free Software Foundation, Inc.
#
# This file is part of GNU Radio
#
# SPDX-License-Identifier: GPL-3.0-or-later
#
#
import time
from gnuradio import gr, gr_unittest, blocks
import pmt
# def all_counts():
# return (gr.block_ncurrently_allocated(),
# gr.block_detail_ncurrently_allocated(),
# gr.buffer_ncurrently_allocated(),
# gr.buffer_reader_ncurrently_allocated(),
# gr.message_ncurrently_allocated())
class test_message(gr_unittest.TestCase):
# def setUp(self):
# # self.msgq = gr.msg_queue()
# def tearDown(self):
# # self.msgq = None
# def leak_check(self, fct):
# begin = all_counts()
# fct()
# # tear down early so we can check for leaks
# self.tearDown()
# end = all_counts()
# self.assertEqual(begin, end)
def test_100(self):
msg = gr.message(0, 1.5, 2.3)
self.assertEquals(0, msg.type())
self.assertAlmostEqual(1.5, msg.arg1())
self.assertAlmostEqual(2.3, msg.arg2())
self.assertEquals(0, msg.length())
def test_101(self):
s = b'This is a test'
msg = gr.message_from_string(s.decode('utf8'))
self.assertEquals(s, msg.to_string())
def test_102_unicodechars(self):
s = u"(╯°□°)╯︵ ┻━┻"
msg = gr.message_from_string(s)
self.assertEquals(s.encode('utf8'), msg.to_string())
# msg_queue was removed from API in 3.8
# def test_200(self):
# self.leak_check(self.body_200)
# def body_200(self):
# self.msgq.insert_tail(gr.message(0))
# self.assertEquals(1, self.msgq.count())
# self.msgq.insert_tail(gr.message(1))
# self.assertEquals(2, self.msgq.count())
# msg0 = self.msgq.delete_head()
# self.assertEquals(0, msg0.type())
# msg1 = self.msgq.delete_head()
# self.assertEquals(1, msg1.type())
# self.assertEquals(0, self.msgq.count())
# def test_201(self):
# self.leak_check(self.body_201)
# def body_201(self):
# self.msgq.insert_tail(gr.message(0))
# self.assertEquals(1, self.msgq.count())
# self.msgq.insert_tail(gr.message(1))
# self.assertEquals(2, self.msgq.count())
# def test_202(self):
# self.leak_check(self.body_202)
def body_202(self):
# global msg
msg = gr.message(666)
def test_300(self):
input_data = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
src = blocks.vector_source_b(input_data)
dst = blocks.vector_sink_b()
tb = gr.top_block()
tb.connect(src, dst)
tb.run()
self.assertEquals(input_data, dst.data())
def test_debug_401(self):
msg = pmt.intern("TESTING")
src = blocks.message_strobe(msg, 500)
snk = blocks.message_debug()
tb = gr.top_block()
tb.msg_connect(src, "strobe", snk, "store")
tb.start()
time.sleep(1)
tb.stop()
tb.wait()
rec_msg = snk.get_message(0)
self.assertTrue(pmt.eqv(rec_msg, msg))
if __name__ == '__main__':
gr_unittest.run(test_message)
|
