Refactor WIFI STAs resource job and Support Wifi 7 (New)

providers/resource/bin/WIFI_phy.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+
+# This file is part of Checkbox.
+#
+# Copyright 2024 Canonical Ltd.
+# Written by:
+#   Isaac Yang <[email protected]>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox. If not, see <http://www.gnu.org/licenses/>
+
+import re
+from subprocess import check_output
+from typing import Dict, List, Tuple
+
+
+def parse_iw_dev_output() -> List[Tuple[str, str]]:
+    """
+    Parses the output of "iw dev" to extract PHY and interface mappings.
+    """
+    output = check_output(["iw", "dev"], universal_newlines=True)
+    iw_dev_ptn = r"(phy.*)[\s\S]*?Interface (\w+)"
+    return re.findall(iw_dev_ptn, output)
+
+
+def parse_phy_info_output(output: str) -> Dict[str, List[str]]:
+    """
+    Parses the output of "iw phy info" to extract bands and STA support.
+    """
+    bands_data = output.split("Supported commands:")[0]
+    bands = {}
+    for band in bands_data.split("Band "):
+        if not band:
+            continue
+        band_ptn = r"(\d+):"
+        band_res = re.match(band_ptn, band)
+        if not band_res:
+            continue
+        band_num, band_content = band.split(":", 1)
+        # get Frequencies paragraph
+        freqs_raw = band_content.split("Frequencies:", 1)[1].split(":", 1)[0]
+        freqs = [
+            freq.strip()
+            for freq in freqs_raw.split("*")
+            if "disabled" not in freq and "MHz" in freq
+        ]
+        bands[band_num] = freqs
+    return bands
+
+
+def check_sta_support(phy_info_output: str) -> Dict[str, str]:
+    """
+    Checks if supported STAs (BE, AX, AC) are present based on keywords in
+    the output.
+    """
+    # Supported STAs and their keywords
+    supported_stas = {"BE": "EHT", "AX": "HE RX MCS", "AC": "VHT RX MCS"}
+
+    sta_supported = {
+        sta: "supported" if sta_keyword in phy_info_output else "unsupported"
+        for sta, sta_keyword in supported_stas.items()
+    }
+    return sta_supported
+
+
+def check_freq_support(bands: Dict[str, List[str]]) -> Dict[str, str]:
+    """
+    Checks if supported frequency (2.4GHz, 5GHz, 6GHz) are present based on
+    band number
+    """
+    # Ex. the frequency 2.4GHz is band 1, 5GHz is band 2, 6GHz is band 4
+    # so if bands[1] is not empty, the device supports 2.4GHz.
+    supported_freqs = {"2.4GHz": "1", "5GHz": "2", "6GHz": "4"}
+
+    freq_supported = {
+        freq: ("supported" if bands.get(band) else "unsupported")
+        for freq, band in supported_freqs.items()
+    }
+    return freq_supported
+
+
+def create_phy_interface_mapping(phy_interface: List[Tuple[str, str]]) -> Dict:
+    """
+    Creates a mapping between interfaces and their PHY, bands, and STA support.
+    """
+    phy_interface_mapping = {}
+    for phy, interface in phy_interface:
+        phy_info_output = check_output(
+            ["iw", phy, "info"], universal_newlines=True
+        )
+        bands = parse_phy_info_output(phy_info_output)
+        freq_supported = check_freq_support(bands)
+        sta_supported = check_sta_support(phy_info_output)
+        phy_interface_mapping[interface] = {
+            "PHY": phy,
+            "Bands": bands,
+            "FREQ_Supported": freq_supported,
+            "STA_Supported": sta_supported,
+        }
+    return phy_interface_mapping
+
+
+def main():
+    # Read and parse "iw dev" output
+    phy_interface = parse_iw_dev_output()
+
+    # Create mapping with interface, PHY, bands, and supported STAs
+    phy_interface_mapping = create_phy_interface_mapping(phy_interface)
+
+    # Print interface summary with detailed information on separate lines
+    for interface, content in phy_interface_mapping.items():
+        for freq, ret in content["FREQ_Supported"].items():
+            # replace . with _ to support resource expressions for 2.4GHz
+            # as . is not a valid character in resource expression names
+            # i.e.: wifi.wlan0_2_4GHz is valid, wifi.wlan0_2.4GHz is not
+            print("{}_{}: {}".format(interface, freq.replace(".", "_"), ret))
+        for sta, ret in content["STA_Supported"].items():
+            print("{}_{}: {}".format(interface, sta.lower(), ret))
+
+
+if __name__ == "__main__":
+    main()

providers/resource/jobs/resource.pxu

@@ -280,13 +280,8 @@ plugin: resource
 category_id: information_gathering
 _summary: Resource job to identify Wi-Fi STA supported protocols
 _description:
- Job listing STA supported 802.11 protocols per interfaces.
-command:
- # shellcheck disable=SC2046
- for i in $(iw dev | grep -oP 'Interface\s+\K\w+'); do iw phy phy$(iw dev "$i" info | grep -oP 'wiphy\s+\K\d+') info | grep -q 'VHT' && echo "$i""_ac: supported" || echo "$i""_ac: unsupported"; done
- # MCS 10 and 11 if present support the ax only 1024-QAM
- # shellcheck disable=SC2046
- for i in $(iw dev | grep -oP 'Interface\s+\K\w+'); do iw phy phy$(iw dev "$i" info | grep -oP 'wiphy\s+\K\d+') info | grep -q 'MCS 0-11' && echo "$i""_ax: supported" || echo "$i""_ax: unsupported"; done
+ Job listing STA supported 802.11 (AC, AX, BE) protocols per interfaces.
+command: WIFI_phy.py
 estimated_duration: 0.5
 flags: preserve-locale
 

providers/resource/tests/test_WIFI_phy.py

@@ -0,0 +1,194 @@
+import os
+from unittest.mock import patch
+import unittest
+from WIFI_phy import (
+    parse_iw_dev_output,
+    parse_phy_info_output,
+    check_sta_support,
+    check_freq_support,
+    create_phy_interface_mapping,
+    main,
+)
+
+
+class WIFIphyData:
+    @staticmethod
+    def get_text(filenmae):
+        full_path = os.path.join(
+            os.path.dirname(os.path.realpath(__file__)),
+            "test_WIFI_phy_data",
+            filenmae,
+        )
+        with open(full_path, "r", encoding="UTF-8") as stream:
+            return stream.read()
+
+
+class WIFITest(unittest.TestCase, WIFIphyData):
+
+    def setUp(self):
+        self.iw_dev_file_list = ["iw_dev.log", "iw_dev2.log"]
+        self.phy_info_file_list = [
+            "RTL8821CE_AC.log",
+            "QCNFA765_AX_6G.log",
+            "BE200_AX.log",
+            "BE200_BE.log",
+        ]
+
+    @patch("WIFI_phy.check_output")
+    def test_parse_iw_dev_output(self, mock_check_output):
+        expected_result = {
+            "iw_dev.log": [("phy#0", "wlp2s0")],
+            "iw_dev2.log": [("phy#0", "wlp3s0f0"), ("phy#1", "wlp3s0f1")],
+        }
+
+        for file in self.iw_dev_file_list:
+            mock_check_output.return_value = WIFIphyData.get_text(file)
+            result = parse_iw_dev_output()
+            self.assertEqual(result, expected_result[file])
+
+    @patch("WIFI_phy.check_output")
+    def test_create_phy_interface_mapping(self, mock_check_output):
+        iw_dev_file = self.iw_dev_file_list[0]
+        iw_dev_output = WIFIphyData.get_text(iw_dev_file)
+        excvected_result = {
+            "RTL8821CE_AC.log": {
+                "wlp2s0": {
+                    "PHY": "phy#0",
+                    "Bands": {
+                        "1": ["2412 MHz [1] (30.0 dBm)"],
+                        "2": ["5180 MHz [36] (23.0 dBm)"],
+                    },
+                    "FREQ_Supported": {
+                        "2.4GHz": "supported",
+                        "5GHz": "supported",
+                        "6GHz": "unsupported",
+                    },
+                    "STA_Supported": {
+                        "BE": "unsupported",
+                        "AX": "unsupported",
+                        "AC": "supported",
+                    },
+                }
+            },
+            "QCNFA765_AX_6G.log": {
+                "wlp2s0": {
+                    "PHY": "phy#0",
+                    "Bands": {
+                        "1": ["2412 MHz [1] (30.0 dBm)"],
+                        "2": ["5180 MHz [36] (24.0 dBm)"],
+                        "4": ["5955 MHz [1] (30.0 dBm)"],
+                    },
+                    "FREQ_Supported": {
+                        "2.4GHz": "supported",
+                        "5GHz": "supported",
+                        "6GHz": "supported",
+                    },
+                    "STA_Supported": {
+                        "BE": "unsupported",
+                        "AX": "supported",
+                        "AC": "supported",
+                    },
+                }
+            },
+            "BE200_AX.log": {
+                "wlp2s0": {
+                    "PHY": "phy#0",
+                    "Bands": {
+                        "1": ["2412 MHz [1] (22.0 dBm)"],
+                        "2": ["5180 MHz [36] (22.0 dBm)"],
+                        "4": [],
+                    },
+                    "FREQ_Supported": {
+                        "2.4GHz": "supported",
+                        "5GHz": "supported",
+                        "6GHz": "unsupported",
+                    },
+                    "STA_Supported": {
+                        "BE": "unsupported",
+                        "AX": "supported",
+                        "AC": "supported",
+                    },
+                }
+            },
+            "BE200_BE.log": {
+                "wlp2s0": {
+                    "PHY": "phy#0",
+                    "Bands": {
+                        "1": ["2412 MHz [1] (22.0 dBm)"],
+                        "2": ["5180 MHz [36] (22.0 dBm)"],
+                        "4": [],
+                    },
+                    "FREQ_Supported": {
+                        "2.4GHz": "supported",
+                        "5GHz": "supported",
+                        "6GHz": "unsupported",
+                    },
+                    "STA_Supported": {
+                        "BE": "supported",
+                        "AX": "supported",
+                        "AC": "supported",
+                    },
+                }
+            },
+        }
+        for file in self.phy_info_file_list:
+            phy_info_output = WIFIphyData.get_text(file)
+            mock_check_output.side_effect = [iw_dev_output, phy_info_output]
+            phy_interface = parse_iw_dev_output()
+            result = create_phy_interface_mapping(phy_interface)
+            self.assertDictEqual(result, excvected_result[file])
+
+    def test_parse_phy_info_output(self):
+        phy_info_output = WIFIphyData.get_text(self.phy_info_file_list[0])
+        result = parse_phy_info_output(phy_info_output)
+
+        expected_result = {
+            "1": ["2412 MHz [1] (30.0 dBm)"],
+            "2": ["5180 MHz [36] (23.0 dBm)"],
+        }
+        self.assertDictEqual(result, expected_result)
+
+    def test_check_sta_support(self):
+        phy_info_output = WIFIphyData.get_text(self.phy_info_file_list[0])
+        result = check_sta_support(phy_info_output)
+
+        expected_result = {
+            "BE": "unsupported",
+            "AX": "unsupported",
+            "AC": "supported",
+        }
+        self.assertDictEqual(result, expected_result)
+
+    def test_check_freq_support(self):
+        bands = {
+            "1": ["2412 MHz [1] (30.0 dBm)"],
+            "2": ["5180 MHz [36] (23.0 dBm)"],
+        }
+        result = check_freq_support(bands)
+        expected_result = {
+            "2.4GHz": "supported",
+            "5GHz": "supported",
+            "6GHz": "unsupported",
+        }
+        self.assertDictEqual(result, expected_result)
+
+    @patch("WIFI_phy.print")
+    @patch("WIFI_phy.check_output")
+    def test_main(self, mock_check_output, mock_print):
+        iw_dev_output = WIFIphyData.get_text(self.iw_dev_file_list[0])
+        phy_info_output = WIFIphyData.get_text(self.phy_info_file_list[0])
+        mock_check_output.side_effect = [iw_dev_output, phy_info_output]
+        main()
+        expected_calls = [
+            unittest.mock.call("wlp2s0_2_4GHz: supported"),
+            unittest.mock.call("wlp2s0_5GHz: supported"),
+            unittest.mock.call("wlp2s0_6GHz: unsupported"),
+            unittest.mock.call("wlp2s0_be: unsupported"),
+            unittest.mock.call("wlp2s0_ax: unsupported"),
+            unittest.mock.call("wlp2s0_ac: supported"),
+        ]
+        mock_print.assert_has_calls(expected_calls, any_order=True)
+
+
+if __name__ == "__main__":
+    unittest.main()