Simplify audio sink iteration in docking station tests (New)

checkbox-support/checkbox_support/scripts/pipewire_utils.py

@@ -78,7 +78,7 @@ class PipewireTest:
 
     logger = logging.getLogger()
 
-    def _get_pw_type(self, media_class) -> str:
+    def _get_pw_type(self, media_class: str) -> str:
         """
         convert sink to Output and source to Input
 
@@ -96,7 +96,9 @@ def _get_pw_type(self, media_class) -> str:
             self.logger.info("Media class:[{}] is unknown".format(media_class))
             return "UNKNOWN CLASS"
 
-    def _get_pw_dump(self, p_type) -> dict:
+    def _get_pw_dump(
+        self, p_type: 't.Literal["Device", "Node"]'
+    ) -> "list[dict[str, t.Any]]":
         """
         Use to convert the json output of pw-dump to dict object
 
@@ -113,7 +115,7 @@ def _get_pw_dump(self, p_type) -> dict:
             return json.loads(pw_dump)
         except (json.decoder.JSONDecodeError, TypeError):
             self.logger.error("pw-dump {} failed !!!".format(p_type))
-            return {}
+            return []
 
     def generate_pw_media_class(self, media_type, media_class) -> str:
         """
@@ -319,7 +321,7 @@ def gst_pipeline(self, pipe, timeout, device) -> int:
 
         return PipewireTestError.NO_ERROR
 
-    def _get_audio_config(self, mode) -> set:
+    def _get_audio_config(self, mode):
         """
         Get simple audio configuration
         This function parse output of pw-dump to find the device type
@@ -331,7 +333,7 @@ def _get_audio_config(self, mode) -> set:
         :type mode: str
         """
         clients = self._get_pw_dump("Device")
-        cfg = set()
+        cfg = set()  # type: set[tuple[str, str, str]]
         for client in clients:
             active_ports = None
             mclass = client["info"]["props"].get("media.class")
@@ -380,7 +382,7 @@ def monitor_active_port_change(self, timeout, mode) -> int:
         self.logger.info("Couldn't detect active port change!")
         return PipewireTestError.NO_CHANGE_DETECTED
 
-    def go_through_ports(self, cmd, mode):
+    def go_through_ports(self, cmd: str, mode):
         """
         Go through available ports for testing
         This script checks if the ports on either sinks
@@ -395,46 +397,109 @@ def go_through_ports(self, cmd, mode):
         """
         clients = self._get_pw_dump("Device")
         for client in clients:
-            ports = None
+            ports = []
             mclass = client["info"]["props"].get("media.class")
             if mclass == "Audio/Device":
                 ports = client["info"]["params"]["EnumRoute"]
-            if ports:
-                for p in ports:
-                    chosen = None
-                    if p["direction"] == self._get_pw_type(mode) and p[
-                        "available"
-                    ] in [
-                        "yes",
-                        "unknown",
-                    ]:
-                        while chosen != "yes":
-                            self.logger.info(
-                                "Please select [{}] for "
-                                "testing (if selected, "
-                                "please enter 'yes')".format(p["description"])
-                            )
+                assert type(ports) is list
+
+            for p in ports:
+                chosen = None
+                if p["direction"] == self._get_pw_type(mode) and p[
+                    "available"
+                ] in [
+                    "yes",
+                    "unknown",
+                ]:
+                    while chosen != "yes":
+                        self.logger.info(
+                            "Please select [{}] for "
+                            "testing (if selected, "
+                            "please enter 'yes')".format(p["description"])
+                        )
 
-                            chosen = input()
-                        checked = None
-                        while checked != "yes":
-                            with subprocess.Popen(
-                                cmd,
-                                shell=True,
-                                stdout=subprocess.PIPE,
-                                universal_newlines=True,
-                            ) as p:
-                                while p.poll() is None:
-                                    line = p.stdout.readline().strip()
-                                    self.logger.info(line)
-                                p.kill()
-                            self.logger.info(
-                                "Is working ?  please enter 'yes' to leave"
-                            )
+                        chosen = input()
+                    checked = None
+                    while checked != "yes":
+                        # check_call will print to stdout for us
+                        subprocess.check_call(cmd, shell=True)
+                        self.logger.info(
+                            "Is working ?  please enter 'yes' to leave"
+                        )
+                        checked = input()
 
-                            checked = input()
+    def iter_audio_sinks(self, cmd: str) -> None:
+        """Execute the cmd for each audio sink discovered by pipewire
 
-    def _get_node_description(self, properties) -> str:
+        :param cmd: the command to run
+        """
+        total_sinks_tested = 0
+
+        pw_audio_devices = [
+            device
+            for device in self._get_pw_dump("Device")
+            if device["info"]["props"].get("media.class") == "Audio/Device"
+        ]
+        pw_sink_nodes = [
+            node
+            for node in self._get_pw_dump("Node")
+            if node["info"]["props"].get("media.class") == "Audio/Sink"
+        ]
+
+        for node in pw_sink_nodes:
+            # IDs of these "nodes" can be passed to wpctl set-default
+            node_id = int(node["id"])
+
+            device_id = int(node["info"]["props"]["device.id"])
+            device = None  # type: dict[str, t.Any] | None
+            for dev in pw_audio_devices:
+                if dev["id"] == device_id:
+                    device = dev
+                    break
+            assert device, "Could not find device {}".format(device_id)
+
+            # now check if the device has at least 1 available route
+            enum_routes = device["info"]["params"]["EnumRoute"]
+            assert type(enum_routes) is list
+            testable = False
+            for route in enum_routes:
+                # try to match the device to this node
+                if (
+                    route["devices"][0]  # this is an array with just 1 value
+                    != node["info"]["props"]["card.profile.device"]
+                ):
+                    continue
+                if route["direction"] != "Output":
+                    print(
+                        "Skipping '{}'".format(route["description"]),
+                        "because it's not a sink",
+                    )
+                    continue
+                if route["available"] not in ("yes", "unknown"):
+                    print(
+                        "Skipping '{}'".format(route["description"]),
+                        "because it's unavailable",
+                    )
+                    continue
+                # correct direction + at least 1 available route
+                testable = True
+
+            if testable:
+                print("=" * 80, flush=True)
+                total_sinks_tested += 1
+                # now switch to this sink
+                subprocess.check_call(["wpctl", "set-default", str(node_id)])
+                print(
+                    "Testing sink '{}', node id = '{}'".format(
+                        node["info"]["props"]["node.description"], node_id
+                    ),
+                    "with command '{}'".format(cmd),
+                )
+                subprocess.check_call(cmd, shell=True)
+                print("=" * 80, flush=True)
+        print("Tested", total_sinks_tested, "audio sinks in total")
+
+    def _get_node_description(self, properties) -> "str | None":
         """
         Get node description from the output of wpctl inspect
 
@@ -490,7 +555,7 @@ def show_default_device(self, device_type):
         except subprocess.CalledProcessError as e:
             raise RuntimeError("Show default device error {}".format(repr(e)))
 
-    def _sort_wpctl_status(self, lines: list) -> list:
+    def _sort_wpctl_status(self, lines: "list[str]") -> "list[str]":
         """
         This method will sort wpctl status for sub-items under catalog only
 
@@ -533,10 +598,10 @@ def compare_wpctl_status(self, status_1: str, status_2: str):
         :param status_2: path to second wpctl status
         """
         with open(status_1, "r") as s1, open(status_2, "r") as s2:
-            status_1 = s1.readlines()
-            status_2 = s2.readlines()
-            sorted_status_1 = self._sort_wpctl_status(status_1)
-            sorted_status_2 = self._sort_wpctl_status(status_2)
+            status_1_lines = s1.readlines()
+            status_2_lines = s2.readlines()
+            sorted_status_1 = self._sort_wpctl_status(status_1_lines)
+            sorted_status_2 = self._sort_wpctl_status(status_2_lines)
             delta = difflib.unified_diff(sorted_status_1, sorted_status_2, n=0)
             diff = "".join(delta)
             if diff:
@@ -753,6 +818,17 @@ def _args_parsing(self, args=sys.argv[1:]):
             "-m", "--mode", type=str, help="Either sinks or sources"
         )
 
+        parser_iter_sink = subparsers.add_parser(
+            "iter-audio-sinks", help="Iterate all available audio sinks"
+        )
+        parser_iter_sink.add_argument(
+            "-c",
+            "--command",
+            type=str,
+            required=True,
+            help="command for testing",
+        )
+
         # Add parser for show default device function
         parser_show = subparsers.add_parser(
             "show", help="show the default device"
@@ -823,6 +899,8 @@ def function_select(self, args):
                 return PipewireTestError.NO_ERROR
             else:
                 return PipewireTestError.NOT_REAL_DEVICE
+        elif args.test_type == "iter-audio-sinks":
+            return self.iter_audio_sinks(args.command)
 
 
 def main():

providers/base/units/dock/jobs.pxu

@@ -2598,7 +2598,7 @@ plugin: user-interact-verify
 estimated_duration: 60.0
 command:
     if check_audio_daemon.sh ; then
-        checkbox-support-pipewire-utils through -m sink -c "speaker-test -c 2 -l 1 -t wav"
+        checkbox-support-pipewire-utils iter-audio-sinks -m sink -c "speaker-test -c 2 -l 1 -t wav"
     else
         indexes=$(pacmd list-sinks | grep -e 'index' -e 'available' | grep -B 1 -e 'available: unknown' -e 'available: yes' | grep index | awk '{print $NF}')
         for index in $indexes