blocks: msg_pair_to_var: Log with block logger, not gr.log

This also adds the missing unit test.

Signed-off-by: Marcus Müller <[email protected]>

gr-blocks/python/blocks/msg_pair_to_var.py

@@ -20,27 +20,29 @@ class msg_pair_to_var(gr.sync_block):
     unpredictable.
     """
 
+    IN_PORT = pmt.intern("inpair")
+
     def __init__(self, callback):
         gr.sync_block.__init__(
             self, name="msg_pair_to_var", in_sig=None, out_sig=None)
 
         self.callback = callback
 
-        self.message_port_register_in(pmt.intern("inpair"))
-        self.set_msg_handler(pmt.intern("inpair"), self.msg_handler)
+        self.message_port_register_in(self.IN_PORT)
+        self.set_msg_handler(self.IN_PORT, self.msg_handler)
 
     def msg_handler(self, msg):
         if not pmt.is_pair(msg) or pmt.is_dict(msg) or pmt.is_pdu(msg):
-            gr.log.warn(
-                "Input message %s is not a simple pair, dropping" % repr(msg))
+            self.logger.warn(
+                f"Input message {msg} is not a simple pair, dropping")
             return
 
         new_val = pmt.to_python(pmt.cdr(msg))
         try:
             self.callback(new_val)
         except Exception as e:
-            gr.log.error("Error when calling " + repr(self.callback) + " with " +
-                         repr(new_val) + " (reason: %s)" % repr(e))
+            self.logger.error(
+                f"Error when calling {self.callback} with {new_val} (reason: {e})")
 
     def stop(self):
         return True

gr-blocks/python/blocks/qa_msg_pair_to_var.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+#
+# Copyright 2023 Marcus Müller
+#
+# This file is part of GNU Radio
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+#
+
+
+from gnuradio import blocks
+from gnuradio import gr_unittest
+from gnuradio import gr
+import pmt
+import sys
+import time
+
+
+class test_msg_pair_to_var(gr_unittest.TestCase):
+
+    def setUp(self):
+        self.backend = gr.dictionary_logger_backend()
+        gr.logging().add_default_sink(self.backend)
+        self.tb = gr.top_block(catch_exceptions=False)
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_calling(self):
+        canary = 0xDEAFBEEF
+        self.test_val = 0
+
+        def test_f(value: int = 0):
+            self.test_val = value
+
+        dut = blocks.msg_pair_to_var(test_f)
+        test_msg = pmt.cons(pmt.intern("foo"), pmt.from_long(canary))
+        src = blocks.message_strobe(test_msg, 10)
+        self.tb.msg_connect(*(src, pmt.intern("strobe")),
+                            *(dut, blocks.msg_pair_to_var.IN_PORT))
+        self.tb.start()
+        time.sleep(15e-3)
+        self.tb.stop()
+        self.tb.wait()
+
+        self.assertEqual(self.test_val, canary,
+                         f"Test value {hex(self.test_val)} not found to be canary {hex(canary)}")
+
+    def test_exceptions(self):
+        lark = 0xDEAFBEEF
+        nightingale = 0xDEAFB16D
+        self.test_val = 0
+
+        def test_f(value: int = 0):
+            if value == nightingale and value != lark:
+                raise Exception("It was the nightingale, and not the lark")
+
+        dut = blocks.msg_pair_to_var(test_f)
+        test_msg = pmt.cons(pmt.intern("birb"), pmt.from_long(nightingale))
+        src = blocks.message_strobe(test_msg, 10)
+        self.tb.msg_connect(*(src, pmt.intern("strobe")),
+                            *(dut, blocks.msg_pair_to_var.IN_PORT))
+        self.tb.start()
+        time.sleep(50e-3)
+        self.tb.stop()
+        self.tb.wait()
+
+        logged = self.backend.get_map()
+        self.assertIn("msg_pair_to_var", logged, "should have logged; didn't.")
+        log_entries = logged["msg_pair_to_var"]
+        for timestamp, entry in log_entries:
+            self.assertRegex(
+                entry,
+                r"Error when calling <function .*\.test_exceptions\..*.test_f.*> with 3736056173" +
+                r".*reason: It was the nightingale, and not the lark.*",
+                f"Log entry '{entry}' from {timestamp} doesn't contain Shakespeare")
+
+
+if __name__ == '__main__':
+    gr_unittest.run(test_msg_pair_to_var)