From 375b9573c6185f2713a211a087dfbabedeb50c05 Mon Sep 17 00:00:00 2001
From: lenhart <malte.lenhart@t-online.de>
Date: Fri, 22 Oct 2021 21:26:08 +0200
Subject: blocks: message strobe - add unit test
* added unittest to the message strobe block (by testing against the message debug block).
Taken test method from gr-zeromq message blocks as pointed out by Vasil.
Signed-off-by: Lenhart <malte.lenhart@mailbox.org>
---
gr-blocks/python/blocks/qa_message_debug.py | 60 +++++++++++++++++++++++++++++
1 file changed, 60 insertions(+)
create mode 100644 gr-blocks/python/blocks/qa_message_debug.py
(limited to 'gr-blocks/python/blocks/qa_message_debug.py')
diff --git a/gr-blocks/python/blocks/qa_message_debug.py b/gr-blocks/python/blocks/qa_message_debug.py
new file mode 100644
index 0000000000..0f1dcff6fe
--- /dev/null
+++ b/gr-blocks/python/blocks/qa_message_debug.py
@@ -0,0 +1,60 @@
+#!/usr/bin/env python
+#
+# Copyright 2021 Malte Lenhart
+#
+# This file is part of GNU Radio
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+#
+
+from gnuradio import gr, gr_unittest
+from gnuradio import blocks
+import pmt
+import time
+
+# this test tests message strobe and message debug blocks against each other
+# similar tests contained in message_strobe class
+
+# this tests only the store port and the message retrival methods of the debug block
+# print() and print_pdu() were omitted as they print to stdout
+class qa_message_debug(gr_unittest.TestCase):
+
+ def setUp(self):
+ self.tb = gr.top_block()
+
+ def tearDown(self):
+ self.tb = None
+
+ def test_001_t(self):
+ test_str = "test_msg"
+ new_msg = "new_msg"
+ message_period_ms = 100
+ msg_strobe = blocks.message_strobe(pmt.intern(test_str), message_period_ms)
+ msg_debug = blocks.message_debug()
+
+ self.tb.msg_connect(msg_strobe, "strobe", msg_debug, "store")
+
+ self.tb.start()
+
+ self.assertAlmostEqual(msg_debug.num_messages(), 0, delta=2) # 1st call, expect 0
+ time.sleep(1) # floor(1000/100) = 10
+ self.assertAlmostEqual(msg_debug.num_messages(), 10, delta=3) # 2nd call == 1
+ time.sleep(1) # floor(2000/100) = 15
+ self.assertAlmostEqual(msg_debug.num_messages(), 20, delta=3) # 3th call == 3
+
+ # change test message
+ msg_strobe.to_basic_block()._post(pmt.intern("set_msg"), pmt.intern(new_msg))
+ time.sleep(1)
+
+ self.tb.stop()
+ self.tb.wait()
+ # check data
+ # first received message matchs initial test message
+ self.assertAlmostEqual(pmt.to_python(msg_debug.get_message(0)), test_str, "mismatch initial test string")
+
+ # last message matches changed test message
+ no_msgs = msg_debug.num_messages()
+ self.assertAlmostEqual(pmt.to_python(msg_debug.get_message(no_msgs - 1)), new_msg, "failed to update string")
+if __name__ == '__main__':
+ gr_unittest.run(qa_message_debug)
--
cgit v1.2.3