Tweak peripheral tests

Author: tannewt

ports/zephyr-cp/boards/nrf5340bsim_nrf5340_cpuapp.conf

@@ -12,7 +12,9 @@ CONFIG_BT_OBSERVER=y
 CONFIG_BT_BROADCASTER=y
 
 CONFIG_BT_L2CAP_TX_MTU=253
+CONFIG_BT_BUF_CMD_TX_COUNT=2
 CONFIG_BT_BUF_CMD_TX_SIZE=255
+CONFIG_BT_HCI_VS=y
 CONFIG_BT_BUF_EVT_RX_COUNT=16
 CONFIG_BT_BUF_EVT_RX_SIZE=255
 CONFIG_BT_BUF_ACL_TX_COUNT=3

ports/zephyr-cp/common-hal/_bleio/Adapter.c

@@ -13,6 +13,7 @@
 #include <zephyr/bluetooth/bluetooth.h>
 #include <zephyr/bluetooth/conn.h>
 #include <zephyr/bluetooth/hci.h>
+#include <zephyr/bluetooth/hci_vs.h>
 
 #include "py/gc.h"
 #include "py/runtime.h"
@@ -278,11 +279,48 @@ bool common_hal_bleio_adapter_get_enabled(bleio_adapter_obj_t *self) {
 }
 
 mp_int_t common_hal_bleio_adapter_get_tx_power(bleio_adapter_obj_t *self) {
-    mp_raise_NotImplementedError(NULL);
+    struct bt_hci_cp_vs_read_tx_power_level *cp;
+    struct bt_hci_rp_vs_read_tx_power_level *rp;
+    struct net_buf *buf, *rsp = NULL;
+
+    buf = bt_hci_cmd_alloc(K_MSEC(1000));
+    if (!buf) {
+        mp_raise_msg(&mp_type_MemoryError, NULL);
+    }
+    cp = net_buf_add(buf, sizeof(*cp));
+    cp->handle_type = BT_HCI_VS_LL_HANDLE_TYPE_ADV;
+    cp->handle = 0;
+
+    int err = bt_hci_cmd_send_sync(BT_HCI_OP_VS_READ_TX_POWER_LEVEL, buf, &rsp);
+    if (err) {
+        raise_zephyr_error(err);
+    }
+
+    rp = (void *)rsp->data;
+    int8_t power = rp->tx_power_level;
+    net_buf_unref(rsp);
+    return power;
 }
 
 void common_hal_bleio_adapter_set_tx_power(bleio_adapter_obj_t *self, mp_int_t tx_power) {
-    mp_raise_NotImplementedError(NULL);
+    struct bt_hci_cp_vs_write_tx_power_level *cp;
+    struct net_buf *buf, *rsp = NULL;
+
+    buf = bt_hci_cmd_alloc(K_MSEC(3000));
+    if (!buf) {
+        mp_raise_msg(&mp_type_MemoryError, NULL);
+    }
+    cp = net_buf_add(buf, sizeof(*cp));
+    cp->handle_type = BT_HCI_VS_LL_HANDLE_TYPE_ADV;
+    cp->handle = 0;
+    cp->tx_power_level = (int8_t)tx_power;
+
+    int err = bt_hci_cmd_send_sync(BT_HCI_OP_VS_WRITE_TX_POWER_LEVEL, buf, &rsp);
+    if (err) {
+        raise_zephyr_error(err);
+    }
+
+    net_buf_unref(rsp);
 }
 
 bleio_address_obj_t *common_hal_bleio_adapter_get_address(bleio_adapter_obj_t *self) {
@@ -321,7 +359,6 @@ void common_hal_bleio_adapter_start_advertising(bleio_adapter_obj_t *self,
     mp_buffer_info_t *advertising_data_bufinfo,
     mp_buffer_info_t *scan_response_data_bufinfo,
     mp_int_t tx_power, const bleio_address_obj_t *directed_to) {
-    (void)tx_power;
     (void)directed_to;
     (void)interval;
 
@@ -391,6 +428,8 @@ void common_hal_bleio_adapter_start_advertising(bleio_adapter_obj_t *self,
             NULL);
     }
 
+    common_hal_bleio_adapter_set_tx_power(self, tx_power);
+
     raise_zephyr_error(bt_le_adv_start(&adv_params,
         adv_data,
         adv_count,
@@ -555,7 +594,7 @@ mp_obj_t common_hal_bleio_adapter_connect(bleio_adapter_obj_t *self, bleio_addre
 
             if (info.state == BT_CONN_STATE_DISCONNECTED) {
                 bt_conn_unref(conn);
-                mp_raise_bleio_BluetoothError(MP_ERROR_TEXT("Failed to connect"));
+                mp_raise_bleio_BluetoothError(MP_ERROR_TEXT("Failed to connect: timeout"));
             }
         } else if (err != -ENOTCONN) {
             bt_conn_unref(conn);

ports/zephyr-cp/tests/bsim/conftest.py

@@ -107,6 +107,8 @@ def bsim_phy(request, bsim_phy_binary, native_sim_env, sim_id):
         "-v=9",  # Cleaning up level is on 9. Connecting is 7.
         f"-s={sim_id}",
         f"-D={devices}",
+        "-argschannel",
+        "-at=40",  # 40 dB attenuation (default 60) so RSSI ~ -40 dBm
     ]
     print("Running:", " ".join(cmd))
     proc = subprocess.Popen(
@@ -213,11 +215,13 @@ def zephyr_sample(request, bsim_phy, native_sim_env, sim_id):
     print(sample_proc.serial.all_output)
 
 
+# pytest markers are defined inside out meaning the bottom one is first in the
+# list and the top is last. So use negative indices to reverse them.
 @pytest.fixture
 def circuitpython1(circuitpython):
-    return circuitpython[0]
+    return circuitpython[-1]
 
 
 @pytest.fixture
 def circuitpython2(circuitpython):
-    return circuitpython[1]
+    return circuitpython[-2]

ports/zephyr-cp/tests/bsim/test_bsim_ble_advertising.py

@@ -4,6 +4,7 @@
 """BLE advertising tests for nrf5340bsim."""
 
 import logging
+import re
 
 import pytest
 
@@ -44,6 +45,39 @@
 """
 
 
+BSIM_TX_POWER_DEFAULT_CODE = """\
+import _bleio
+import time
+
+adapter = _bleio.adapter
+
+name = b"CPTXPWR"
+advertisement = bytes((2, 0x01, 0x06, len(name) + 1, 0x09)) + name
+
+print("advertising default")
+adapter.start_advertising(advertisement)
+time.sleep(4)
+adapter.stop_advertising()
+print("done")
+"""
+
+BSIM_TX_POWER_LOW_CODE = """\
+import _bleio
+import time
+
+adapter = _bleio.adapter
+
+name = b"CPTXPWR"
+advertisement = bytes((2, 0x01, 0x06, len(name) + 1, 0x09)) + name
+
+print("advertising low")
+adapter.start_advertising(advertisement, tx_power=-20)
+time.sleep(4)
+adapter.stop_advertising()
+print("done")
+"""
+
+
 @pytest.mark.zephyr_sample("bluetooth/observer")
 @pytest.mark.circuitpy_drive({"code.py": BSIM_ADV_CODE})
 def test_bsim_advertise_and_scan(bsim_phy, circuitpython, zephyr_sample):
@@ -90,3 +124,51 @@ def test_bsim_advertise_ctrl_c_reload(bsim_phy, circuitpython, zephyr_sample):
     assert cp_output.count("adv run done") >= 1
     assert observer_output.count("Device found:") >= observer_count_before + 1
     assert "Already advertising" not in cp_output
+
+
+@pytest.mark.zephyr_sample("bluetooth/observer")
+@pytest.mark.circuitpy_drive({"code.py": BSIM_TX_POWER_DEFAULT_CODE})
+def test_bsim_tx_power_default_rssi(bsim_phy, circuitpython, zephyr_sample):
+    """Verify default TX power produces expected RSSI."""
+    observer = zephyr_sample
+
+    circuitpython.wait_until_done()
+
+    cp_output = circuitpython.serial.all_output
+    obs_output = observer.serial.all_output
+
+    assert "advertising default" in cp_output
+    assert "done" in cp_output
+
+    # Observer: "Device found: <addr> (RSSI <n>), type <t>, AD data len <l>"
+    # Advertisement is 12 bytes: flags (3) + name (9).
+    # With 40 dB channel attenuation and 0 dBm TX → RSSI ~ -39
+    rssi_pattern = re.compile(r"RSSI (-?\d+)\), type \d+, AD data len 12")
+    all_rssi = [int(m.group(1)) for m in rssi_pattern.finditer(obs_output)]
+    logger.info("RSSI values: %s", all_rssi)
+
+    assert len(all_rssi) > 0, "Observer saw no advertisements"
+    assert all_rssi[0] == -39, f"Expected RSSI -39 (0 dBm TX), got {all_rssi[0]}"
+
+
+@pytest.mark.zephyr_sample("bluetooth/observer")
+@pytest.mark.circuitpy_drive({"code.py": BSIM_TX_POWER_LOW_CODE})
+def test_bsim_tx_power_low_rssi(bsim_phy, circuitpython, zephyr_sample):
+    """Verify low TX power reduces RSSI."""
+    observer = zephyr_sample
+
+    circuitpython.wait_until_done()
+
+    cp_output = circuitpython.serial.all_output
+    obs_output = observer.serial.all_output
+
+    assert "advertising low" in cp_output
+    assert "done" in cp_output
+
+    # With 40 dB channel attenuation and -20 dBm TX → RSSI ~ -59
+    rssi_pattern = re.compile(r"RSSI (-?\d+)\), type \d+, AD data len 12")
+    all_rssi = [int(m.group(1)) for m in rssi_pattern.finditer(obs_output)]
+    logger.info("RSSI values: %s", all_rssi)
+
+    assert len(all_rssi) > 0, "Observer saw no advertisements"
+    assert all_rssi[0] < -39, f"Expected lower RSSI with -20 dBm TX, got {all_rssi[0]}"

ports/zephyr-cp/tests/bsim/test_bsim_ble_peripheral.py

@@ -10,6 +10,7 @@
 BSIM_PERIPHERAL_CODE = """\
 import _bleio
 import time
+import sys
 
 adapter = _bleio.adapter
 
@@ -20,50 +21,27 @@
 adapter.start_advertising(advertisement, connectable=True)
 print("advertising", adapter.advertising)
 
-for _ in range(80):
-    if adapter.connected:
-        break
-    time.sleep(0.1)
-
-print("connected", adapter.connected, "advertising", adapter.advertising)
-
-for _ in range(80):
-    if not adapter.connected:
-        break
-    time.sleep(0.1)
-
-print("disconnected", adapter.connected, len(adapter.connections))
-"""
+was_connected = False
+timeout = time.monotonic() + 8.0
+while not was_connected and time.monotonic() < timeout:
+    time.sleep(0.01)
+    was_connected = adapter.connected
 
-BSIM_PERIPHERAL_CP_CENTRAL_PERIPHERAL_CODE = """\
-import _bleio
-import time
+if not was_connected:
+    print("connect timed out")
+    sys.exit(-1)
 
-adapter = _bleio.adapter
+print("connected", was_connected, "advertising", adapter.advertising)
 
-name = b"CPPERIPH"
-advertisement = bytes((2, 0x01, 0x06, len(name) + 1, 0x09)) + name
-
-print("peripheral start")
-adapter.start_advertising(advertisement, connectable=True)
-print("advertising", adapter.advertising)
-
-for _ in range(80):
-    if adapter.connected:
-        break
-    time.sleep(0.1)
-
-print("connected", adapter.connected, "advertising", adapter.advertising)
-
-for _ in range(80):
-    if not adapter.connected:
-        break
-    time.sleep(0.1)
+if was_connected:
+    timeout = time.monotonic() + 8.0
+    while adapter.connected and time.monotonic() < timeout:
+        time.sleep(0.1)
 
 print("disconnected", adapter.connected, len(adapter.connections))
 """
 
-BSIM_PERIPHERAL_CP_CENTRAL_CODE = """\
+BSIM_CENTRAL_CODE = """\
 import _bleio
 import time
 
@@ -86,9 +64,8 @@
 print("connected", connection.connected, adapter.connected, len(adapter.connections))
 connection.disconnect()
 
-for _ in range(40):
-    if not connection.connected and not adapter.connected:
-        break
+timeout = time.monotonic() + 4.0
+while (connection.connected or adapter.connected) and time.monotonic() < timeout:
     time.sleep(0.1)
 
 print("disconnected", connection.connected, adapter.connected, len(adapter.connections))
@@ -118,8 +95,8 @@ def test_bsim_peripheral_zephyr_central(bsim_phy, circuitpython, zephyr_sample):
 
 
 @pytest.mark.duration(14)
-@pytest.mark.circuitpy_drive({"code.py": BSIM_PERIPHERAL_CP_CENTRAL_CODE})
-@pytest.mark.circuitpy_drive({"code.py": BSIM_PERIPHERAL_CP_CENTRAL_PERIPHERAL_CODE})
+@pytest.mark.circuitpy_drive({"code.py": BSIM_PERIPHERAL_CODE})
+@pytest.mark.circuitpy_drive({"code.py": BSIM_CENTRAL_CODE})
 def test_bsim_peripheral_cp_central(bsim_phy, circuitpython1, circuitpython2):
     """Two CP instances: device 0 peripheral, device 1 central."""
     peripheral = circuitpython1