diff --git a/tools/cp-caps/README.md b/tools/cp-caps/README.md index dc97564e0..c6a5c761e 100644 --- a/tools/cp-caps/README.md +++ b/tools/cp-caps/README.md @@ -41,36 +41,23 @@ $ git clone git@github.com:openthread/ot-nrf528xx.git $ cd ot-nrf528xx/ $ git submodule update --init $ ./script/bootstrap -$ ./script/build nrf52840 UART_trans -DOT_DIAGNOSTIC=ON -DOT_CSL_RECEIVER=ON -DOT_LINK_METRICS_INITIATOR=ON -DOT_LINK_METRICS_SUBJECT=ON +$ ./script/build nrf52840 UART_trans -DOT_DIAGNOSTIC=ON -DOT_CSL_RECEIVER=ON -DOT_LINK_METRICS_INITIATOR=ON -DOT_LINK_METRICS_SUBJECT=ON -DOT_WAKEUP_COORDINATOR=ON $ arm-none-eabi-objcopy -O ihex build/bin/ot-cli-ftd ot-cli-ftd.hex $ nrfjprog -f nrf52 --chiperase --program ot-cli-ftd.hex --reset ``` ## Test Commands -### Help - -Show help info. +### Help Info ```bash -$ python3 ./tools/cp-caps/rcp_caps_test.py -h -usage: rcp_caps_test.py [-h] [-c] [-l] [-d] [-f] [-p] [-t] [-T] [-v] [-D] +usage: [device configurations] python3 -m unittest -q [test cases] -This script is used for testing RCP capabilities. +This unittest is used for testing RCP capabilities. -options: - -h, --help show this help message and exit - -c, --csl test whether the RCP supports CSL transmitter - -l, --link-metrics test whether the RCP supports link metrics - -d, --diag-commands test whether the RCP supports all diag commands - -f, --frame-format test whether the RCP supports 802.15.4 frames of all formats - -p, --data-poll test whether the RCP supports data poll - -t, --throughput test Thread network 1-hop throughput - -T, --tx-info test mTxInfo field of the radio frame - -v, --version output version - -D, --debug output debug information - -Device Interfaces: +Device configurations: + ADB_KEY= Full path to the adb key + DEBUG=on Enable the debug information DUT_ADB_TCP= Connect to the DUT via adb tcp DUT_ADB_USB= Connect to the DUT via adb usb DUT_CLI_SERIAL= Connect to the DUT via cli serial port @@ -78,15 +65,35 @@ Device Interfaces: REF_ADB_USB= Connect to the reference device via adb usb REF_CLI_SERIAL= Connect to the reference device via cli serial port REF_SSH= Connect to the reference device via ssh - ADB_KEY= Full path to the adb key -Example: - DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -d +Test cases: + test_csl test whether the RCP supports CSL transmitter + test_data_poll test whether the RCP supports data poll + test_diag_commands test whether the RCP supports all diag commands + test_frame_formats test whether the RCP supports 802.15.4 frames of all formats + test_link_metrics test whether the RCP supports link metrics + test_radio_frame_tx_info test mTxInfo field of the radio frame + test_throughput test Thread network 1-hop throughput + +Examples: + +Run test cases of the specified categories: + DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest -q test_csl test_data_poll + +Run all test cases: + DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest discover -q -s ./ -p 'test_*.py' + +Run specified test case: + DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest -q test_diag_commands.TestDiagCommands.test_diag_channel + +These commands should be run in the path `tools/cp-caps`. ``` +> Note: If you get an error of `LIBUSB_ERROR_BUSY` when you are using the ADB USB interface, please run the command `adb kill-server` to kill the ADB server. + ### Test Diag Commands -The option `-d` or `--diag-commands` tests all diag commands. +The test case `test_diag_commands` tests all diag commands. Following environment variables are used to configure diag command parameters: @@ -94,147 +101,143 @@ Following environment variables are used to configure diag command parameters: - DUT_DIAG_RAW_POWER_SETTING: Diag raw power setting value. The default value is `112233` if it is not set. - DUT_DIAG_POWER: Diag power value. The default value is `10` if it is not set. -> Note: If you meet the error `LIBUSB_ERROR_BUSY` when you are using the ADB usb interface, please run the command `adb kill-server` to kill the adb server. - ```bash -$ DUT_ADB_USB=1269UCKFZTAM95OR REF_CLI_SERIAL=/dev/ttyACM0 DUT_DIAG_GPIO=2 DUT_DIAG_RAW_POWER_SETTING=44556688 DUT_DIAG_POWER=11 python3 ./tools/cp-caps/rcp_caps_test.py -d +tools/cp-caps$ DUT_ADB_USB=1269UCKFZTAM95OR REF_CLI_SERIAL=/dev/ttyACM0 DUT_DIAG_GPIO=2 DUT_DIAG_RAW_POWER_SETTING=44556688 DUT_DIAG_POWER=11 python3 -m unittest -q test_diag_commands diag channel --------------------------------------------- OK diag channel 20 ------------------------------------------ OK +diag cw start -------------------------------------------- OK +diag cw stop --------------------------------------------- OK +diag echo 0123456789 ------------------------------------- OK +diag echo -n 10 ------------------------------------------ OK +diag frame 00010203040506070809 -------------------------- OK +diag gpio mode 2 ----------------------------------------- OK +diag gpio mode 2 in -------------------------------------- OK +diag gpio mode 2 out ------------------------------------- OK +diag gpio get 2 ------------------------------------------ OK +diag gpio set 2 0 ---------------------------------------- OK +diag gpio set 2 1 ---------------------------------------- OK diag power ----------------------------------------------- OK diag power 11 -------------------------------------------- OK -diag radio sleep ----------------------------------------- OK +diag powersettings --------------------------------------- NotSupported +diag powersettings 20 ------------------------------------ NotSupported diag radio receive --------------------------------------- OK +diag radio sleep ----------------------------------------- OK diag radio state ----------------------------------------- OK +diag rawpowersetting enable ------------------------------ NotSupported +diag rawpowersetting 44556688 ---------------------------- NotSupported +diag rawpowersetting ------------------------------------- NotSupported +diag rawpowersetting disable ----------------------------- NotSupported diag repeat 10 64 ---------------------------------------- OK diag repeat stop ----------------------------------------- OK diag send 100 64 ----------------------------------------- OK diag stats ----------------------------------------------- OK diag stats clear ----------------------------------------- OK -diag frame 00010203040506070809 -------------------------- OK -diag echo 0123456789 ------------------------------------- OK -diag echo -n 10 ------------------------------------------ OK -diag cw start -------------------------------------------- OK -diag cw stop --------------------------------------------- OK -diag stream start ---------------------------------------- OK -diag stream stop ----------------------------------------- OK -diag stats ----------------------------------------------- OK -diag stats clear ----------------------------------------- OK -diag rawpowersetting enable ------------------------------ NotSupported -diag rawpowersetting 44556688 ---------------------------- NotSupported -diag rawpowersetting ------------------------------------- NotSupported -diag rawpowersetting disable ----------------------------- NotSupported -diag powersettings --------------------------------------- NotSupported -diag powersettings 20 ------------------------------------ NotSupported -diag gpio mode 2 ----------------------------------------- NotSupported -diag gpio mode 2 in -------------------------------------- NotSupported -diag gpio mode 2 out ------------------------------------- NotSupported -diag gpio get 2 ------------------------------------------ NotSupported -diag gpio set 2 0 ---------------------------------------- NotSupported -diag gpio set 2 1 ---------------------------------------- NotSupported +diag stream start ---------------------------------------- NotSupported +diag stream stop ----------------------------------------- NotSupported ``` ### Test CSL Transmitter -The option `-c` or `--csl` tests whether the RCP supports the CSL transmitter. +The test case `test_csl` tests whether the RCP supports the CSL transmitter. ```bash -$ DUT_ADB_USB=TW69UCKFZTGM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -c +tools/cp-caps$ DUT_ADB_USB=TW69UCKFZTGM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest -q test_csl CSL Transmitter ------------------------------------------ OK ``` ### Test Data Poll -The option `-p` or `--data-poll` tests whether the RCP supports data poll. +The test case `test_data_poll` tests whether the RCP supports data poll. ```bash -$ DUT_ADB_USB=1269UCKFZTAM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -p -Data Poll Parent ----------------------------------------- OK +tools/cp-caps$ DUT_ADB_USB=TW69UCKFZTGM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest -q test_data_poll Data Poll Child ------------------------------------------ OK +Data Poll Parent ----------------------------------------- OK ``` ### Test Link Metrics -The option `-l` or `--link-metrics` tests whether the RCP supports link metrics. +The test case `test_link_metrics` tests whether the RCP supports link metrics. ```bash -$ DUT_ADB_USB=1269UCKFZTAM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -l +tools/cp-caps$ DUT_ADB_USB=TW69UCKFZTGM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest -q test_link_metrics Link Metrics Initiator ----------------------------------- OK Link Metrics Subject ------------------------------------- OK ``` ### Test Throughput -The option `-t` or `--throughput` tests the Thread network 1-hop throughput of the DUT. +The test case `test_throughput` tests the Thread network 1-hop throughput of the DUT. ```bash -$ DUT_ADB_USB=1269UCKFZTAM95OR REF_ADB_USB=44061HFAG01AQK python3 ./tools/cp-caps/rcp_caps_test.py -t +tools/cp-caps$ DUT_ADB_USB=TW69UCKFZTGM95OR REF_ADB_USB=44061HFAG01AQK python3 -m unittest -q test_throughput Throughput ----------------------------------------------- 75.6 Kbits/sec ``` ### Test Frame Format -The option `-f` or `--frame-format` tests whether the RCP supports sending and receiving 802.15.4 frames of all formats. +The test case `test_frame_formats` tests whether the RCP supports sending and receiving 802.15.4 frames of all formats. ```bash -$ DUT_ADB_USB=1269UCKFZTAM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -f -TX ver:2003,Cmd,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -RX ver:2003,Cmd,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -TX ver:2003,Bcon,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:30 ---------------- OK -RX ver:2003,Bcon,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:30 ---------------- OK -TX ver:2003,MP,noseq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie[ren con],plen:0 --------- OK -RX ver:2003,MP,noseq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie[ren con],plen:0 --------- OK -TX ver:2006,Cmd,seq,dst[addr:short,pan:id],src[addr:short,pan:no],sec:l5,ie:no,plen:0 -------------- OK -RX ver:2006,Cmd,seq,dst[addr:short,pan:id],src[addr:short,pan:no],sec:l5,ie:no,plen:0 -------------- OK -TX ver:2006,Cmd,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie:no,plen:0 ---------------- OK -RX ver:2006,Cmd,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie:no,plen:0 ---------------- OK -TX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 --------------- OK -RX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 --------------- OK -TX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ------------- OK -RX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ------------- OK -TX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -RX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -TX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ---------------- OK -RX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ---------------- OK -TX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 ------------------- OK -RX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 ------------------- OK -TX ver:2015,Data,seq,dst[addr:no,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ------------------- OK -RX ver:2015,Data,seq,dst[addr:no,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ------------------- OK -TX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -RX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -TX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -RX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 ----------------- OK -TX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:0 ----------------- OK -RX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:0 ----------------- OK -TX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 ----------------- OK -RX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 ----------------- OK -TX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:no,ie:no,plen:0 --------------- OK -RX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:no,ie:no,plen:0 --------------- OK -TX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 --------------- OK -RX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 --------------- OK -TX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ------------- OK -RX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ------------- OK -TX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 -------------- OK -RX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 -------------- OK -TX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 -------------- OK -RX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 -------------- OK -TX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie[csl],plen:0 ----------- OK -RX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie[csl],plen:0 ----------- OK -TX ver:2015,Data,noseq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ----------- OK -RX ver:2015,Data,noseq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ----------- OK +tools/cp-caps$ DUT_ADB_USB=TW69UCKFZTGM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest -q test_frame_formats +1 TX ver:2003,Cmd,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 --------------- OK +1 RX ver:2003,Cmd,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 --------------- OK +2 TX ver:2003,Bcon,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:30 -------------- OK +2 RX ver:2003,Bcon,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:30 -------------- OK +3 TX ver:2003,MP,noseq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie[ren con],plen:0 ------- OK +3 RX ver:2003,MP,noseq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie[ren con],plen:0 ------- OK +4 TX ver:2006,Cmd,seq,dst[addr:short,pan:id],src[addr:short,pan:no],sec:l5,ie:no,plen:0 ------------ OK +4 RX ver:2006,Cmd,seq,dst[addr:short,pan:id],src[addr:short,pan:no],sec:l5,ie:no,plen:0 ------------ OK +5 TX ver:2006,Cmd,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie:no,plen:0 -------------- OK +5 RX ver:2006,Cmd,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie:no,plen:0 -------------- OK +6 TX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 ------------- OK +6 RX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 ------------- OK +7 TX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ----------- OK +7 RX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ----------- OK +8 TX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 --------------- OK +8 RX ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 --------------- OK +9 TX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 -------------- OK +9 RX ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 -------------- OK +10 TX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 ---------------- OK +10 RX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 ---------------- OK +11 TX ver:2015,Data,seq,dst[addr:no,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ---------------- OK +11 RX ver:2015,Data,seq,dst[addr:no,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 ---------------- OK +12 TX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 -------------- OK +12 RX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0 -------------- OK +13 TX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 -------------- OK +13 RX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0 -------------- OK +14 TX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:0 -------------- OK +14 RX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:0 -------------- OK +15 TX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 -------------- OK +15 RX ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 -------------- OK +16 TX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:no,ie:no,plen:0 ------------ OK +16 RX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:no,ie:no,plen:0 ------------ OK +17 TX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 ------------ OK +17 RX ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0 ------------ OK +18 TX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ---------- OK +18 RX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ---------- OK +19 TX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 ----------- OK +19 RX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0 ----------- OK +20 TX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ----------- OK +20 RX ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 ----------- OK +21 TX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie[csl],plen:0 -------- OK +21 RX ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie[csl],plen:0 -------- OK +22 TX ver:2015,Data,noseq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 -------- OK +22 RX ver:2015,Data,noseq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0 -------- OK ``` ### Test mTxInfo field of the radio frame. -The option `-T` or `--tx-info` tests whether the RCP supports the mTxInfo field of the radio frame. +The test case `test_radio_frame_tx_info` tests whether the RCP supports the mTxInfo field of the radio frame. ```bash -$ DUT_ADB_USB=1269UCKFZTAM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 ./tools/cp-caps/rcp_caps_test.py -T -mIsSecurityProcessed=True -------------------------------- OK -mIsSecurityProcessed=False ------------------------------- OK -mTxDelayBaseTime=now,mTxDelay=500000 --------------------- OK -mRxChannelAfterTxDone ------------------------------------ OK +tools/cp-caps$ DUT_ADB_USB=TW69UCKFZTGM95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 -m unittest -q test_radio_frame_tx_info mCsmaCaEnabled=0 ----------------------------------------- OK mCsmaCaEnabled=1 ----------------------------------------- OK -mMaxCsmaBackoffs=0 --------------------------------------- OK (6 ms) -mMaxCsmaBackoffs=100 ------------------------------------- OK (560 ms) +mIsSecurityProcessed=True -------------------------------- OK +mIsSecurityProcessed=False ------------------------------- OK +mMaxCsmaBackoffs=0 --------------------------------------- OK (12 ms) +mMaxCsmaBackoffs=100 ------------------------------------- OK (541 ms) +mRxChannelAfterTxDone ------------------------------------ OK +mTxDelayBaseTime=now,mTxDelay=500000 --------------------- OK ``` diff --git a/tools/cp-caps/device_manager.py b/tools/cp-caps/device_manager.py new file mode 100644 index 000000000..77b42b7e1 --- /dev/null +++ b/tools/cp-caps/device_manager.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import logging +import os +import sys +import threading +import queue + +from dataclasses import dataclass +from typing import Optional + +import otci +from otci import OTCI +from otci.errors import ExpectLineTimeoutError, InvalidArgumentsError + +CP_CAPABILITY_VERSION = "0.1.1-dev" + +logging.basicConfig(level=logging.WARNING) + + +@dataclass +class Frame: + """Represents a Thread radio frame. + + Attributes: + name: The description of the frame. + tx_frame: The psdu of the frame that is to be sent. + dst_address: The destination MAC address of the `tx_frame`. It is used by the receiver to filter + out the `tx_frame`. + is_security_processed: The value of the `otRadioFrame.mInfo.mTxInfo.mIsSecurityProcessed` field. + If it is set to False, the `active_dataset` and `src_ext_address` should also be set for the + radio driver to encrypt the `tx_frame`. + expect_rx_frame: The frame expected to be received. The expected frame should be the same as the + `tx_frame` if the `expect_rx_frame` is set to None. + active_dataset: The active dataset. + src_ext_address: The source extended MAC address of the transmitter. + """ + name: str + tx_frame: str + dst_address: str + is_security_processed: Optional[bool] = True + expect_rx_frame: Optional[str] = None + active_dataset: Optional[str] = None + src_ext_address: Optional[str] = None + + +class DeviceManager(object): + """Provides the DUT, reference device and common methods for test cases.""" + + DEFAULT_FORMAT_ALIGN_LENGTH = 58 # The default formatted string alignment length + + def __init__(self): + self.__dut = self.__connect_dut() + self.__ref = self.__connect_reference_device() + + if self.__get_debug_enabled(): + logger = logging.getLogger() + logger.setLevel(logging.DEBUG) + + @property + def dut(self) -> OTCI: + return self.__dut + + @property + def ref(self) -> OTCI: + return self.__ref + + def output_test_result_string(self, name: str, value: str, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH): + prefix = (name + ' ').ljust(align_length, '-') + print(f'{prefix} {value}') + + def output_test_result_bool(self, name: str, value: bool, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH): + self.output_test_result_string(name, 'OK' if value else 'NotSupported', align_length) + + def get_default_dataset(self): + return self.__dut.create_dataset(channel=20, network_key='00112233445566778899aabbccddcafe') + + def send_formatted_frames_retries(self, sender: OTCI, receiver: OTCI, frame: Frame, max_send_retries: int = 5): + for i in range(0, max_send_retries): + if self.__send_formatted_frame(sender, receiver, frame): + return True + + return False + + # + # Private methods + # + def __get_debug_enabled(self) -> bool: + enabled = os.getenv('DEBUG', None) + return enabled == 'on' + + def __get_adb_key(self) -> Optional[str]: + return os.getenv('ADB_KEY', None) + + def __connect_dut(self) -> OTCI: + if os.getenv('DUT_ADB_TCP'): + node = otci.connect_otbr_adb_tcp(os.getenv('DUT_ADB_TCP'), adb_key=self.__get_adb_key()) + elif os.getenv('DUT_ADB_USB'): + node = otci.connect_otbr_adb_usb(os.getenv('DUT_ADB_USB'), adb_key=self.__get_adb_key()) + elif os.getenv('DUT_CLI_SERIAL'): + node = otci.connect_cli_serial(os.getenv('DUT_CLI_SERIAL')) + elif os.getenv('DUT_SSH'): + node = otci.connect_otbr_ssh(os.getenv('DUT_SSH')) + else: + raise InvalidArgumentsError( + "Please set DUT_ADB_TCP, DUT_ADB_USB, DUT_CLI_SERIAL or DUT_SSH to connect to the DUT device.") + + return node + + def __connect_reference_device(self) -> OTCI: + if os.getenv('REF_CLI_SERIAL'): + node = otci.connect_cli_serial(os.getenv('REF_CLI_SERIAL')) + elif os.getenv('REF_SSH'): + node = otci.connect_otbr_ssh(os.getenv('REF_SSH')) + elif os.getenv('REF_ADB_USB'): + node = otci.connect_otbr_adb_usb(os.getenv('REF_ADB_USB'), adb_key=self.__get_adb_key()) + else: + raise InvalidArgumentsError( + "Please set REF_CLI_SERIAL, REF_SSH or REF_ADB_USB to connect to the reference device.") + + return node + + def __send_formatted_frame(self, sender: OTCI, receiver: OTCI, frame: Frame): + # When `is_security_processed` is False, the frame may need the radio driver + # to encrypt the frame. Here sets the active dataset and the MAC source address for the + # radio driver to encrypt the frame. + if frame.is_security_processed is False: + if frame.active_dataset is None or frame.src_ext_address is None: + raise InvalidArgumentsError( + "When the 'is_security_processed' is 'False', the 'active_dataset' and 'src_ext_address' must be set" + ) + # Do the factory reset to set the frame counter to 0. + sender.factory_reset() + receiver.factory_reset() + sender.set_dataset_bytes('active', frame.active_dataset) + sender.set_extaddr(frame.src_ext_address) + + sender.diag_start() + receiver.diag_start() + + channel = 11 + num_sent_frames = 1 + + sender.diag_set_channel(channel) + receiver.diag_set_channel(channel) + receiver.diag_radio_receive() + receiver.diag_set_radio_receive_filter_dest_mac_address(frame.dst_address) + receiver.diag_enable_radio_receive_filter() + + result_queue = queue.Queue() + receive_task = threading.Thread(target=self.__radio_receive_task, + args=(receiver, num_sent_frames, result_queue), + daemon=True) + receive_task.start() + + sender.wait(0.1) + + sender.diag_frame(frame.tx_frame, + is_security_processed=frame.is_security_processed, + max_csma_backoffs=4, + max_frame_retries=4, + csma_ca_enabled=True) + sender.diag_send(num_sent_frames, is_async=False) + + receive_task.join() + + if result_queue.empty(): + ret = False # No frame is received. + else: + received_frames = result_queue.get() + if len(received_frames) != num_sent_frames: + ret = False + else: + # The radio driver may not append the FCF field to the received frame. Do not check the FCF field here. + FCF_LENGTH = 4 + expect_frame = frame.expect_rx_frame or frame.tx_frame + ret = expect_frame[:-FCF_LENGTH] == received_frames[0]['psdu'][:-FCF_LENGTH] + + if ret: + sender.diag_stop() + receiver.diag_stop() + else: + # The command `diag radio receive ` may fail to receive specified number of frames in default + # timeout time. In this case, the diag module still in `sync` mode, and it only allows users to run the + # command `factoryreset` to terminate the test. + sender.factory_reset() + receiver.factory_reset() + + return ret + + def __radio_receive_task(self, receiver: OTCI, number: int, result_queue: queue): + try: + receiver.set_execute_command_retry(0) + result = receiver.diag_radio_receive_number(number) + except ExpectLineTimeoutError: + pass + else: + result_queue.put(result) + finally: + receiver.set_execute_command_retry(OTCI.DEFAULT_EXEC_COMMAND_RETRY) diff --git a/tools/cp-caps/rcp_caps_test.py b/tools/cp-caps/rcp_caps_test.py deleted file mode 100644 index ef12e91dd..000000000 --- a/tools/cp-caps/rcp_caps_test.py +++ /dev/null @@ -1,1093 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (c) 2024, The OpenThread Authors. -# All rights reserved. -# -# Redistribution and use in source and binary forms, with or without -# modification, are permitted provided that the following conditions are met: -# 1. Redistributions of source code must retain the above copyright -# notice, this list of conditions and the following disclaimer. -# 2. Redistributions in binary form must reproduce the above copyright -# notice, this list of conditions and the following disclaimer in the -# documentation and/or other materials provided with the distribution. -# 3. Neither the name of the copyright holder nor the -# names of its contributors may be used to endorse or promote products -# derived from this software without specific prior written permission. -# -# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" -# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE -# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE -# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE -# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR -# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF -# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS -# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN -# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) -# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE -# POSSIBILITY OF SUCH DAMAGE. -# - -import argparse -import logging -import os -import sys -import textwrap -import threading -import time -import queue - -from dataclasses import dataclass -from typing import List, Optional - -import otci -from otci import OTCI -from otci.types import Ip6Addr -from otci.errors import ExpectLineTimeoutError, InvalidArgumentsError - -CP_CAPABILITY_VERSION = "0.1.1-dev" - -logging.basicConfig(level=logging.WARNING) - - -class RcpCaps(object): - """ - This class represents an OpenThread RCP capability test instance. - """ - - DEFAULT_FORMAT_ALIGN_LENGTH = 58 # The default formatted string alignment length - - def __init__(self): - self.__dut = self.__connect_dut() - self.__ref = self.__connect_reference_device() - - def test_diag_commands(self): - """Test all diag commands.""" - self.__dut.factory_reset() - self.__ref.factory_reset() - - ret = self.__dut.is_command_supported('diag start') - if ret is False: - print('All diag commands are not supported') - return - - self.__dut.diag_start() - self.__ref.diag_start() - - self.__test_diag_channel() - self.__test_diag_power() - self.__test_diag_radio() - self.__test_diag_repeat() - self.__test_diag_send() - self.__test_diag_frame() - - # The self.__test_diag_frame will factoryreset the device - self.__dut.diag_start() - self.__ref.diag_start() - - self.__test_diag_echo() - self.__test_diag_utils() - self.__test_diag_rawpowersetting() - self.__test_diag_powersettings() - self.__test_diag_gpio_mode() - self.__test_diag_gpio_value() - - self.__ref.diag_stop() - self.__dut.diag_stop() - - @dataclass - class Frame: - """Represents a thread radio frame. - - Attributes: - name: The description of the frame. - tx_frame: The psdu of the frame that to be sent. - dst_address: The destination MAC address of the tx_frame. It is used by the receiver to filter - out the tx_frame. - is_security_processed: The value of the otRadioFrame.mInfo.mTxInfo.mIsSecurityProcessed field. - If it is set to False, the active_dataset and src_ext_address are should also be set for the - radio driver to encrypt the tx_frame. - expect_rx_frame: The frame expected to be received. The frame expected to be received should be - the same as the tx_frame if the expect_rx_frame is set to None. - active_dataset: The active dataset. - src_ext_address: The source extended MAC address of the transmitter. - """ - name: str - tx_frame: str - dst_address: str - is_security_processed: Optional[bool] = True - expect_rx_frame: Optional[str] = None - active_dataset: Optional[str] = None - src_ext_address: Optional[str] = None - - def test_frame_format(self): - """Test whether the DUT supports sending and receiving 802.15.4 frames of all formats.""" - frames = [ - self.Frame(name='ver:2003,Cmd,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', - tx_frame='030800ffffffff070000', - dst_address='0xffff'), - self.Frame( - name='ver:2003,Bcon,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:30', - tx_frame='00c000eeee0102030405060708ff0f000003514f70656e54687265616400000000000001020304050607080000', - dst_address='-'), - self.Frame( - name='ver:2003,MP,noseq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie[ren con],plen:0', - tx_frame= - 'fd87dddd1020304050607080010203040506070815000000007265616401820ee80305009bb8ea011c807aa1120000', - dst_address='8070605040302010'), - self.Frame(name='ver:2006,Cmd,seq,dst[addr:short,pan:id],src[addr:short,pan:no],sec:l5,ie:no,plen:0', - tx_frame='4b9800ddddaaaabbbb0d0000000001043daa1aea0000', - dst_address='0xaaaa'), - self.Frame(name='ver:2006,Cmd,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie:no,plen:0', - tx_frame='4bdc00dddd102030405060708001020304050607080d000000000104483cb8a90000', - dst_address='8070605040302010'), - self.Frame(name='ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0', - tx_frame='41dc00dddd102030405060708001020304050607080000', - dst_address='8070605040302010'), - self.Frame(name='ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', - tx_frame='019800ddddaaaaeeeebbbb0000', - dst_address='0xaaaa'), - self.Frame(name='ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', - tx_frame='011c00dddd10203040506070800000', - dst_address='8070605040302010'), - self.Frame(name='ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', - tx_frame='011800ddddaaaa0000', - dst_address='0xaaaa'), - self.Frame(name='ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0', - tx_frame='0120000000', - dst_address='-'), - self.Frame(name='ver:2015,Data,seq,dst[addr:no,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', - tx_frame='412000dddd0000', - dst_address='-'), - self.Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', - tx_frame='012c00dddd10203040506070800000', - dst_address='8070605040302010'), - self.Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0', - tx_frame='412c0010203040506070800000', - dst_address='8070605040302010'), - self.Frame(name='ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:0', - tx_frame='01e000eeee01020304050607080000', - dst_address='-'), - self.Frame(name='ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0', - tx_frame='41e00001020304050607080000', - dst_address='-'), - self.Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:no,ie:no,plen:0', - tx_frame='01ec00dddd102030405060708001020304050607080000', - dst_address='8070605040302010'), - self.Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0', - tx_frame='41ec00102030405060708001020304050607080000', - dst_address='8070605040302010'), - self.Frame(name='ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', - tx_frame='01a800ddddaaaaeeeebbbb0000', - dst_address='0xaaaa'), - self.Frame(name='ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0', - tx_frame='01e800ddddaaaaeeee01020304050607080000', - dst_address='0xaaaa'), - self.Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', - tx_frame='01ac00dddd1020304050607080eeeebbbb0000', - dst_address='8070605040302010'), - self.Frame(name='ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie[csl],plen:0', - tx_frame='01aa00ddddaaaaeeeebbbb040dc800e8030000', - dst_address='0xaaaa'), - self.Frame(name='ver:2015,Data,noseq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', - tx_frame='01a9ddddaaaaeeeebbbb0000', - dst_address='0xaaaa'), - ] - - for frame in frames: - ret = self.__test_send_formatted_frames_retries(self.__dut, self.__ref, frame) - self.__output_format_bool(f'TX {frame.name}', ret, align_length=100) - ret = self.__test_send_formatted_frames_retries(self.__ref, self.__dut, frame) - self.__output_format_bool(f'RX {frame.name}', ret, align_length=100) - - def test_csl(self): - """Test whether the DUT supports CSL transmitter.""" - self.__dataset = self.__get_default_dataset() - self.__test_csl_transmitter() - - def test_data_poll(self): - """Test whether the DUT supports data poll parent and child.""" - self.__dataset = self.__get_default_dataset() - self.__test_data_poll_parent() - self.__test_data_poll_child() - - def test_throughput(self): - """Test Thread network 1 hop throughput.""" - if not self.__dut.support_iperf3(): - print("The DUT doesn't support the tool iperf3") - return - - if not self.__ref.support_iperf3(): - print("The reference device doesn't support the tool iperf3") - return - - bitrate = 90000 - length = 1232 - transmit_time = 30 - max_wait_time = 30 - timeout = transmit_time + max_wait_time - - self.__dut.factory_reset() - self.__ref.factory_reset() - - dataset = self.__get_default_dataset() - - self.__dut.join(dataset) - self.__dut.wait_for('state', 'leader') - - self.__ref.set_router_selection_jitter(1) - self.__ref.join(dataset) - self.__ref.wait_for('state', ['child', 'router']) - - ref_mleid = self.__ref.get_ipaddr_mleid() - - ref_iperf3_server = threading.Thread(target=self.__ref_iperf3_server_task, - args=(ref_mleid, timeout), - daemon=True) - ref_iperf3_server.start() - self.__dut.wait(1) - - results = self.__dut.iperf3_client(host=ref_mleid, bitrate=bitrate, transmit_time=transmit_time, length=length) - ref_iperf3_server.join() - - if not results: - print('Failed to run the iperf3') - return - - self.__output_format_string('Throughput', self.__bitrate_to_string(results['receiver']['bitrate'])) - - def test_link_metrics(self): - """Test whether the DUT supports Link Metrics Initiator and Subject.""" - self.__dataset = self.__get_default_dataset() - - self.__dut.factory_reset() - self.__ref.factory_reset() - - self.__dut.join(self.__dataset) - self.__dut.wait_for('state', 'leader') - - self.__ref.join(self.__dataset) - self.__ref.wait_for('state', ['child', 'router']) - - test_case = 'Link Metrics Initiator' - ref_linklocal_address = self.__ref.get_ipaddr_linklocal() - ret = self.__run_link_metrics_test_commands(initiator=self.__dut, subject_address=ref_linklocal_address) - self.__output_format_bool(test_case, ret) - - test_case = 'Link Metrics Subject' - dut_linklocal_address = self.__dut.get_ipaddr_linklocal() - ret = self.__run_link_metrics_test_commands(initiator=self.__ref, subject_address=dut_linklocal_address) - self.__output_format_bool(test_case, ret) - - self.__ref.leave() - self.__dut.leave() - - def test_radio_frame_tx_info(self): - self.__test_radio_frame_tx_info_is_security_processed() - self.__test_radio_frame_tx_info_tx_delay() - self.__test_radio_frame_tx_info_rx_channel_after_tx_done() - self.__test_radio_frame_tx_info_csma_ca_enabled() - self.__test_radio_frame_tx_info_max_csma_backoffs() - - # - # Private methods - # - def __test_radio_frame_tx_info_is_security_processed(self): - self.__dut.factory_reset() - active_dataset = self.__get_default_dataset() - - frames = [ - self.Frame( - name='mIsSecurityProcessed=True', - tx_frame='09ec00dddd102030405060708001020304050607080d0000000001db622c220fde64408db9128c93d50000', - dst_address='8070605040302010', - is_security_processed=True), - self.Frame( - name='mIsSecurityProcessed=False', - tx_frame='09ec00dddd102030405060708001020304050607080d000000000000010203040506070809000000000000', - expect_rx_frame= - '09ec00dddd102030405060708001020304050607080d0000000001db622c220fde64408db9128c93d50000', - is_security_processed=False, - dst_address='8070605040302010', - active_dataset=active_dataset, - src_ext_address='0807060504030201'), - ] - - for frame in frames: - ret = self.__test_send_formatted_frame(self.__dut, self.__ref, frame) - self.__output_format_bool(frame.name, ret) - - def __test_radio_frame_tx_info_tx_delay(self): - self.__dut.factory_reset() - self.__ref.factory_reset() - - # Enable the IPv6 interface to force the host and the RCP to synchronize the radio time. - self.__dut.set_dataset_bytes('active', self.__get_default_dataset()) - self.__dut.ifconfig_up() - self.__dut.wait(0.5) - self.__dut.ifconfig_down() - - self.__dut.diag_start() - self.__ref.diag_start() - - channel = 11 - packets = 1 - dut_tx_delay_sec = 0.5 - dut_tx_delay_us = int(dut_tx_delay_sec * 1000000) - ref_rx_delay_sec = dut_tx_delay_sec / 2 - ref_address = 'dead00beefcafe01' - dut_tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' - - self.__dut.diag_set_channel(channel) - self.__ref.diag_set_channel(channel) - self.__ref.diag_radio_receive() - self.__ref.diag_set_radio_receive_filter_dest_mac_address(ref_address) - self.__ref.diag_enable_radio_receive_filter() - - self.__dut.diag_frame(dut_tx_frame, tx_delay=dut_tx_delay_us) - self.__dut.diag_send(packets, is_async=True) - - self.__ref.wait(ref_rx_delay_sec) - stats = self.__ref.diag_get_stats() - ret = stats['received_packets'] == 0 - - if ret is True: - self.__ref.wait(ref_rx_delay_sec) - stats = self.__ref.diag_get_stats() - ret = stats['received_packets'] == 1 - - self.__output_format_bool(f'mTxDelayBaseTime=now,mTxDelay={dut_tx_delay_us}', ret) - - self.__ref.diag_stop() - self.__dut.diag_stop() - - def __test_radio_frame_tx_info_rx_channel_after_tx_done(self): - self.__dut.factory_reset() - self.__ref.factory_reset() - - self.__dut.diag_start() - self.__ref.diag_start() - - channel = 11 - num_sent_frames = 1 - rx_channel_after_tx_done = 25 - dut_address = 'dead00beefcafe01' - dut_tx_frame = '01ec00dddd02fecaefbe00adde01fecaefbe00adde000102030405060708090000' - ref_tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' - - self.__dut.diag_set_channel(channel) - self.__dut.diag_set_radio_receive_filter_dest_mac_address(dut_address) - self.__dut.diag_enable_radio_receive_filter() - self.__dut.diag_stats_clear() - - self.__dut.diag_frame(dut_tx_frame, rx_channel_after_tx_done=rx_channel_after_tx_done) - self.__dut.diag_send(num_sent_frames, is_async=False) - stats = self.__dut.diag_get_stats() - ret = stats['sent_success_packets'] == num_sent_frames - - if ret: - self.__ref.diag_set_channel(rx_channel_after_tx_done) - self.__ref.diag_frame(ref_tx_frame) - self.__ref.diag_send(num_sent_frames, is_async=False) - stats = self.__dut.diag_get_stats() - ret = stats['received_packets'] == num_sent_frames - - self.__ref.diag_stop() - self.__dut.diag_stop() - - self.__output_format_bool('mRxChannelAfterTxDone', ret) - - def __test_radio_frame_tx_info_csma_ca_enabled(self): - self.__dut.factory_reset() - self.__ref.factory_reset() - - self.__dut.diag_start() - self.__ref.diag_start() - - channel = 11 - num_sent_frames = 1 - tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' - - self.__dut.diag_set_channel(channel) - self.__ref.diag_set_channel(channel) - self.__ref.diag_cw_start() - - self.__dut.diag_stats_clear() - self.__dut.diag_frame(tx_frame, csma_ca_enabled=False) - self.__dut.diag_send(num_sent_frames, is_async=False) - dut_stats = self.__dut.diag_get_stats() - ret = dut_stats['sent_success_packets'] == num_sent_frames - self.__output_format_bool('mCsmaCaEnabled=0', ret) - - self.__dut.diag_stats_clear() - self.__dut.diag_frame(tx_frame, csma_ca_enabled=True) - self.__dut.diag_send(num_sent_frames, is_async=False) - dut_stats = self.__dut.diag_get_stats() - ret = dut_stats['sent_error_cca_packets'] == num_sent_frames - self.__output_format_bool('mCsmaCaEnabled=1', ret) - - self.__ref.diag_cw_stop() - self.__ref.diag_stop() - self.__dut.diag_stop() - - def __test_radio_frame_tx_info_max_csma_backoffs(self): - self.__dut.factory_reset() - self.__ref.factory_reset() - - self.__dut.diag_start() - self.__ref.diag_start() - - channel = 11 - num_sent_frames = 1 - tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' - - self.__dut.diag_set_channel(channel) - self.__ref.diag_set_channel(channel) - - self.__ref.diag_cw_start() - self.__dut.wait(0.05) - - # When the max_csma_backoffs is set to 0, the radio driver should skip backoff and do CCA once. - # The CCA time is 192 us. Theoretically, the `diag send` command should return after 192us. - # But considering the Spinel delay and the system IO delay, here sets the max_time_cost to 20 ms. - max_time_cost = 20 - max_csma_backoffs = 0 - self.__dut.diag_stats_clear() - self.__dut.diag_frame(tx_frame, csma_ca_enabled=True, max_frame_retries=0, max_csma_backoffs=max_csma_backoffs) - start_time = time.time() - self.__dut.diag_send(num_sent_frames, is_async=False) - end_time = time.time() - time_cost = int((end_time - start_time) * 1000) - ret = 'OK' if time_cost < max_time_cost else 'NotSupported' - self.__output_format_string(f'mMaxCsmaBackoffs={max_csma_backoffs}', f'{ret} ({time_cost} ms)') - - # Basic information for calculating the backoff time: - # aTurnaroundTime = 192 us - # aCCATime = 128 us - # backoffExponent = (macMinBe, macMaxBe) = (3, 5) - # backoffPeriod = random() % (1 << backoffExponent) - # backoff = backoffPeriod * aUnitBackoffPeriod = backoffPeriod * (aTurnaroundTime + aCCATime) - # = backoffPeriod * 320 us - # backoff = (random() % (1 << backoffExponent)) * 320us - # - # The max_csma_backoffs is set to 100 here, the `backoffExponent` will be set to 5 in most retries. - # backoff = (random() % 32) * 320us - # average_backoff = 16 * 320us = 5120 us - # total_backoff = average_backoff * 100 = 5120 us * 100 = 512 ms - # - # Here sets the max_time_cost to half of total_backoff. - # - max_time_cost = 256 - max_frame_retries = 0 - max_csma_backoffs = 100 - self.__dut.diag_frame(tx_frame, csma_ca_enabled=True, max_frame_retries=0, max_csma_backoffs=max_csma_backoffs) - start_time = time.time() - self.__dut.diag_send(num_sent_frames, is_async=False) - end_time = time.time() - time_cost = int((end_time - start_time) * 1000) - ret = 'OK' if time_cost > max_time_cost else 'NotSupported' - self.__output_format_string(f'mMaxCsmaBackoffs={max_csma_backoffs}', f'{ret} ({time_cost} ms)') - - self.__ref.diag_cw_stop() - self.__ref.diag_stop() - self.__dut.diag_stop() - - def __run_link_metrics_test_commands(self, initiator: OTCI, subject_address: Ip6Addr) -> bool: - seriesid = 1 - series_flags = 'ldra' - link_metrics_flags = 'qr' - probe_length = 10 - - if not initiator.linkmetrics_config_enhanced_ack_register(subject_address, link_metrics_flags): - return False - - if not initiator.linkmetrics_config_forward(subject_address, seriesid, series_flags, link_metrics_flags): - return False - - initiator.linkmetrics_probe(subject_address, seriesid, probe_length) - - results = initiator.linkmetrics_request_single(subject_address, link_metrics_flags) - if not ('lqi' in results.keys() and 'rssi' in results.keys()): - return False - - results = initiator.linkmetrics_request_forward(subject_address, seriesid) - if not ('lqi' in results.keys() and 'rssi' in results.keys()): - return False - - if not initiator.linkmetrics_config_enhanced_ack_clear(subject_address): - return False - - return True - - def __ref_iperf3_server_task(self, bind_address: str, timeout: int): - self.__ref.iperf3_server(bind_address, timeout=timeout) - - def __bitrate_to_string(self, bitrate: float): - units = ['bits/sec', 'Kbits/sec', 'Mbits/sec', 'Gbits/sec', 'Tbits/sec'] - unit_index = 0 - - while bitrate >= 1000 and unit_index < len(units) - 1: - bitrate /= 1000 - unit_index += 1 - - return f'{bitrate:.2f} {units[unit_index]}' - - def __get_default_dataset(self): - return self.__dut.create_dataset(channel=20, network_key='00112233445566778899aabbccddcafe') - - def __test_csl_transmitter(self): - packets = 10 - - self.__dut.factory_reset() - self.__ref.factory_reset() - - self.__dut.join(self.__dataset) - self.__dut.wait_for('state', 'leader') - - # Set the reference device as an SSED - self.__ref.set_mode('-') - self.__ref.config_csl(channel=15, period=320000, timeout=100) - self.__ref.join(self.__dataset) - self.__ref.wait_for('state', 'child') - - child_table = self.__dut.get_child_table() - ret = len(child_table) == 1 and child_table[1]['csl'] - - if ret: - ref_mleid = self.__ref.get_ipaddr_mleid() - result = self.__dut.ping(ref_mleid, count=packets, interval=1) - ret = result['transmitted_packets'] == result['received_packets'] == packets - - self.__dut.leave() - self.__ref.leave() - - self.__output_format_bool('CSL Transmitter', ret) - - def __test_data_poll_parent(self): - packets = 10 - - self.__dut.factory_reset() - self.__ref.factory_reset() - - self.__dut.join(self.__dataset) - self.__dut.wait_for('state', 'leader') - - # Set the reference device as an SED - self.__ref.set_mode('-') - self.__ref.set_poll_period(500) - self.__ref.join(self.__dataset) - self.__ref.wait_for('state', 'child') - - dut_mleid = self.__dut.get_ipaddr_mleid() - result = self.__ref.ping(dut_mleid, count=packets, interval=1) - - self.__dut.leave() - self.__ref.leave() - - ret = result['transmitted_packets'] == result['received_packets'] == packets - self.__output_format_bool('Data Poll Parent', ret) - - def __test_data_poll_child(self): - packets = 10 - - self.__dut.factory_reset() - self.__ref.factory_reset() - - self.__ref.join(self.__dataset) - self.__ref.wait_for('state', 'leader') - - # Set the DUT as an SED - self.__dut.set_mode('-') - self.__dut.set_poll_period(500) - self.__dut.join(self.__dataset) - self.__dut.wait_for('state', 'child') - - dut_mleid = self.__dut.get_ipaddr_mleid() - result = self.__ref.ping(dut_mleid, count=packets, interval=1) - - self.__dut.leave() - self.__ref.leave() - - ret = result['transmitted_packets'] == result['received_packets'] == packets - self.__output_format_bool('Data Poll Child', ret) - - def __test_diag_channel(self): - channel = 20 - commands = ['diag channel', f'diag channel {channel}'] - - if self.__support_commands(commands): - self.__dut.diag_set_channel(channel) - value = self.__dut.diag_get_channel() - ret = value == channel - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_power(self): - power = self.__get_dut_diag_power() - commands = ['diag power', f'diag power {power}'] - - if self.__support_commands(commands): - self.__dut.diag_set_power(power) - value = self.__dut.diag_get_power() - ret = value == power - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_radio(self): - commands = ['diag radio receive', 'diag radio sleep', 'diag radio state'] - - if self.__support_commands(commands): - self.__dut.diag_radio_receive() - receive_state = self.__dut.diag_get_radio_state() - self.__dut.wait(0.1) - self.__dut.diag_radio_sleep() - sleep_state = self.__dut.diag_get_radio_state() - - ret = sleep_state == 'sleep' and receive_state == 'receive' - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_gpio_value(self): - gpio = self.__get_dut_diag_gpio() - commands = [f'diag gpio get {gpio}', f'diag gpio set {gpio} 0', f'diag gpio set {gpio} 1'] - - if self.__support_commands(commands): - self.__dut.diag_set_gpio_value(gpio, 0) - value_0 = self.__dut.diag_get_gpio_value(gpio) - self.__dut.diag_set_gpio_value(gpio, 1) - value_1 = self.__dut.diag_get_gpio_value(gpio) - - ret = value_0 == 0 and value_1 == 1 - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_gpio_mode(self): - gpio = self.__get_dut_diag_gpio() - commands = [f'diag gpio mode {gpio}', f'diag gpio mode {gpio} in', f'diag gpio mode {gpio} out'] - - if self.__support_commands(commands): - self.__dut.diag_set_gpio_mode(gpio, 'in') - mode_in = self.__dut.diag_get_gpio_mode(gpio) - self.__dut.diag_set_gpio_value(gpio, 'out') - mode_out = self.__dut.diag_get_gpio_mode(gpio) - - ret = mode_in == 'in' and mode_out == 'out' - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_echo(self): - echo_msg = '0123456789' - cmd_diag_echo = f'diag echo {echo_msg}' - cmd_diag_echo_num = f'diag echo -n 10' - - if self.__dut.is_command_supported(cmd_diag_echo): - reply = self.__dut.diag_echo(echo_msg) - ret = reply == echo_msg - else: - ret = False - self.__output_format_bool(cmd_diag_echo, ret) - - if self.__dut.is_command_supported(cmd_diag_echo_num): - reply = self.__dut.diag_echo_number(10) - ret = reply == echo_msg - else: - ret = False - self.__output_format_bool(cmd_diag_echo_num, ret) - - def __test_diag_utils(self): - commands = [ - 'diag cw start', 'diag cw stop', 'diag stream start', 'diag stream stop', 'diag stats', 'diag stats clear' - ] - - for command in commands: - ret = self.__dut.is_command_supported(command) - self.__output_format_bool(command, ret) - - def __test_diag_rawpowersetting(self): - rawpowersetting = self.__get_dut_diag_raw_power_setting() - commands = [ - 'diag rawpowersetting enable', f'diag rawpowersetting {rawpowersetting}', 'diag rawpowersetting', - 'diag rawpowersetting disable' - ] - - if self.__support_commands(commands): - self.__dut.diag_enable_rawpowersetting() - self.__dut.diag_set_rawpowersetting(rawpowersetting) - reply = self.__dut.diag_get_rawpowersetting() - self.__dut.diag_disable_rawpowersetting() - - ret = reply == rawpowersetting - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_powersettings(self): - commands = ['diag powersettings', 'diag powersettings 20'] - - if self.__support_commands(commands): - powersettings = self.__dut.diag_get_powersettings() - ret = len(powersettings) > 0 - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_send(self): - packets = 100 - threshold = 80 - length = 64 - channel = 20 - commands = [f'diag send {packets} {length}', f'diag stats', f'diag stats clear'] - - if self.__support_commands(commands): - self.__dut.wait(1) - self.__dut.diag_set_channel(channel) - self.__ref.diag_set_channel(channel) - self.__ref.diag_radio_receive() - - self.__dut.diag_stats_clear() - self.__ref.diag_stats_clear() - - self.__dut.diag_send(packets, length) - self.__dut.wait(1) - dut_stats = self.__dut.diag_get_stats() - ref_stats = self.__ref.diag_get_stats() - - ret = dut_stats['sent_success_packets'] == packets and ref_stats['received_packets'] > threshold - else: - ret = False - - self.__output_results(commands, ret) - - def __test_diag_repeat(self): - delay = 10 - threshold = 80 - length = 64 - channel = 20 - cmd_diag_repeat = f'diag repeat {delay} {length}' - cmd_diag_repeat_stop = 'diag repeat stop' - commands = [cmd_diag_repeat, 'diag repeat stop', 'diag stats', 'diag stats clear'] - - if self.__support_commands(commands): - self.__dut.diag_set_channel(channel) - self.__ref.diag_set_channel(channel) - self.__ref.diag_radio_receive() - - self.__dut.diag_stats_clear() - self.__ref.diag_stats_clear() - - self.__dut.diag_repeat(delay, length) - self.__dut.wait(1) - self.__dut.diag_repeat_stop() - dut_stats = self.__dut.diag_get_stats() - ref_stats = self.__ref.diag_get_stats() - - ret = dut_stats['sent_success_packets'] > threshold and ref_stats['received_packets'] > threshold - else: - ret = False - - self.__output_format_bool(cmd_diag_repeat, ret) - self.__output_format_bool(cmd_diag_repeat_stop, ret) - - def __test_send_formatted_frames_retries(self, - sender: OTCI, - receiver: OTCI, - frame: Frame, - max_send_retries: int = 5): - for i in range(0, max_send_retries): - if self.__test_send_formatted_frame(sender, receiver, frame): - return True - - return False - - def __test_send_formatted_frame(self, sender: OTCI, receiver: OTCI, frame: Frame): - sender.factory_reset() - receiver.factory_reset() - - # When the 'is_security_processed' is False, it means the frame may need the radio driver - # to encrypt the frame. Here sets the active dataset and the MAC source address for the - # radio driver to encrypt the frame. - if frame.is_security_processed is False: - if frame.active_dataset is None or frame.src_ext_address is None: - raise InvalidArgumentsError( - "When the 'is_security_processed' is 'False', the 'active_dataset' and 'src_ext_address' must be set" - ) - sender.set_dataset_bytes('active', frame.active_dataset) - sender.set_extaddr(frame.src_ext_address) - - sender.diag_start() - receiver.diag_start() - - channel = 11 - num_sent_frames = 1 - - sender.diag_set_channel(channel) - receiver.diag_set_channel(channel) - receiver.diag_radio_receive() - receiver.diag_set_radio_receive_filter_dest_mac_address(frame.dst_address) - receiver.diag_enable_radio_receive_filter() - - result_queue = queue.Queue() - receive_task = threading.Thread(target=self.__radio_receive_task, - args=(receiver, num_sent_frames, result_queue), - daemon=True) - receive_task.start() - - sender.wait(0.1) - - sender.diag_frame(frame.tx_frame, - is_security_processed=frame.is_security_processed, - max_csma_backoffs=4, - max_frame_retries=4, - csma_ca_enabled=True) - sender.diag_send(num_sent_frames, is_async=False) - - receive_task.join() - - if result_queue.empty(): - ret = False # No frame is received. - else: - received_frames = result_queue.get() - if len(received_frames) != num_sent_frames: - ret = False - else: - # The radio driver may not append the FCF field to the received frame. Do not check the FCF field here. - FCF_LENGTH = 4 - expect_frame = frame.expect_rx_frame or frame.tx_frame - ret = expect_frame[:-FCF_LENGTH] == received_frames[0]['psdu'][:-FCF_LENGTH] - - if ret: - sender.diag_stop() - receiver.diag_stop() - else: - # The command 'diag radio receive ' may fail to receive specified number of frames in default - # timeout time. In this case, the diag module still in 'sync' mode, and it only allows users to run the - # command `factoryreset` to terminate the test. - sender.factory_reset() - receiver.factory_reset() - - return ret - - def __radio_receive_task(self, receiver: OTCI, number: int, result_queue: queue): - try: - receiver.set_execute_command_retry(0) - result = receiver.diag_radio_receive_number(number) - except ExpectLineTimeoutError: - pass - else: - result_queue.put(result) - finally: - receiver.set_execute_command_retry(OTCI.DEFAULT_EXEC_COMMAND_RETRY) - - def __test_diag_frame(self): - frame = self.Frame(name='diag frame 00010203040506070809', tx_frame='00010203040506070809', dst_address='-') - ret = self.__test_send_formatted_frames_retries(self.__dut, self.__ref, frame) - self.__output_format_bool(frame.name, ret) - - def __support_commands(self, commands: List[str]) -> bool: - ret = True - - for command in commands: - if self.__dut.is_command_supported(command) is False: - ret = False - break - - return ret - - def __output_results(self, commands: List[str], support: bool): - for command in commands: - self.__output_format_bool(command, support) - - def __get_dut_diag_power(self) -> int: - return int(os.getenv('DUT_DIAG_POWER', '10')) - - def __get_dut_diag_gpio(self) -> int: - return int(os.getenv('DUT_DIAG_GPIO', '0')) - - def __get_dut_diag_raw_power_setting(self) -> str: - return os.getenv('DUT_DIAG_RAW_POWER_SETTING', '112233') - - def __get_adb_key(self) -> Optional[str]: - return os.getenv('ADB_KEY', None) - - def __connect_dut(self) -> OTCI: - if os.getenv('DUT_ADB_TCP'): - node = otci.connect_otbr_adb_tcp(os.getenv('DUT_ADB_TCP'), adb_key=self.__get_adb_key()) - elif os.getenv('DUT_ADB_USB'): - node = otci.connect_otbr_adb_usb(os.getenv('DUT_ADB_USB'), adb_key=self.__get_adb_key()) - elif os.getenv('DUT_CLI_SERIAL'): - node = otci.connect_cli_serial(os.getenv('DUT_CLI_SERIAL')) - elif os.getenv('DUT_SSH'): - node = otci.connect_otbr_ssh(os.getenv('DUT_SSH')) - else: - self.__fail("Please set DUT_ADB_TCP, DUT_ADB_USB, DUT_CLI_SERIAL or DUT_SSH to connect to the DUT device.") - - return node - - def __connect_reference_device(self) -> OTCI: - if os.getenv('REF_CLI_SERIAL'): - node = otci.connect_cli_serial(os.getenv('REF_CLI_SERIAL')) - elif os.getenv('REF_SSH'): - node = otci.connect_otbr_ssh(os.getenv('REF_SSH')) - elif os.getenv('REF_ADB_USB'): - node = otci.connect_otbr_adb_usb(os.getenv('REF_ADB_USB'), adb_key=self.__get_adb_key()) - else: - self.__fail("Please set REF_CLI_SERIAL, REF_SSH or REF_ADB_USB to connect to the reference device.") - - return node - - def __output_format_string(self, name: str, value: str, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH): - prefix = (name + ' ').ljust(align_length, '-') - print(f'{prefix} {value}') - - def __output_format_bool(self, name: str, value: bool, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH): - self.__output_format_string(name, 'OK' if value else 'NotSupported', align_length) - - def __fail(self, value: str): - print(f'{value}') - sys.exit(1) - - -def parse_arguments(): - """Parse all arguments.""" - description_msg = 'This script is used for testing RCP capabilities.' - epilog_msg = textwrap.dedent( - 'Device Interfaces:\r\n' - ' DUT_ADB_TCP= Connect to the DUT via adb tcp\r\n' - ' DUT_ADB_USB= Connect to the DUT via adb usb\r\n' - ' DUT_CLI_SERIAL= Connect to the DUT via cli serial port\r\n' - ' DUT_SSH= Connect to the DUT via ssh\r\n' - ' REF_ADB_USB= Connect to the reference device via adb usb\r\n' - ' REF_CLI_SERIAL= Connect to the reference device via cli serial port\r\n' - ' REF_SSH= Connect to the reference device via ssh\r\n' - ' ADB_KEY= Full path to the adb key\r\n' - '\r\n' - 'Example:\r\n' - f' DUT_ADB_USB=1169UC2F2T0M95OR REF_CLI_SERIAL=/dev/ttyACM0 python3 {sys.argv[0]} -d\r\n') - - parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter, - description=description_msg, - epilog=epilog_msg) - - parser.add_argument( - '-c', - '--csl', - action='store_true', - default=False, - help='test whether the RCP supports CSL transmitter', - ) - - parser.add_argument( - '-l', - '--link-metrics', - action='store_true', - default=False, - help='test whether the RCP supports link metrics', - ) - - parser.add_argument( - '-d', - '--diag-commands', - action='store_true', - default=False, - help='test whether the RCP supports all diag commands', - ) - - parser.add_argument( - '-f', - '--frame-format', - action='store_true', - default=False, - help='test whether the RCP supports 802.15.4 frames of all formats', - ) - - parser.add_argument( - '-p', - '--data-poll', - action='store_true', - default=False, - help='test whether the RCP supports data poll', - ) - - parser.add_argument( - '-t', - '--throughput', - action='store_true', - default=False, - help='test Thread network 1-hop throughput', - ) - - parser.add_argument( - '-T', - '--tx-info', - action='store_true', - default=False, - help='test mTxInfo field of the radio frame', - ) - - parser.add_argument( - '-v', - '--version', - action='store_true', - default=False, - help='output version', - ) - - parser.add_argument( - '-D', - '--debug', - action='store_true', - default=False, - help='output debug information', - ) - - return parser.parse_args() - - -def main(): - arguments = parse_arguments() - - if arguments.version: - print(f'Version: {CP_CAPABILITY_VERSION}') - exit() - - if arguments.debug: - logger = logging.getLogger() - logger.setLevel(logging.DEBUG) - - rcp_caps = RcpCaps() - - if arguments.diag_commands: - rcp_caps.test_diag_commands() - - if arguments.csl: - rcp_caps.test_csl() - - if arguments.data_poll: - rcp_caps.test_data_poll() - - if arguments.link_metrics: - rcp_caps.test_link_metrics() - - if arguments.throughput: - rcp_caps.test_throughput() - - if arguments.tx_info: - rcp_caps.test_radio_frame_tx_info() - - if arguments.frame_format: - rcp_caps.test_frame_format() - - -if __name__ == '__main__': - main() diff --git a/tools/cp-caps/test_csl.py b/tools/cp-caps/test_csl.py new file mode 100644 index 000000000..d47cec642 --- /dev/null +++ b/tools/cp-caps/test_csl.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import unittest + +from otci import OTCI +from device_manager import DeviceManager + + +class TestCsl(unittest.TestCase): + """Test whether the DUT supports the CSL feature.""" + + @classmethod + def setUpClass(cls): + cls.__device_manager = DeviceManager() + cls.__dut = cls.__device_manager.dut + cls.__ref = cls.__device_manager.ref + + def setUp(self): + self.__dut.factory_reset() + self.__ref.factory_reset() + + def test_csl_transmitter(self): + """Test whether the DUT supports the CSL transmitter.""" + packets = 10 + dataset = self.__device_manager.get_default_dataset() + + self.__dut.join(dataset) + self.__dut.wait_for('state', 'leader') + + # Set the reference device as an SSED + self.__ref.set_mode('-') + self.__ref.config_csl(channel=15, period=320000, timeout=100) + self.__ref.join(dataset) + self.__ref.wait_for('state', 'child') + + child_table = self.__dut.get_child_table() + ret = len(child_table) == 1 and child_table[1]['csl'] + + if ret: + ref_mleid = self.__ref.get_ipaddr_mleid() + result = self.__dut.ping(ref_mleid, count=packets, interval=1) + ret = result['transmitted_packets'] == result['received_packets'] == packets + + self.__dut.leave() + self.__ref.leave() + + self.__device_manager.output_test_result_bool('CSL Transmitter', ret) + + def tearDown(self): + pass + + @classmethod + def tearDownClass(cls): + cls.__dut.close() + cls.__ref.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/cp-caps/test_data_poll.py b/tools/cp-caps/test_data_poll.py new file mode 100644 index 000000000..f89f9d8b1 --- /dev/null +++ b/tools/cp-caps/test_data_poll.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import unittest + +from otci import OTCI +from device_manager import DeviceManager + + +class TestDataPoll(unittest.TestCase): + """Test whether the DUT supports the data poll feature.""" + + @classmethod + def setUpClass(cls): + cls.__device_manager = DeviceManager() + cls.__dut = cls.__device_manager.dut + cls.__ref = cls.__device_manager.ref + + def setUp(self): + self.__dut.factory_reset() + self.__ref.factory_reset() + + def test_data_poll_child(self): + """Test whether the DUT supports the data poll child.""" + packets = 10 + + self.__dataset = self.__device_manager.get_default_dataset() + self.__ref.join(self.__dataset) + self.__ref.wait_for('state', 'leader') + + # Set the DUT as an SED + self.__dut.set_mode('-') + self.__dut.set_poll_period(500) + self.__dut.join(self.__dataset) + self.__dut.wait_for('state', 'child') + + dut_mleid = self.__dut.get_ipaddr_mleid() + result = self.__ref.ping(dut_mleid, count=packets, interval=1) + + self.__dut.leave() + self.__ref.leave() + + ret = result['transmitted_packets'] == result['received_packets'] == packets + self.__device_manager.output_test_result_bool('Data Poll Child', ret) + + def test_data_poll_parent(self): + """Test whether the DUT supports the data poll parent.""" + packets = 10 + + self.__dataset = self.__device_manager.get_default_dataset() + self.__dut.join(self.__dataset) + self.__dut.wait_for('state', 'leader') + + # Set the reference device as an SED + self.__ref.set_mode('-') + self.__ref.set_poll_period(500) + self.__ref.join(self.__dataset) + self.__ref.wait_for('state', 'child') + + dut_mleid = self.__dut.get_ipaddr_mleid() + result = self.__ref.ping(dut_mleid, count=packets, interval=1) + + self.__dut.leave() + self.__ref.leave() + + ret = result['transmitted_packets'] == result['received_packets'] == packets + self.__device_manager.output_test_result_bool('Data Poll Parent', ret) + + def tearDown(self): + pass + + @classmethod + def tearDownClass(cls): + cls.__dut.close() + cls.__ref.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/cp-caps/test_diag_commands.py b/tools/cp-caps/test_diag_commands.py new file mode 100644 index 000000000..f43fb071c --- /dev/null +++ b/tools/cp-caps/test_diag_commands.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import unittest +import os + +from typing import List +from otci import OTCI +from device_manager import DeviceManager, Frame + + +class TestDiagCommands(unittest.TestCase): + """Test whether the DUT supports all diag commands.""" + + @classmethod + def setUpClass(cls): + cls.__device_manager = DeviceManager() + cls.__dut = cls.__device_manager.dut + cls.__ref = cls.__device_manager.ref + cls.__dut.factory_reset() + cls.__ref.factory_reset() + + def setUp(self): + self.__dut.diag_start() + self.__ref.diag_start() + + def test_diag_channel(self): + """Test whether the DUT supports the `diag channel` commands.""" + channel = 20 + commands = ['diag channel', f'diag channel {channel}'] + + if self.__support_commands(commands): + self.__dut.diag_set_channel(channel) + value = self.__dut.diag_get_channel() + ret = value == channel + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_cw(self): + """Test whether the DUT supports the `diag cw` commands.""" + commands = ['diag cw start', 'diag cw stop'] + + for command in commands: + ret = self.__dut.is_command_supported(command) + self.__device_manager.output_test_result_bool(command, ret) + + def test_diag_echo(self): + """Test whether the DUT supports the `diag echo` commands.""" + echo_msg = '0123456789' + cmd_diag_echo = f'diag echo {echo_msg}' + cmd_diag_echo_num = f'diag echo -n 10' + + if self.__dut.is_command_supported(cmd_diag_echo): + reply = self.__dut.diag_echo(echo_msg) + ret = reply == echo_msg + else: + ret = False + self.__device_manager.output_test_result_bool(cmd_diag_echo, ret) + + if self.__dut.is_command_supported(cmd_diag_echo_num): + reply = self.__dut.diag_echo_number(10) + ret = reply == echo_msg + else: + ret = False + self.__device_manager.output_test_result_bool(cmd_diag_echo_num, ret) + + def test_diag_frame(self): + """Test whether the DUT supports the `diag frame` commands.""" + frame = Frame(name='diag frame 00010203040506070809', tx_frame='00010203040506070809', dst_address='-') + ret = self.__device_manager.send_formatted_frames_retries(self.__dut, self.__ref, frame) + self.__device_manager.output_test_result_bool(frame.name, ret) + # The method send_formatted_frames_retries() calls the `diag stop`. + # Here calls the `diag start` to make the 'tearDown()' happy. + self.__dut.diag_start() + self.__ref.diag_start() + + def test_diag_gpio_mode(self): + """Test whether the DUT supports the `diag gpio mode` commands.""" + gpio = self.__get_dut_diag_gpio() + commands = [f'diag gpio mode {gpio}', f'diag gpio mode {gpio} in', f'diag gpio mode {gpio} out'] + + if self.__support_commands(commands): + self.__dut.diag_set_gpio_mode(gpio, 'in') + mode_in = self.__dut.diag_get_gpio_mode(gpio) + self.__dut.diag_set_gpio_mode(gpio, 'out') + mode_out = self.__dut.diag_get_gpio_mode(gpio) + + ret = mode_in == 'in' and mode_out == 'out' + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_gpio_value(self): + """Test whether the DUT supports the `diag gpio get` and 'diag gpio set' commands.""" + gpio = self.__get_dut_diag_gpio() + commands = [f'diag gpio get {gpio}', f'diag gpio set {gpio} 0', f'diag gpio set {gpio} 1'] + + if self.__support_commands(commands): + self.__dut.diag_set_gpio_value(gpio, 0) + value_0 = self.__dut.diag_get_gpio_value(gpio) + self.__dut.diag_set_gpio_value(gpio, 1) + value_1 = self.__dut.diag_get_gpio_value(gpio) + + ret = value_0 == 0 and value_1 == 1 + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_power(self): + """Test whether the DUT supports the `diag power` commands.""" + power = self.__get_dut_diag_power() + commands = ['diag power', f'diag power {power}'] + + if self.__support_commands(commands): + self.__dut.diag_set_power(power) + value = self.__dut.diag_get_power() + ret = value == power + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_powersettings(self): + """Test whether the DUT supports the `diag powersettings` commands.""" + commands = ['diag powersettings', 'diag powersettings 20'] + + if self.__support_commands(commands): + powersettings = self.__dut.diag_get_powersettings() + ret = len(powersettings) > 0 + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_radio(self): + """Test whether the DUT supports the `diag radio` commands.""" + commands = ['diag radio receive', 'diag radio sleep', 'diag radio state'] + + if self.__support_commands(commands): + self.__dut.diag_radio_receive() + receive_state = self.__dut.diag_get_radio_state() + self.__dut.wait(0.1) + self.__dut.diag_radio_sleep() + sleep_state = self.__dut.diag_get_radio_state() + + ret = sleep_state == 'sleep' and receive_state == 'receive' + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_rawpowersetting(self): + """Test whether the DUT supports the `diag rawpowersetting` commands.""" + rawpowersetting = self.__get_dut_diag_raw_power_setting() + commands = [ + 'diag rawpowersetting enable', f'diag rawpowersetting {rawpowersetting}', 'diag rawpowersetting', + 'diag rawpowersetting disable' + ] + + if self.__support_commands(commands): + self.__dut.diag_enable_rawpowersetting() + self.__dut.diag_set_rawpowersetting(rawpowersetting) + reply = self.__dut.diag_get_rawpowersetting() + self.__dut.diag_disable_rawpowersetting() + + ret = reply == rawpowersetting + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_repeat(self): + """Test whether the DUT supports the `diag repeat` commands.""" + delay = 10 + threshold = 80 + length = 64 + channel = 20 + cmd_diag_repeat = f'diag repeat {delay} {length}' + cmd_diag_repeat_stop = 'diag repeat stop' + commands = [cmd_diag_repeat, 'diag repeat stop', 'diag stats', 'diag stats clear'] + + if self.__support_commands(commands): + self.__dut.diag_set_channel(channel) + self.__ref.diag_set_channel(channel) + self.__ref.diag_radio_receive() + + self.__dut.diag_stats_clear() + self.__ref.diag_stats_clear() + + self.__dut.diag_repeat(delay, length) + self.__dut.wait(1) + self.__dut.diag_repeat_stop() + dut_stats = self.__dut.diag_get_stats() + ref_stats = self.__ref.diag_get_stats() + + ret = dut_stats['sent_success_packets'] > threshold and ref_stats['received_packets'] > threshold + else: + ret = False + + self.__device_manager.output_test_result_bool(cmd_diag_repeat, ret) + self.__device_manager.output_test_result_bool(cmd_diag_repeat_stop, ret) + + def test_diag_send(self): + """Test whether the DUT supports the `diag send` commands.""" + packets = 100 + threshold = 80 + length = 64 + channel = 20 + commands = [f'diag send {packets} {length}'] + + if self.__support_commands(commands): + self.__dut.wait(1) + self.__dut.diag_set_channel(channel) + self.__ref.diag_set_channel(channel) + self.__ref.diag_radio_receive() + + self.__dut.diag_stats_clear() + self.__ref.diag_stats_clear() + + self.__dut.diag_send(packets, length) + self.__dut.wait(1) + dut_stats = self.__dut.diag_get_stats() + ref_stats = self.__ref.diag_get_stats() + + ret = dut_stats['sent_success_packets'] == packets and ref_stats['received_packets'] > threshold + else: + ret = False + + self.__output_results(commands, ret) + + def test_diag_stats(self): + """Test whether the DUT supports the `diag stats` commands.""" + commands = ['diag stats', 'diag stats clear'] + + for command in commands: + ret = self.__dut.is_command_supported(command) + self.__device_manager.output_test_result_bool(command, ret) + + def test_diag_stream(self): + """Test whether the DUT supports the 'diag stream' commands.""" + commands = ['diag stream start', 'diag stream stop'] + + for command in commands: + ret = self.__dut.is_command_supported(command) + self.__device_manager.output_test_result_bool(command, ret) + + def __get_dut_diag_power(self) -> int: + return int(os.getenv('DUT_DIAG_POWER', '10')) + + def __get_dut_diag_gpio(self) -> int: + return int(os.getenv('DUT_DIAG_GPIO', '0')) + + def __get_dut_diag_raw_power_setting(self) -> str: + return os.getenv('DUT_DIAG_RAW_POWER_SETTING', '112233') + + def __support_commands(self, commands: List[str]) -> bool: + ret = True + + for command in commands: + if self.__dut.is_command_supported(command) is False: + ret = False + break + + return ret + + def __output_results(self, commands: List[str], support: bool): + for command in commands: + self.__device_manager.output_test_result_bool(command, support) + + def tearDown(self): + self.__ref.diag_stop() + self.__dut.diag_stop() + + @classmethod + def tearDownClass(cls): + cls.__dut.close() + cls.__ref.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/cp-caps/test_frame_formats.py b/tools/cp-caps/test_frame_formats.py new file mode 100644 index 000000000..9c20f9b69 --- /dev/null +++ b/tools/cp-caps/test_frame_formats.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import unittest + +from otci import OTCI +from device_manager import DeviceManager, Frame + + +class TestFrameFormats(unittest.TestCase): + """Test whether the DUT supports all different 15.4 frame formats.""" + + @classmethod + def setUpClass(cls): + cls.__device_manager = DeviceManager() + cls.__dut = cls.__device_manager.dut + cls.__ref = cls.__device_manager.ref + + cls.__dut.factory_reset() + cls.__ref.factory_reset() + + def setUp(self): + pass + + def test_frame_format_01(self): + frame = Frame(name='ver:2003,Cmd,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', + tx_frame='030800ffffffff070000', + dst_address='0xffff') + self.__test_frame_format(1, frame) + + def test_frame_format_02(self): + frame = Frame( + name='ver:2003,Bcon,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:30', + tx_frame='00c000eeee0102030405060708ff0f000003514f70656e54687265616400000000000001020304050607080000', + dst_address='-') + self.__test_frame_format(2, frame) + + def test_frame_format_03(self): + frame = Frame( + name='ver:2003,MP,noseq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie[ren con],plen:0', + tx_frame='fd87dddd1020304050607080010203040506070815000000007265616401820ee80305009bb8ea011c807aa1120000', + dst_address='8070605040302010') + self.__test_frame_format(3, frame) + + def test_frame_format_04(self): + frame = Frame(name='ver:2006,Cmd,seq,dst[addr:short,pan:id],src[addr:short,pan:no],sec:l5,ie:no,plen:0', + tx_frame='4b9800ddddaaaabbbb0d0000000001043daa1aea0000', + dst_address='0xaaaa') + self.__test_frame_format(4, frame) + + def test_frame_format_05(self): + frame = Frame(name='ver:2006,Cmd,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:l5,ie:no,plen:0', + tx_frame='4bdc00dddd102030405060708001020304050607080d000000000104483cb8a90000', + dst_address='8070605040302010') + self.__test_frame_format(5, frame) + + def test_frame_format_06(self): + frame = Frame(name='ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0', + tx_frame='41dc00dddd102030405060708001020304050607080000', + dst_address='8070605040302010') + self.__test_frame_format(6, frame) + + def test_frame_format_07(self): + frame = Frame(name='ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', + tx_frame='019800ddddaaaaeeeebbbb0000', + dst_address='0xaaaa') + self.__test_frame_format(7, frame) + + def test_frame_format_08(self): + frame = Frame(name='ver:2006,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', + tx_frame='011c00dddd10203040506070800000', + dst_address='8070605040302010') + self.__test_frame_format(8, frame) + + def test_frame_format_09(self): + frame = Frame(name='ver:2006,Data,seq,dst[addr:short,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', + tx_frame='011800ddddaaaa0000', + dst_address='0xaaaa') + self.__test_frame_format(9, frame) + + def test_frame_format_10(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0', + tx_frame='0120000000', + dst_address='-') + self.__test_frame_format(10, frame) + + def test_frame_format_11(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:no,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', + tx_frame='412000dddd0000', + dst_address='-') + self.__test_frame_format(11, frame) + + def test_frame_format_12(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:no,pan:no],sec:no,ie:no,plen:0', + tx_frame='012c00dddd10203040506070800000', + dst_address='8070605040302010') + self.__test_frame_format(12, frame) + + def test_frame_format_13(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:no,pan:no],sec:no,ie:no,plen:0', + tx_frame='412c0010203040506070800000', + dst_address='8070605040302010') + self.__test_frame_format(13, frame) + + def test_frame_format_14(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:id],sec:no,ie:no,plen:0', + tx_frame='01e000eeee01020304050607080000', + dst_address='-') + self.__test_frame_format(14, frame) + + def test_frame_format_15(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:no,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0', + tx_frame='41e00001020304050607080000', + dst_address='-') + self.__test_frame_format(15, frame) + + def test_frame_format_16(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:extd,pan:no],sec:no,ie:no,plen:0', + tx_frame='01ec00dddd102030405060708001020304050607080000', + dst_address='8070605040302010') + self.__test_frame_format(16, frame) + + def test_frame_format_17(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:no],src[addr:extd,pan:no],sec:no,ie:no,plen:0', + tx_frame='41ec00102030405060708001020304050607080000', + dst_address='8070605040302010') + self.__test_frame_format(17, frame) + + def test_frame_format_18(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', + tx_frame='01a800ddddaaaaeeeebbbb0000', + dst_address='0xaaaa') + self.__test_frame_format(18, frame) + + def test_frame_format_19(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:extd,pan:id],sec:no,ie:no,plen:0', + tx_frame='01e800ddddaaaaeeee01020304050607080000', + dst_address='0xaaaa') + self.__test_frame_format(19, frame) + + def test_frame_format_20(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:extd,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', + tx_frame='01ac00dddd1020304050607080eeeebbbb0000', + dst_address='8070605040302010') + self.__test_frame_format(20, frame) + + def test_frame_format_21(self): + frame = Frame(name='ver:2015,Data,seq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie[csl],plen:0', + tx_frame='01aa00ddddaaaaeeeebbbb040dc800e8030000', + dst_address='0xaaaa') + self.__test_frame_format(21, frame) + + def test_frame_format_22(self): + frame = Frame(name='ver:2015,Data,noseq,dst[addr:short,pan:id],src[addr:short,pan:id],sec:no,ie:no,plen:0', + tx_frame='01a9ddddaaaaeeeebbbb0000', + dst_address='0xaaaa') + self.__test_frame_format(22, frame) + + def __test_frame_format(self, number: int, frame: Frame): + ret = self.__device_manager.send_formatted_frames_retries(self.__dut, self.__ref, frame) + self.__device_manager.output_test_result_bool(f'{number} TX {frame.name}', ret, align_length=100) + ret = self.__device_manager.send_formatted_frames_retries(self.__ref, self.__dut, frame) + self.__device_manager.output_test_result_bool(f'{number} RX {frame.name}', ret, align_length=100) + + def tearDown(self): + pass + + @classmethod + def tearDownClass(cls): + cls.__dut.close() + cls.__ref.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/cp-caps/test_link_metrics.py b/tools/cp-caps/test_link_metrics.py new file mode 100644 index 000000000..8b9e1a351 --- /dev/null +++ b/tools/cp-caps/test_link_metrics.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import unittest + +from otci import OTCI +from otci.types import Ip6Addr +from device_manager import DeviceManager + + +class TestLinkMetrics(unittest.TestCase): + """Test whether the DUT supports Link Metrics feature.""" + + @classmethod + def setUpClass(cls): + cls.__device_manager = DeviceManager() + cls.__dut = cls.__device_manager.dut + cls.__ref = cls.__device_manager.ref + + def setUp(self): + self.__dut.factory_reset() + self.__ref.factory_reset() + dataset = self.__device_manager.get_default_dataset() + + self.__dut.join(dataset) + self.__dut.wait_for('state', 'leader') + + self.__ref.join(dataset) + self.__ref.wait_for('state', ['child', 'router']) + + def test_link_metrics_initiator(self): + """Test whether the DUT supports Link Metrics Initiator.""" + ref_linklocal_address = self.__ref.get_ipaddr_linklocal() + ret = self.__run_link_metrics_test_commands(initiator=self.__dut, subject_address=ref_linklocal_address) + self.__device_manager.output_test_result_bool('Link Metrics Initiator', ret) + + def test_link_metrics_subject(self): + """Test whether the DUT supports Link Metrics Subject.""" + dut_linklocal_address = self.__dut.get_ipaddr_linklocal() + ret = self.__run_link_metrics_test_commands(initiator=self.__ref, subject_address=dut_linklocal_address) + self.__device_manager.output_test_result_bool('Link Metrics Subject', ret) + + def __run_link_metrics_test_commands(self, initiator: OTCI, subject_address: Ip6Addr) -> bool: + seriesid = 1 + series_flags = 'ldra' + link_metrics_flags = 'qr' + probe_length = 10 + + if not initiator.linkmetrics_config_enhanced_ack_register(subject_address, link_metrics_flags): + return False + + if not initiator.linkmetrics_config_forward(subject_address, seriesid, series_flags, link_metrics_flags): + return False + + initiator.linkmetrics_probe(subject_address, seriesid, probe_length) + + results = initiator.linkmetrics_request_single(subject_address, link_metrics_flags) + if not ('lqi' in results.keys() and 'rssi' in results.keys()): + return False + + results = initiator.linkmetrics_request_forward(subject_address, seriesid) + if not ('lqi' in results.keys() and 'rssi' in results.keys()): + return False + + if not initiator.linkmetrics_config_enhanced_ack_clear(subject_address): + return False + + return True + + def tearDown(self): + self.__ref.leave() + self.__dut.leave() + + @classmethod + def tearDownClass(cls): + cls.__dut.close() + cls.__ref.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/cp-caps/test_radio_frame_tx_info.py b/tools/cp-caps/test_radio_frame_tx_info.py new file mode 100644 index 000000000..74d12676b --- /dev/null +++ b/tools/cp-caps/test_radio_frame_tx_info.py @@ -0,0 +1,248 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import unittest +import time + +from otci import OTCI +from device_manager import DeviceManager, Frame + + +class TestRadioFrameTxInfo(unittest.TestCase): + """Test whether the DUT supports otRadioFrame.mInfo.mTxInfo field.""" + + @classmethod + def setUpClass(cls): + cls.__device_manager = DeviceManager() + cls.__dut = cls.__device_manager.dut + cls.__ref = cls.__device_manager.ref + + def setUp(self): + self.__dut.factory_reset() + self.__ref.factory_reset() + + def test_radio_frame_tx_info_csma_ca_enabled(self): + """Test whether the DUT supports mInfo.mTxInfo.mCsmaCaEnabled field.""" + self.__dut.diag_start() + self.__ref.diag_start() + + channel = 11 + num_sent_frames = 1 + tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' + + self.__dut.diag_set_channel(channel) + self.__ref.diag_set_channel(channel) + self.__ref.diag_cw_start() + + self.__dut.diag_stats_clear() + self.__dut.diag_frame(tx_frame, csma_ca_enabled=False) + self.__dut.diag_send(num_sent_frames, is_async=False) + dut_stats = self.__dut.diag_get_stats() + ret = dut_stats['sent_success_packets'] == num_sent_frames + self.__device_manager.output_test_result_bool('mCsmaCaEnabled=0', ret) + + self.__dut.diag_stats_clear() + self.__dut.diag_frame(tx_frame, csma_ca_enabled=True) + self.__dut.diag_send(num_sent_frames, is_async=False) + dut_stats = self.__dut.diag_get_stats() + ret = dut_stats['sent_error_cca_packets'] == num_sent_frames + self.__device_manager.output_test_result_bool('mCsmaCaEnabled=1', ret) + + self.__ref.diag_cw_stop() + + def test_radio_frame_tx_info_is_security_processed(self): + """Test whether the DUT supports mInfo.mTxInfo.mIsSecurityProcessed field.""" + active_dataset = self.__device_manager.get_default_dataset() + + frames = [ + Frame(name='mIsSecurityProcessed=True', + tx_frame='09ec00dddd102030405060708001020304050607080d0000000001db622c220fde64408db9128c93d50000', + dst_address='8070605040302010', + is_security_processed=True), + Frame(name='mIsSecurityProcessed=False', + tx_frame='09ec00dddd102030405060708001020304050607080d000000000000010203040506070809000000000000', + expect_rx_frame= + '09ec00dddd102030405060708001020304050607080d0000000001db622c220fde64408db9128c93d50000', + is_security_processed=False, + dst_address='8070605040302010', + active_dataset=active_dataset, + src_ext_address='0807060504030201'), + ] + + for frame in frames: + ret = self.__device_manager.send_formatted_frames_retries(self.__dut, self.__ref, frame) + self.__device_manager.output_test_result_bool(frame.name, ret) + + # The method `send_formatted_frames_retries()` calls the `diag stop`. + # Here calls the `diag start` to make the `tearDown()` happy. + self.__ref.diag_start() + self.__dut.diag_start() + + def test_radio_frame_tx_info_max_csma_backoffs(self): + """Test whether the DUT supports mInfo.mTxInfo.mMaxCsmaBackoffs field.""" + self.__dut.diag_start() + self.__ref.diag_start() + + channel = 11 + num_sent_frames = 1 + tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' + + self.__dut.diag_set_channel(channel) + self.__ref.diag_set_channel(channel) + + self.__ref.diag_cw_start() + self.__dut.wait(0.05) + + # When `max_csma_backoffs` is set to 0, the radio driver should skip backoff and do CCA once. + # The CCA time is 192 us. Theoretically, the `diag send` command should return after 192us. + # But considering the Spinel delay and the system IO delay, here sets `max_time_cost` to 20 ms. + max_time_cost = 20 + max_csma_backoffs = 0 + self.__dut.diag_stats_clear() + self.__dut.diag_frame(tx_frame, csma_ca_enabled=True, max_frame_retries=0, max_csma_backoffs=max_csma_backoffs) + start_time = time.time() + self.__dut.diag_send(num_sent_frames, is_async=False) + end_time = time.time() + time_cost = int((end_time - start_time) * 1000) + ret = 'OK' if time_cost < max_time_cost else 'NotSupported' + self.__device_manager.output_test_result_string(f'mMaxCsmaBackoffs={max_csma_backoffs}', + f'{ret} ({time_cost} ms)') + + # Basic information for calculating the backoff time: + # aTurnaroundTime = 192 us + # aCCATime = 128 us + # backoffExponent = (macMinBe, macMaxBe) = (3, 5) + # backoffPeriod = random() % (1 << backoffExponent) + # backoff = backoffPeriod * aUnitBackoffPeriod = backoffPeriod * (aTurnaroundTime + aCCATime) + # = backoffPeriod * 320 us + # backoff = (random() % (1 << backoffExponent)) * 320us + # + # `max_csma_backoffs` is set to 100 here, `backoffExponent` will be set to 5 in most retries. + # backoff = (random() % 32) * 320us + # average_backoff = 16 * 320us = 5120 us + # total_backoff = average_backoff * 100 = 5120 us * 100 = 512 ms + # + # Here sets the `max_time_cost` to half of `total_backoff`. + # + max_time_cost = 256 + max_frame_retries = 0 + max_csma_backoffs = 100 + self.__dut.diag_frame(tx_frame, csma_ca_enabled=True, max_frame_retries=0, max_csma_backoffs=max_csma_backoffs) + start_time = time.time() + self.__dut.diag_send(num_sent_frames, is_async=False) + end_time = time.time() + time_cost = int((end_time - start_time) * 1000) + ret = 'OK' if time_cost > max_time_cost else 'NotSupported' + self.__device_manager.output_test_result_string(f'mMaxCsmaBackoffs={max_csma_backoffs}', + f'{ret} ({time_cost} ms)') + + self.__ref.diag_cw_stop() + + def test_radio_frame_tx_info_rx_channel_after_tx_done(self): + """Test whether the DUT supports mInfo.mTxInfo.mRxChannelAfterTxDone field.""" + self.__dut.diag_start() + self.__ref.diag_start() + + channel = 11 + num_sent_frames = 1 + rx_channel_after_tx_done = 25 + dut_address = 'dead00beefcafe01' + dut_tx_frame = '01ec00dddd02fecaefbe00adde01fecaefbe00adde000102030405060708090000' + ref_tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' + + self.__dut.diag_set_channel(channel) + self.__dut.diag_set_radio_receive_filter_dest_mac_address(dut_address) + self.__dut.diag_enable_radio_receive_filter() + self.__dut.diag_stats_clear() + + self.__dut.diag_frame(dut_tx_frame, rx_channel_after_tx_done=rx_channel_after_tx_done) + self.__dut.diag_send(num_sent_frames, is_async=False) + stats = self.__dut.diag_get_stats() + ret = stats['sent_success_packets'] == num_sent_frames + + if ret: + self.__ref.diag_set_channel(rx_channel_after_tx_done) + self.__ref.diag_frame(ref_tx_frame) + self.__ref.diag_send(num_sent_frames, is_async=False) + stats = self.__dut.diag_get_stats() + ret = stats['received_packets'] == num_sent_frames + + self.__device_manager.output_test_result_bool('mRxChannelAfterTxDone', ret) + + def test_radio_frame_tx_info_tx_delay(self): + """Test whether the DUT supports mInfo.mTxInfo.mTxDelayBaseTime and mInfo.mTxInfo.mTxDelay fields.""" + # Enable the IPv6 interface to force the host and the RCP to synchronize the radio time. + self.__dut.set_dataset_bytes('active', self.__device_manager.get_default_dataset()) + self.__dut.ifconfig_up() + self.__dut.wait(0.5) + self.__dut.ifconfig_down() + + self.__dut.diag_start() + self.__ref.diag_start() + + channel = 11 + packets = 1 + dut_tx_delay_sec = 0.5 + dut_tx_delay_us = int(dut_tx_delay_sec * 1000000) + ref_rx_delay_sec = dut_tx_delay_sec / 2 + ref_address = 'dead00beefcafe01' + dut_tx_frame = '01ec00dddd01fecaefbe00adde02fecaefbe00adde000102030405060708090000' + + self.__dut.diag_set_channel(channel) + self.__ref.diag_set_channel(channel) + self.__ref.diag_radio_receive() + self.__ref.diag_set_radio_receive_filter_dest_mac_address(ref_address) + self.__ref.diag_enable_radio_receive_filter() + + self.__dut.diag_frame(dut_tx_frame, tx_delay=dut_tx_delay_us) + self.__dut.diag_send(packets, is_async=True) + + self.__ref.wait(ref_rx_delay_sec) + stats = self.__ref.diag_get_stats() + ret = stats['received_packets'] == 0 + + if ret is True: + self.__ref.wait(ref_rx_delay_sec) + stats = self.__ref.diag_get_stats() + ret = stats['received_packets'] == 1 + + self.__device_manager.output_test_result_bool(f'mTxDelayBaseTime=now,mTxDelay={dut_tx_delay_us}', ret) + + def tearDown(self): + self.__ref.diag_stop() + self.__dut.diag_stop() + + @classmethod + def tearDownClass(cls): + cls.__dut.close() + cls.__ref.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/cp-caps/test_throughput.py b/tools/cp-caps/test_throughput.py new file mode 100644 index 000000000..1951fb7c5 --- /dev/null +++ b/tools/cp-caps/test_throughput.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2025, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import threading +import unittest + +from otci import OTCI +from device_manager import DeviceManager + + +class TestThroughput(unittest.TestCase): + """Test Thread network 1 hop throughput.""" + + @classmethod + def setUpClass(cls): + cls.__device_manager = DeviceManager() + cls.__dut = cls.__device_manager.dut + cls.__ref = cls.__device_manager.ref + + def setUp(self): + self.__dut.factory_reset() + self.__ref.factory_reset() + + def test_throughput(self): + """Test Thread network 1 hop throughput.""" + if not self.__dut.support_iperf3(): + print("The DUT doesn't support the tool iperf3") + return + + if not self.__ref.support_iperf3(): + print("The reference device doesn't support the tool iperf3") + return + + bitrate = 90000 + length = 1232 + transmit_time = 30 + max_wait_time = 30 + timeout = transmit_time + max_wait_time + + dataset = self.__device_manager.get_default_dataset() + + self.__dut.join(dataset) + self.__dut.wait_for('state', 'leader') + + self.__ref.set_router_selection_jitter(1) + self.__ref.join(dataset) + self.__ref.wait_for('state', ['child', 'router']) + + ref_mleid = self.__ref.get_ipaddr_mleid() + + ref_iperf3_server = threading.Thread(target=self.__ref_iperf3_server_task, + args=(ref_mleid, timeout), + daemon=True) + ref_iperf3_server.start() + self.__dut.wait(1) + + results = self.__dut.iperf3_client(host=ref_mleid, bitrate=bitrate, transmit_time=transmit_time, length=length) + ref_iperf3_server.join() + + if not results: + print('Failed to run the iperf3') + return + + self.__device_manager.output_test_result_string('Throughput', + self.__bitrate_to_string(results['receiver']['bitrate'])) + + def __ref_iperf3_server_task(self, bind_address: str, timeout: int): + self.__ref.iperf3_server(bind_address, timeout=timeout) + + def __bitrate_to_string(self, bitrate: float): + units = ['bits/sec', 'Kbits/sec', 'Mbits/sec', 'Gbits/sec', 'Tbits/sec'] + unit_index = 0 + + while bitrate >= 1000 and unit_index < len(units) - 1: + bitrate /= 1000 + unit_index += 1 + + return f'{bitrate:.2f} {units[unit_index]}' + + def tearDown(self): + pass + + @classmethod + def tearDownClass(cls): + cls.__dut.close() + cls.__ref.close() + + +if __name__ == '__main__': + unittest.main()