Use regular expression to specify interfaces to check

scripts/monitor_dbus_signals.py

@@ -18,6 +18,7 @@
 with respect to their properties.
 """
 
+_INTERFACE_RE = None
 _MO = None
 _SERVICE = None
 _TOP_OBJECT = None
@@ -41,6 +42,7 @@
     # isort: STDLIB
     import argparse
     import os
+    import re
     import sys
     import time
     import xml.etree.ElementTree as ET
@@ -150,7 +152,14 @@ def _make_mo():
         the result of calling Properties.GetAll on the top object for
         selected interfaces.
         """
+
         mos = _OBJECT_MANAGER.Methods.GetManagedObjects(_TOP_OBJECT, {})
+
+        mos = {
+            o: {k: v for k, v in d.items() if re.fullmatch(_INTERFACE_RE, k)}
+            for o, d in mos.items()
+        }
+
         mos[_TOP_OBJECT_PATH] = {}
 
         for interface in _TOP_OBJECT_INTERFACES:
@@ -170,6 +179,10 @@ def _interfaces_added(object_path, interfaces_added):
         :param str object_path: D-Bus object path
         :param dict interfaces_added: map of interfaces to D-Bus properties
         """
+        interfaces_added = {
+            k: v for k, v in interfaces_added.items() if re.fullmatch(_INTERFACE_RE, k)
+        }
+
         if object_path == _TOP_OBJECT_PATH:
             interfaces_added = {
                 k: v for k, v in interfaces_added.items() if k in _TOP_OBJECT_INTERFACES
@@ -202,11 +215,6 @@ def _interfaces_removed(object_path, interfaces):
         :param str object_path: D-Bus object path
         :param list interfaces: list of interfaces removed
         """
-        if object_path == _TOP_OBJECT_PATH:
-            interfaces = {
-                k: v for k, v in interfaces.items() if k in _TOP_OBJECT_INTERFACES
-            }
-
         try:
             print(
                 "Interfaces removed:",
@@ -220,7 +228,8 @@ def _interfaces_removed(object_path, interfaces):
 
             if object_path in _MO.keys():
                 for interface in interfaces:
-                    del _MO[object_path][interface]
+                    if interface in _MO[object_path]:
+                        del _MO[object_path][interface]
 
                 # The InterfacesRemoved signal is sent when an object is
                 # removed as well as when a single interface is removed.
@@ -248,6 +257,9 @@ def _properties_changed(*props_changed, object_path=None):
                 return
 
             interface_name = props_changed[0]
+            if not re.fullmatch(_INTERFACE_RE, interface_name):
+                return
+
             properties_changed = props_changed[1]
             properties_invalidated = props_changed[2]
 
@@ -297,23 +309,26 @@ def _properties_changed(*props_changed, object_path=None):
         except Exception as exc:  # pylint: disable=broad-except
             _CALLBACK_ERRORS.append(exc)
 
-    def _monitor(service, manager, manager_interfaces):
+    def _monitor(service, manager, manager_interfaces, interface_re):
         """
         Monitor the signals and properties of the manager object.
 
         :param str service: the service to monitor
         :param str manager: object path that of the ObjectManager implementor
         :param manager_interfaces: list of manager interfaces
         :type manager_interfaces: list of str
+        :param interface_re: regular expression to match interfaces to check
+        :type interface_re: re.Pattern
         """
 
-        global _TOP_OBJECT, _TOP_OBJECT_PATH, _TOP_OBJECT_INTERFACES, _SERVICE, _MO  # pylint: disable=global-statement
+        global _TOP_OBJECT, _TOP_OBJECT_PATH, _TOP_OBJECT_INTERFACES, _SERVICE, _MO, _INTERFACE_RE  # pylint: disable=global-statement
 
         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
         bus = dbus.SystemBus()
         _SERVICE = service
         _TOP_OBJECT_PATH = manager
         _TOP_OBJECT_INTERFACES = manager_interfaces
+        _INTERFACE_RE = interface_re
 
         while True:
             try:
@@ -387,6 +402,13 @@ def _gen_parser():
             help="interface belonging to the top object",
         )
 
+        parser.add_argument(
+            "--only-check",
+            default=".*",
+            type=re.compile,
+            help="regular expression that restricts interfaces to check",
+        )
+
         return parser
 
     def main():
@@ -398,7 +420,7 @@ def main():
 
         args = parser.parse_args()
 
-        _monitor(args.service, args.manager, args.top_interface)
+        _monitor(args.service, args.manager, args.top_interface, args.only_check)
 
     if __name__ == "__main__":
         main()

testlib/infra.py

@@ -343,6 +343,16 @@ def setUp(self):
                     + 1
                 )
             )
+
+            only_check = (
+                StratisDbus.BUS_NAME.replace(".", r"\.")
+                + r"\."
+                + ".*"
+                + r"\."
+                + f"r[0-{StratisDbus.REVISION_NUMBER}]"
+            )
+            command.append(f'--only-check="{only_check}"')
+
             # pylint: disable=consider-using-with
             try:
                 self.trace = subprocess.Popen(