From f3fd2f699fe4e793444d54a89768d61056a5c073 Mon Sep 17 00:00:00 2001
From: Aditya Kumar Singh <quic_adisi@quicinc.com>
Date: Tue, 13 Aug 2024 14:08:52 +0530
Subject: [PATCH] tests: AP MLD control socket connectivity test case

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

Signed-off-by: Aditya Kumar Singh <quic_adisi@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 000000000..623702c00
--- /dev/null
+++ b/tests/hwsim/mld.py
@@ -0,0 +1,36 @@
+# Python class for controlling Multi Link Device
+# Copyright (c) 2024, Jouni Malinen <j@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 MultiLinkDevice:
+    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 = MultiLinkDevice(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 c9dfdafc4..a9fcf87a7 100644
--- a/tests/hwsim/test_eht.py
+++ b/tests/hwsim/test_eht.py
@@ -13,6 +13,7 @@ from hwsim import HWSimRadio
 import hwsim_utils
 from wpasupplicant import WpaSupplicant
 import re
+import mld
 from tshark import run_tshark
 from test_gas import hs20_ap_params
 from test_dpp import check_dpp_capab, wait_auth_success
@@ -2215,3 +2216,36 @@ def test_eht_mlo_color_change(dev, apdev):
 
         hapd0.dump_monitor()
         hapd1.dump_monitor()
+
+def test_eht_mld_control_socket_connectivity(dev, apdev):
+    """AP MLD control 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"}
+        link1_params = {"ssid": ssid,
+                        "hw_mode": "g",
+                        "channel": "2"}
+
+        hapd0 = eht_mld_enable_ap(hapd_iface, 0, link0_params)
+        hapd1 = eht_mld_enable_ap(hapd_iface, 1, link1_params)
+
+        mld_dev = mld.get_mld_obj(hapd_iface)
+
+        # Check status of each link
+        res = str(mld_dev.request("LINKID 0 STATUS"))
+        logger.info("LINK 0 STATUS:\n" + res)
+        if "state" not in res:
+            raise Exception("Failed to get link 0 status via MLD socket")
+        if 'link_id=0' not in res.splitlines():
+            raise Exception("link_id=0 not reported for link 0")
+
+        res = mld_dev.request("LINKID 1 STATUS")
+        logger.info("LINK 1 STATUS:\n" + res)
+        if "state" not in res:
+            raise Exception("Failed to get link 1 status via MLD socket")
+        if 'link_id=1' not in res.splitlines():
+            raise Exception("link_id=0 not reported for link 1")