[PATCH 6/6] tests: MLO: add MLD socket connectivity test case

[Aditya Kumar Singh]

Add simple test case to bring up a 2 link MLD and get the status of each
link via the MLD level socket.

Signed-off-by: Aditya Kumar Singh <quic_adisi at quicinc.com>
---
 tests/hwsim/mld.py      | 36 ++++++++++++++++++++++++++++++++++++
 tests/hwsim/test_eht.py | 34 ++++++++++++++++++++++++++++++++++
 2 files changed, 70 insertions(+)
 create mode 100644 tests/hwsim/mld.py

diff --git a/tests/hwsim/mld.py b/tests/hwsim/mld.py
new file mode 100644
index 000000000000..af6695e601df
--- /dev/null
+++ b/tests/hwsim/mld.py
@@ -0,0 +1,36 @@
+# Python class for controlling Multi Link Device
+# Copyright (c) 2024, Jouni Malinen <j at w1.fi>
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+
+import os
+import logging
+import wpaspy
+
+logger = logging.getLogger()
+hapd_ctrl = '/var/run/hostapd'
+
+class Multi_Link_Device:
+    def __init__(self, ifname, ctrl=hapd_ctrl, port=8877):
+        self.ifname = ifname
+        self.ctrl = wpaspy.Ctrl(os.path.join(ctrl, ifname))
+        self.dbg = ifname
+
+    def close_ctrl(self):
+        self.ctrl.close()
+        self.ctrl = None
+
+    def request(self, cmd):
+        logger.debug(self.dbg + ": MLD CTRL: " + cmd)
+        return self.ctrl.request(cmd)
+
+    def ping(self):
+        return "PONG" in self.request("PING")
+
+def get_mld_obj(ifname, ctrl=hapd_ctrl, port=8877):
+    mld = Multi_Link_Device(ifname, ctrl, port)
+    if not mld.ping():
+        raise Exception("Could not ping MLD %s" % ifname)
+
+    return mld
diff --git a/tests/hwsim/test_eht.py b/tests/hwsim/test_eht.py
index 8acb6a57bc7b..6b44dad3c09f 100644
--- a/tests/hwsim/test_eht.py
+++ b/tests/hwsim/test_eht.py
@@ -15,6 +15,7 @@ from tshark import run_tshark
 from test_gas import hs20_ap_params
 from test_dpp import check_dpp_capab, wait_auth_success
 from test_rrm import build_beacon_request, run_req_beacon, BeaconReport
+import mld
 
 def eht_verify_wifi_version(dev):
     status = dev.get_status()
@@ -1905,3 +1906,36 @@ def test_eht_mlo_csa(dev, apdev):
             traffic_test(wpas, hapd0)
 
             #TODO: CSA on non-first link
+
+def test_eht_mld_socket_connectivity(dev, apdev):
+    """EHT MLD Socket Connectivity"""
+    with HWSimRadio(use_mlo=True) as (hapd_radio, hapd_iface), \
+        HWSimRadio(use_mlo=True) as (wpas_radio, wpas_iface):
+
+        ssid = "mld_ap"
+        link0_params = {"ssid": ssid,
+                        "hw_mode": "g",
+                        "channel": "1",
+                        "link_id": "0"}
+        link1_params = {"ssid": ssid,
+                        "hw_mode": "g",
+                        "channel": "2",
+                        "link_id": "1"}
+
+        hapd0 = eht_mld_enable_ap(hapd_iface, link0_params)
+        hapd1 = eht_mld_enable_ap(hapd_iface, link1_params)
+
+        mld_dev = mld.get_mld_obj(hapd_iface)
+
+        # Check status of each link
+        res = str(mld_dev.request("STATUS LINKID 0"))
+        if "state" not in res:
+            raise Exception("Failed to get link 0 status via MLD socket")
+
+        logger.info("LINK 0 STATUS: \n" + res)
+
+        res = mld_dev.request("STATUS LINKID 1")
+        if "state" not in res:
+            raise Exception("Failed to get link 1 status via MLD socket")
+
+        logger.info("LINK 1 STATUS: \n" + res)
-- 
2.34.1