Aksal/nuttx
3
1
Fork 1
mirror of https://github.com/apache/nuttx.git synced 2026-09-03 23:52:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

tools/ci: remove obsolete testrun harness

Browse source 
NuttX CI runtime test cases have migrated to NTFC.
Remove the unused pytest harness and related files.

Assisted-by: Claude Code
Signed-off-by: raiden00pl <raiden00@railab.me>
This commit is contained in:
raiden00pl 2026-09-01 14:28:25 +02:00 • committed by Alan C. Assis
parent 83b4239e1d
commit 179bf35fdd
23 changed files with 5 additions and 13137 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.github/CODEOWNERS vendored
View file

@ -190,7 +190,6 @@ Documentation/guides/automounter.rst raiden00@railab.me
Documentation/guides/building_nuttx_with_app_out_of_src_tree.rst ludovicvanasse@gmail.com
Documentation/guides/building_uclibcpp.rst ludovicvanasse@gmail.com
Documentation/guides/changing_systemclockconfig.rst ludovicvanasse@gmail.com devel@sumpfralle.de
Documentation/guides/citests.rst michallenc@seznam.cz devel@sumpfralle.de ludovicvanasse@gmail.com 59230071+hartmannathan@users.noreply.github.com
Documentation/guides/cpp_cmake.rst flavio.alves@phiinnovations.com xiaoxiang@xiaomi.com longdh.2017.vn@gmail.com raiden00@railab.me 59230071+hartmannathan@users.noreply.github.com
Documentation/guides/custom_app_directories.rst ludovicvanasse@gmail.com
Documentation/guides/customapps.rst 56726697+TimJTi@users.noreply.github.com acassis@gmail.com raiden00@railab.me 31182898+dlstjq7685@users.noreply.github.com

59
Documentation/testing/citests.rst
View file

@ -1,59 +0,0 @@
.. include:: /substitutions.rst
.. _citests:
======================================
Running CI Runtime Test Suites Locally
======================================
NuttX automatically runs continuous integration (CI) tests on
`simulator <https://nuttx.apache.org/docs/latest/guides/simulator.html>`__
and QEMU targets when a new pull request is submitted. To avoid the tests failing
you can also run them locally on your computer prior to submitting a new pull request.
This page describes the step by step manual to do so.
Configuring NuttX
=================
NuttX has a simulator target that allows the user to run NuttX as a regular
program on a computer. The simulator target with CI test is configured and
compiled followingly.
.. code-block:: console
$ cd nuttx
$ ./tools/configure.sh sim:citest
$ make
You should see NuttX shell with built in test applications. Now you can exit
the simulator.
.. code-block:: console
nsh> poweroff
$
$ # we're back at the Linux prompt.
Running CI Tests with ``tools/ci/testrun``
==========================================
Running CI tests locally requires Minicom and Python 3.6 or newer to be
installed on the system. Other requirements can be installed with following
set of commands.
.. code-block:: console
$ cd tools/ci/testrun/env
$ pip3 install -r requirements.txt
$ cd ..
$ cd script
Now you have everything prepared to run CI tests. The tests themselves are run
by following command.
.. code-block:: console
$ python3 -m pytest -m 'common or sim' ./ -B sim -P <nuttx-path> -L <log-path> -R sim -C --json=<log-path>/pytest.json
Where nuttx-path is an absolute path to NuttX root directory and log-path is
a user defined directory to which tests log are saved.

5
Documentation/testing/index.rst
View file

@ -7,9 +7,12 @@ This page contains information related to NuttX testing.
For information about included NuttX testing applications, please see
:doc:`/applications/testing/index`.
.. note::
NuttX CI runtime test cases use :doc:`NTFC </testing/ntfc>`.
.. toctree::
nuttx-ci.rst
citests.rst
ntfc.rst
nxdart.rst

2
Documentation/testing/ntfc.rst
View file

@ -14,7 +14,7 @@ The framework and official test cases are available at:
Detailed documentation is available in the framework repository.
CI migration to NTFC is in progress. Configurations currently using NTFC:
CI runtime test cases use NTFC. Configurations with an NTFC runner include:
- :doc:`risc-v/qemu-rv </platforms/risc-v/qemu-rv/index>`: ``rv-virt/citest64``
- :doc:`risc-v/qemu-rv </platforms/risc-v/qemu-rv/index>`: ``rv-virt/citest``

68
tools/ci/cirun.sh
View file

@ -1,68 +0,0 @@
#!/usr/bin/env bash
############################################################################
# tools/ci/cirun.sh
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
set -e
set -o xtrace
WD=$(cd $(dirname $0) && pwd)
WORKSPACE=$(cd $WD/../../../../../../../ && pwd -P)
nuttx=$WORKSPACE/nuttx
logs=${WD}/logs
BOARD=`echo $WD |awk -F '/' '{print $(NF-2)}'`
echo $WD
echo $WORKSPACE
config=$(basename $WD)
if [ "$BOARD" == "sim" ]; then
target="sim"
mark="common or ${BOARD}"
else
if [ "${config:$((-2))}" == "64" ]; then
BOARD="${BOARD}64"
fi
if [ "$BOARD" == "rv-virt" ]; then
target="qemu"
mark="qemu or rv_virt"
else
target="qemu"
mark=$target
fi
fi
core=$target
image=`find ${nuttx} -type f -name 'nuttx'`
path=${image%/*}
cd ${nuttx}/tools/ci/testrun/script
python3 -m pytest -m "${mark}" ./ -B ${BOARD} -P ${path} -L ${logs}/${BOARD}/${core} -R ${target} -C --json=${logs}/${BOARD}/${core}/pytest.json
ret="$?"
#clean
find ${nuttx}/tools/ci/testrun -name '__pycache__' |xargs rm -rf
find ${nuttx}/tools/ci/testrun -name '.pytest_cache' |xargs rm -rf
rm -rf ${logs}
rm -f ${nuttx}/fatfs.img
echo $ret
exit $ret

43
tools/ci/testrun/README.md
View file

@ -1,43 +0,0 @@
# Requirements
- Python 3.6 or above
- pexpect + minicom
- pytest
```bash
sudo apt-get install minicom
cd env
pip3 install -r requirements.txt
```
# Pytest Original Parameter usage refer to help
# Customized Parameters
```bash
pytest
-D specify device, for example: /dev/ttyUSB0
-B specify board, for example: sim,best1600_ep
-P specify nuttx path
-F specify filesystem, for example: /data
-L specify log path
-O specify ota version, for example: VELA-2.0
-S enable sudo as run sim
-C enable pre-checkin run
-U specify core: ap, audio, cp, sensor, tee
-M serial or minicom
-R specify target type: target|sim|qemu|module
```
# Example
```bash
cd script
# testsuite
pytest test_[testsuite].py -D /dev/ttyUSBx -B <board> -L <logpath> -F <filesystem folder> -P <nuttx path> -R <target|sim|qemu|module>
# mark
pytest -m <mark_name> ./ -D /dev/ttyUSBx -B <board> -L <logpath> -F <filesystem folder> -P <nuttx path> -R <target|sim|qemu|module> --json=<logpath>/pytest.json
# sim
pytest -m sim ./ -B sim -P <nuttx path> -L <logpath> -F <filesystem folder> -P <nuttx path> -R <target|sim|qemu|module> --json=<logpath>/pytest.json
# stress test
pytest -m miwear ./ -B miwear -P <nuttx path> -L <logpath> -F <filesystem path> --json=<logpath>/pytest.json --count=100 --repeat-scope=session
```
# Generate Report
```bash
cd utils/report
python3 report_gen.py -l <json log path> -b <branch>
```

6
tools/ci/testrun/env/requirements.txt vendored
View file

@ -1,6 +0,0 @@
pexpect==4.8.0
pytest==6.2.5
pytest-repeat==0.9.1
pytest-json==0.4.0
pytest-ordering==0.6
pyserial==3.5

18
tools/ci/testrun/pytest.ini
View file

@ -1,18 +0,0 @@
# pytest.ini
[pytest]
# -p : Disable warning capture -p no:warnings
# -s : Capture print of the script
# -v : --verbose
# --html=../report/reports.html --self-contained-html
# --alluredir=../report/allure
# --count=5 --repeat-scope=session
# --capture=sys
# --json=pytest_ota_m33.json
addopts = -sv --capture=sys
markers =
common : 'marks tests as common'
sim : 'marks tests as simulator'
qemu : 'marks tests as qemu'
rv_virt : 'marks tests as rv-virt'
disable_autouse : 'disable autouse'

23
tools/ci/testrun/script/__init__.py
View file

@ -1,23 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/script/__init__.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8

156
tools/ci/testrun/script/conftest.py
View file

@ -1,156 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/script/conftest.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8
import pytest
from utils.common import connectNuttx
def pytest_addoption(parser):
parser.addoption(
"-D",
action="store",
default=None,
help="specify device, for example: /dev/ttyUSB0",
)
parser.addoption(
"-B", action="store", default="sim", help="specify board, for example: sim"
)
parser.addoption(
"-P",
action="store",
default=None,
help="specify vela path, for example: /home/root/vela",
)
parser.addoption(
"-F",
action="store",
default="/data",
help="specify filesystem, for example: /data or /tmp",
)
parser.addoption(
"-L",
action="store",
default=None,
help="specify log path, for example: /home/root/vela/logs",
)
parser.addoption("-O", action="store", default=None, help="specify ota version")
parser.addoption(
"-S", action="store_true", default=False, help="enable sudo as run sim"
)
parser.addoption(
"-C", action="store_true", default=False, help="enable pre-checkin run"
)
parser.addoption(
"-U",
action="store",
default=None,
help="specify core: ap, audio, cp, sensor, tee",
)
parser.addoption(
"-M",
action="store",
default="minicom",
help="serial open method:serial or minicom",
)
parser.addoption(
"-R",
action="store",
default="sim",
help="specify the target type: target|qemu|sim|module, default is sim",
)
@pytest.fixture(scope="session")
def get_option(pytestconfig):
dev = pytestconfig.getoption("-D")
board = pytestconfig.getoption("-B")
vela_path = pytestconfig.getoption("-P")
fs = pytestconfig.getoption("-F")
log_path = pytestconfig.getoption("-L")
ota_version = pytestconfig.getoption("-O")
sudo = pytestconfig.getoption("-S")
ci = pytestconfig.getoption("-C")
core = pytestconfig.getoption("-U")
method = pytestconfig.getoption("-M")
target = pytestconfig.getoption("-R")
yield dev, board, vela_path, fs, log_path, ota_version, core, sudo, ci, method, target
@pytest.fixture(scope="session", name="p")
def connect_nuttx_session(get_option):
(
dev,
board,
vela_path,
fs,
log_path,
ota_version,
core,
sudo,
ci,
method,
target,
) = get_option
print(get_option)
p = connectNuttx(
board, vela_path, dev, log_path, fs, ota_version, core, sudo, ci, method, target
)
p.setup()
yield p
p.cleanup()
@pytest.fixture(scope="function", autouse=True)
def do_free_ps(request, p):
if "disable_autouse" in request.keywords:
yield
else:
yield
p.sendCommand("free", "total", flag=">")
p.sendCommand("ps", "PID", flag=">")
p.sendCommand("ls %s" % p.fs, flag=">")
@pytest.fixture(scope="function", name="pp")
def connect_nuttx_function(get_option):
(
dev,
board,
vela_path,
fs,
log_path,
ota_version,
core,
sudo,
ci,
method,
target,
) = get_option
print(get_option)
p = connectNuttx(
board, vela_path, dev, log_path, fs, ota_version, core, sudo, ci, method, target
)
p.setup()
yield p
# p.cleanup()

23
tools/ci/testrun/script/test_example/__init__.py
View file

@ -1,23 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/script/text_example/__init__.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8

51
tools/ci/testrun/script/test_example/test_example.py
View file

@ -1,51 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/script/text_example/test_example.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8
import pytest
pytestmark = [pytest.mark.common, pytest.mark.qemu]
def test_hello(p):
ret = p.sendCommand("hello", "Hello, World!!")
assert ret == 0
def test_helloxx(p):
ret = p.sendCommand("helloxx", "Hello, World!!")
assert ret == 0
def test_pipe(p):
ret = p.sendCommand("pipe", "redirect_reader: Returning success", timeout=60)
assert ret == 0
def test_popen(p):
ret = p.sendCommand("popen", "Calling pclose()")
assert ret == 0
def test_usrsocktest(p):
ret = p.sendCommand("usrsocktest", "FAILED:0", timeout=60)
assert ret == 0

23
tools/ci/testrun/script/test_framework/__init__.py
View file

@ -1,23 +0,0 @@
#!/usr/bin/python3
############################################################################
# tools/ci/testrun/script/test_framework/__init__.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8

61
tools/ci/testrun/script/test_framework/test_cmocka.py
View file

@ -1,61 +0,0 @@
#!/usr/bin/python3
############################################################################
# tools/ci/testrun/script/test_framework/test_cmocka.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8
import os
import pytest
pytestmark = [pytest.mark.common, pytest.mark.rv_virt]
cmocka_list_start = "cmocka_list_start"
cmocka_list_end = "cmocka_list_end"
cmocka_test_start = "cmocka_test_start"
cmocka_test_end = "cmocka_test_end"
@pytest.mark.run(order=1)
def test_cmocka(p):
if p.board == "sim":
os.mkdir("./test")
ret = p.sendCommand("mount -t hostfs -o fs=./test /data")
if p.board == "rv-virt":
ret = p.sendCommand("mount -t vfat /dev/virtblk0 /data")
p.sendCommand(f"echo {cmocka_list_start}")
p.sendCommand("cmocka --list", "Cmocka Test Completed")
p.sendCommand(f"echo {cmocka_list_end}")
p.sendCommand(f"echo {cmocka_test_start}")
ret = p.sendCommand(
"cmocka --skip drivertest_pm|drivertest_posix_timer|drivertest_block*|"
"test_playback.*|test_interaction.*|test_stress.*|test_capture.*",
"Cmocka Test Completed",
timeout=1200,
)
p.sendCommand(f"echo {cmocka_test_end}")
if p.board == "sim":
os.rmdir("./test")
assert ret == 0

23
tools/ci/testrun/script/test_libuv/__init__.py
View file

@ -1,23 +0,0 @@
#!/usr/bin/python3
############################################################################
# tools/ci/testrun/script/test_libuv/__init__.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8

448
tools/ci/testrun/script/test_libuv/test_libuv.py
View file

@ -1,448 +0,0 @@
#!/usr/bin/python3
############################################################################
# tools/ci/testrun/script/test_libuv/test_libuv.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8
import pytest
class TestLibuv:
pytestmark = [pytest.mark.sim]
def test_test_macros(self, p):
ret = p.sendCommand(
"uv_run_tests test_macros", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_close_order(self, p):
ret = p.sendCommand(
"uv_run_tests close_order", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_run_once(self, p):
ret = p.sendCommand(
"uv_run_tests run_once", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_run_nowait(self, p):
ret = p.sendCommand(
"uv_run_tests run_nowait", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_loop_alive(self, p):
ret = p.sendCommand(
"uv_run_tests loop_alive", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_loop_close(self, p):
ret = p.sendCommand(
"uv_run_tests loop_close", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_loop_instant_close(self, p):
ret = p.sendCommand(
"uv_run_tests loop_instant_close", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_loop_stop(self, p):
ret = p.sendCommand(
"uv_run_tests loop_stop", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_loop_backend_timeout(self, p):
ret = p.sendCommand(
"uv_run_tests loop_backend_timeout", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_default_loop_close(self, p):
ret = p.sendCommand(
"uv_run_tests default_loop_close", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_barrier_1(self, p):
ret = p.sendCommand(
"uv_run_tests barrier_1", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_barrier_2(self, p):
ret = p.sendCommand(
"uv_run_tests barrier_2", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_barrier_3(self, p):
ret = p.sendCommand(
"uv_run_tests barrier_3", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_barrier_serial_thread(self, p):
ret = p.sendCommand(
"uv_run_tests barrier_serial_thread", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_barrier_serial_thread_single(self, p):
ret = p.sendCommand(
"uv_run_tests barrier_serial_thread_single",
["not ok 1 -", "ok 1 -"],
timeout=10,
)
assert ret == 1
def test_condvar_1(self, p):
ret = p.sendCommand(
"uv_run_tests condvar_1", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_condvar_2(self, p):
ret = p.sendCommand(
"uv_run_tests condvar_2", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_condvar_3(self, p):
ret = p.sendCommand(
"uv_run_tests condvar_3", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_condvar_4(self, p):
ret = p.sendCommand(
"uv_run_tests condvar_4", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_condvar_5(self, p):
ret = p.sendCommand(
"uv_run_tests condvar_5", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_semaphore_1(self, p):
ret = p.sendCommand(
"uv_run_tests semaphore_1", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_semaphore_2(self, p):
ret = p.sendCommand(
"uv_run_tests semaphore_2", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_semaphore_3(self, p):
ret = p.sendCommand(
"uv_run_tests semaphore_3", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer(self, p):
ret = p.sendCommand("uv_run_tests timer", ["not ok 1 -", "ok 1 -"], timeout=10)
assert ret == 1
def test_timer_init(self, p):
ret = p.sendCommand(
"uv_run_tests timer_init", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_again(self, p):
ret = p.sendCommand(
"uv_run_tests timer_again", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_start_twice(self, p):
ret = p.sendCommand(
"uv_run_tests timer_start_twice", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_order(self, p):
ret = p.sendCommand(
"uv_run_tests timer_order", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_huge_timeout(self, p):
ret = p.sendCommand(
"uv_run_tests timer_huge_timeout", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_huge_repeat(self, p):
ret = p.sendCommand(
"uv_run_tests timer_huge_repeat", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_run_once(self, p):
ret = p.sendCommand(
"uv_run_tests timer_run_once", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_from_check(self, p):
ret = p.sendCommand(
"uv_run_tests timer_from_check", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_is_closing(self, p):
ret = p.sendCommand(
"uv_run_tests timer_is_closing", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_null_callback(self, p):
ret = p.sendCommand(
"uv_run_tests timer_null_callback", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_timer_early_check(self, p):
ret = p.sendCommand(
"uv_run_tests timer_early_check", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_loop_handles(self, p):
ret = p.sendCommand(
"uv_run_tests loop_handles", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_walk_handles(self, p):
ret = p.sendCommand(
"uv_run_tests walk_handles", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_active(self, p):
ret = p.sendCommand("uv_run_tests active", ["not ok 1 -", "ok 1 -"], timeout=10)
assert ret == 1
def test_embed(self, p):
if p.board in ["sim"]:
pytest.skip("unsupported at %s" % p.board)
ret = p.sendCommand("uv_run_tests embed", ["not ok 1 -", "ok 1 -"], timeout=10)
assert ret == 1
@pytest.mark.skip(reason="VELAPLATFO-6346")
def test_async(self, p):
if p.ci:
pytest.skip("unsupported at %s" % p.board)
if p.board in ["sim", "vela"]:
pytest.skip("unsupported at %s" % p.board)
ret = p.sendCommand("uv_run_tests async", ["not ok 1 -", "ok 1 -"], timeout=10)
assert ret == 1
def test_async_null_cb(self, p):
if p.ci:
pytest.skip("unsupported at %s" % p.board)
if p.board in ["sim", "vela"]:
pytest.skip("unsupported at %s" % p.board)
ret = p.sendCommand(
"uv_run_tests async_null_cb", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_homedir(self, p):
ret = p.sendCommand(
"uv_run_tests homedir", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_tmpdir(self, p):
ret = p.sendCommand("uv_run_tests tmpdir", ["not ok 1 -", "ok 1 -"], timeout=10)
assert ret == 1
def test_hrtime(self, p):
ret = p.sendCommand("uv_run_tests hrtime", ["not ok 1 -", "ok 1 -"], timeout=25)
assert ret == 1
def test_gettimeofday(self, p):
ret = p.sendCommand(
"uv_run_tests gettimeofday", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_poll_oob(self, p):
if p.board in ["sim"]:
pytest.skip("unsupported at %s" % p.board)
ret = p.sendCommand(
"uv_run_tests poll_oob", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_threadpool_queue_work_simple(self, p):
ret = p.sendCommand(
"uv_run_tests threadpool_queue_work_simple",
["not ok 1 -", "ok 1 -"],
timeout=10,
)
assert ret == 1
def test_threadpool_queue_work_einval(self, p):
ret = p.sendCommand(
"uv_run_tests threadpool_queue_work_einval",
["not ok 1 -", "ok 1 -"],
timeout=10,
)
assert ret == 1
def test_threadpool_cancel_getnameinfo(self, p):
ret = p.sendCommand(
"uv_run_tests threadpool_cancel_getnameinfo",
["not ok 1 -", "ok 1 -"],
timeout=10,
)
assert ret == 1
def test_threadpool_cancel_random(self, p):
ret = p.sendCommand(
"uv_run_tests threadpool_cancel_random",
["not ok 1 -", "ok 1 -"],
timeout=10,
)
assert ret == 1
def test_threadpool_cancel_work(self, p):
ret = p.sendCommand(
"uv_run_tests threadpool_cancel_work", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_threadpool_cancel_single(self, p):
ret = p.sendCommand(
"uv_run_tests threadpool_cancel_single",
["not ok 1 -", "ok 1 -"],
timeout=10,
)
assert ret == 1
def test_thread_local_storage(self, p):
ret = p.sendCommand(
"uv_run_tests thread_local_storage", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_thread_stack_size(self, p):
ret = p.sendCommand(
"uv_run_tests thread_stack_size", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_thread_mutex(self, p):
ret = p.sendCommand(
"uv_run_tests thread_mutex", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_thread_mutex_recursive(self, p):
ret = p.sendCommand(
"uv_run_tests thread_mutex_recursive", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_thread_rwlock(self, p):
ret = p.sendCommand(
"uv_run_tests thread_rwlock", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_thread_rwlock_trylock(self, p):
ret = p.sendCommand(
"uv_run_tests thread_rwlock_trylock", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_thread_create(self, p):
ret = p.sendCommand(
"uv_run_tests thread_create", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_thread_equal(self, p):
ret = p.sendCommand(
"uv_run_tests thread_equal", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_queue_foreach_delete(self, p):
ret = p.sendCommand(
"uv_run_tests queue_foreach_delete", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_random_async(self, p):
if p.ci:
pytest.skip("unsupported at %s" % p.board)
ret = p.sendCommand(
"uv_run_tests random_async", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_random_sync(self, p):
if p.ci:
pytest.skip("unsupported at %s" % p.board)
ret = p.sendCommand(
"uv_run_tests random_sync", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_handle_type_name(self, p):
ret = p.sendCommand(
"uv_run_tests handle_type_name", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_req_type_name(self, p):
ret = p.sendCommand(
"uv_run_tests req_type_name", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_utf8_decode1(self, p):
ret = p.sendCommand(
"uv_run_tests utf8_decode1", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1
def test_utf8_decode1_overrun(self, p):
ret = p.sendCommand(
"uv_run_tests utf8_decode1_overrun", ["not ok 1 -", "ok 1 -"], timeout=10
)
assert ret == 1

23
tools/ci/testrun/script/test_open_posix/__init__.py
View file

@ -1,23 +0,0 @@
#!/usr/bin/python3
############################################################################
# tools/ci/testrun/script/test_open_posix/__init__.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8

11205
tools/ci/testrun/script/test_open_posix/test_openposix_.py
View file

File diff suppressed because it is too large Load diff

23
tools/ci/testrun/script/test_os/__init__.py
View file

@ -1,23 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/script/test_os/__init__.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8

109
tools/ci/testrun/script/test_os/test_os.py
View file

@ -1,109 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/script/test_os/test_os.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8
import os
import pytest
pytestmark = [pytest.mark.common, pytest.mark.qemu]
do_not_support = ["sabre-6quad", "rv-virt", "rv-virt64", "esp32c3-devkit", "bl602evb"]
def test_ostest(p):
if p.board == "sim":
os.mkdir("./test")
ret = p.sendCommand("mount -t hostfs -o fs=./test /data")
ret = p.sendCommand("ostest", "Exiting with status 0", timeout=300)
assert ret == 0
def test_mm(p):
if p.board in do_not_support:
pytest.skip("unsupported at {}".format(p.board))
ret = p.sendCommand("heap", "TEST COMPLETE", timeout=120)
assert ret == 0
def test_cxxtest(p):
if p.board in do_not_support:
pytest.skip("unsupported at {}".format(p.board))
ret = p.sendCommand("cxxtest", "Test std::map")
assert ret == 0
def test_scanftest(p):
if p.board in do_not_support:
pytest.skip("unsupported at {}".format(p.board))
ret = p.sendCommand("scanftest", "FAILED: 0")
assert ret == 0
def test_getprime(p):
if p.board in ["rv-virt", "rv-virt64"]:
pytest.skip("unsupported at {}".format(p.board))
ret = p.sendCommand("getprime", "getprime took")
assert ret == 0
def test_stdio(p):
if p.board in do_not_support:
pytest.skip("unsupported at {}".format(p.board))
ret = p.sendCommand("fopencookie_test", "fopencokie tests were successful.")
assert ret == 0
ret = p.sendCommand("fmemopen_test", "FAILED: 0")
assert ret == 0
ret = p.sendCommand("open_memstream_test", "FAILED: 0")
assert ret == 0
def test_stdbit(p):
if p.board in do_not_support:
pytest.skip("unsupported at {}".format(p.board))
ret = p.sendCommand("stdbit_test", "stdbit tests: SUCCESSFUL: 14; FAILED: 0")
assert ret == 0
@pytest.mark.run(order=-2)
def test_fs_test(p):
if p.board in do_not_support:
pytest.skip("unsupported at {}".format(p.board))
fstest_dir = "{}/{}_fstest".format(p.fs, p.core)
p.sendCommand("mkdir %s" % fstest_dir)
ret = p.sendCommand("fstest -n 10 -m %s" % fstest_dir, "FAILED: 0", timeout=2000)
p.sendCommand("ls %s" % fstest_dir)
p.sendCommand("rmdir %s" % fstest_dir)
if p.board == "sim":
os.rmdir("./test")
assert ret == 0
@pytest.mark.run(order=-1)
def test_psram_test(p):
if p.board in do_not_support:
pytest.skip("unsupported at {}".format(p.board))
if p.sendCommand("ls /", "tmp/") == 0:
ret = p.sendCommand("fstest -n 10 -m /tmp", "Final memory usage", timeout=500)
p.sendCommand("ls /tmp")
assert ret == 0

23
tools/ci/testrun/utils/__init__.py
View file

@ -1,23 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/utils/__init__.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
# encoding: utf-8

492
tools/ci/testrun/utils/common.py
View file

@ -1,492 +0,0 @@
#!/usr/bin/env python3
############################################################################
# tools/ci/testrun/utils/common.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
import os
import re
import subprocess
import time
from enum import Enum
import pexpect
import pexpect.fdpexpect
import pexpect.spawnbase
import serial
rootPath = os.path.dirname(os.path.abspath(__file__))
tmp_read_nonblocking = pexpect.spawnbase.SpawnBase.read_nonblocking
def enhanced_read_nonblocking(self, size=1, timeout=None):
return re.sub(
r"(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]",
"",
tmp_read_nonblocking(self, size, timeout).decode(errors="ignore"),
).encode()
pexpect.spawnbase.SpawnBase.read_nonblocking = enhanced_read_nonblocking
class StatusCodeEnum(Enum):
NORMAL = (0, "Normal")
TIMEOUT_ERR = (-1, "Timeout")
EOF_ERR = (-2, "EOF")
CRASH_ERR = (-3, "Crash happened")
BUSYLOOP_ERR = (-4, "Busy loop happened")
UNKNOWN_ERR = (-5, "Unknown")
@staticmethod
def get_enum_msg_by_code(status_code):
for status in StatusCodeEnum:
if status.value[0] == status_code:
return status.value[1]
class connectNuttx(object):
def __init__(
self,
board,
vela_path,
dev,
log_path,
fs,
ota_version,
core,
sudo,
ci,
method,
target,
):
self.board = board
self.path = vela_path
self.dev = dev
self.log_path = log_path
self.fs = fs
self.version = ota_version
self.sudo = sudo
self.ci = ci
self.core = core
self.method = method
self.target = target
self.enter = "\r"
self.debug_flag = 0
self.format_str_len = 105
# get PROMPT value and rate value
self.PROMPT = getConfigValue(
self.path, self.board, core=self.core, flag="NSH_PROMPT_STRING"
)
self.rate = getConfigValue(
self.path, self.board, core=self.core, flag="UART0_BAUD"
)
if not os.path.exists(self.log_path):
os.makedirs(self.log_path)
def setup(self):
self.start_time = time.strftime("%Y%m%d_%H%M%S", time.localtime())
print("============ start at %s ============" % self.start_time)
if self.target == "sim":
# start nuttx for sim
start.startNuttx(self, self.path, self.log_path, self.board, self.sudo)
if self.target == "qemu":
start.startQemu(self, self.path, self.log_path, self.board, self.sudo)
if self.target in ["target", "module"]:
# start minicom
if self.method == "minicom":
start.startMinicom(
self, self.dev, self.board, self.log_path, self.core, self.rate
)
else:
start.startSerial(
self, self.dev, self.board, self.log_path, self.core, self.rate
)
def sendcontrol(self, char):
char = char.lower()
a = ord(char)
if 97 <= a <= 122:
a = a - ord("a") + 1
byte = bytes([a])
return byte
d = {
"@": 0,
"`": 0,
"[": 27,
"{": 27,
"\\": 28,
"|": 28,
"]": 29,
"}": 29,
"^": 30,
"~": 30,
"_": 31,
"?": 127,
}
if char not in d:
return b""
byte = bytes(d[char])
return byte
def sendControlCmd(self, cmd, expect="ap>", timeout=10):
if self.method == "minicom":
self.process.sendcontrol(cmd)
time.sleep(1)
self.process.sendcontrol(cmd)
else:
byte = self.sendcontrol(cmd)
time.sleep(1)
self.process.send(byte)
time.sleep(1)
self.process.send(byte)
def print_format_str(self, string, type="text"):
str_prefix = "+"
str_suffix = "+"
if type == "head":
rest_char_len = self.format_str_len - 2 - len(string)
half_len = int(rest_char_len / 2)
print(
str_prefix
+ "-" * half_len
+ string
+ "-" * (rest_char_len - half_len)
+ str_suffix
)
elif type == "tail":
rest_char_len = self.format_str_len - 2
print(str_prefix + "-" * rest_char_len + str_suffix)
elif type == "text":
str_prefix = "| "
str_suffix = " |"
rest_char_len = (
self.format_str_len - len(str_prefix) - len(str_suffix) - len(string)
)
print(
str_prefix
+ string
+ " " * (1 if rest_char_len < 1 else rest_char_len)
+ str_suffix
)
else:
print(string)
def clean_buffer(self):
i = -1
while True:
if (
(
self.process.before is not None
and self.process.before.decode(errors="ignore")
.replace("\r", "")
.replace("\n", "")
!= ""
)
or (
self.process.after is not None
and self.process.after != pexpect.TIMEOUT
and self.process.after.decode(errors="ignore")
.replace("\r", "")
.replace("\n", "")
!= ""
)
or i == 0
):
i = self.process.expect(
[re.compile(b".+"), pexpect.TIMEOUT, pexpect.EOF], timeout=0.1
)
else:
while True:
try:
self.process.read_nonblocking(
size=self.process.maxread, timeout=0.1
)
except Exception:
break
self.process.before = b""
self.process.after = b""
break
# send command to nsh
def sendCommand(self, cmd, *argc, **argv):
expect = []
timeout = 10
ret = StatusCodeEnum.NORMAL.value[0]
length = len(argc)
if length == 0:
expect.append(self.PROMPT)
else:
for i in argc:
expect.append(i)
length = len(argv)
if length != 0:
for key, value in argv.items():
if key == "timeout":
timeout = value
if self.method != "minicom":
time.sleep(0.5)
if self.target == "qemu":
self.clean_buffer()
self.process.sendline(cmd)
else:
self.clean_buffer()
self.process.sendline(cmd)
time.sleep(0.1)
self.process.send("\r\n\r\n")
try:
for i in expect:
ret = self.process.expect(i, timeout=timeout)
except Exception as e:
self.print_format_str(" Catch Exception ", type="head")
if isinstance(e, pexpect.TIMEOUT):
ret = StatusCodeEnum.TIMEOUT_ERR.value[0]
elif isinstance(e, pexpect.EOF):
ret = StatusCodeEnum.EOF_ERR.value[0]
self.print_format_str(f"An pexpect.EOF error occurred: {str(e)}")
else:
ret = StatusCodeEnum.UNKNOWN_ERR.value[0]
self.print_format_str(f"An unexpected error occurred: {str(e)}")
self.print_format_str(" Result ", type="head")
self.print_format_str(f"Command : '{cmd}'")
self.print_format_str(f"Expect value: {str(expect)}")
self.print_format_str(f"Timeout : {timeout}s")
self.print_format_str(
f"Test result : {StatusCodeEnum.get_enum_msg_by_code(ret)}"
)
self.print_format_str("", type="tail")
finally:
self.debug(cmd, ret)
if self.method != "minicom":
time.sleep(0.5)
return ret
def switch_to_original_core(self):
if self.target == "target":
self.sendControlCmd("c")
if self.core != "ap":
self.process.sendline("cu -l /dev/tty%s\n" % self.core.upper())
self.process.expect_exact(self.PROMPT)
def debug(self, cmd, ret):
if self.debug_flag:
print("********************* DEBUG START ********************")
if cmd == "\n":
cmd = r"\n"
print("cmd: {}".format(cmd))
print("ret: {}".format(ret))
print("before: {}".format(self.process.before.decode(errors="ignore")))
print("after: {}".format(self.process.after.decode(errors="ignore")))
print("buffer: {}".format(self.process.buffer.decode(errors="ignore")))
print("********************** DEBUG END **********************")
def cleanup(self):
if self.target == "sim":
self.process.sendline("poweroff")
if self.target == "qemu":
self.sendControlCmd("a", self.PROMPT)
self.process.sendline("x")
class start:
def startMinicom(self, dev, board, log_path, core, rate):
self.log = "{}/{}_{}.cap".format(log_path, dev[-4:], self.start_time)
self.process = pexpect.spawn(
r"sudo minicom -D {} -b {} -o -C {}".format(dev, rate, self.log),
maxread=200000,
)
self.process.expect("Welcome to minicom")
self.switch_to_original_core()
def startSerial(self, dev, board, log_path, core, rate):
self.log = "{}/{}_{}.cap".format(log_path, dev[-4:], self.start_time)
self.logFile = open(self.log, "ab+")
self.set = serial.Serial(port=dev, baudrate=int(rate))
self.process = pexpect.fdpexpect.fdspawn(
self.set, "wb", maxread=20000, logfile=self.logFile
)
self.switch_to_original_core()
def startNuttx(self, path, log_path, board="sim", sudo=False):
os.chdir(path)
self.log = "{}/{}_{}.log".format(log_path, board, self.start_time)
if sudo:
if board in ["sim_rpserver", "sim_rpproxy"]:
os.chdir(path)
self.process = pexpect.spawn(
"bash", ["-c", "sudo ./rpserver/nuttx/nuttx | tee %s" % self.log]
)
else:
self.process = pexpect.spawn(
"bash", ["-c", "sudo ./nuttx | tee %s" % self.log]
)
else:
self.process = pexpect.spawn("bash", ["-c", "./nuttx | tee %s" % self.log])
self.process.expect(self.PROMPT)
def startQemu(self, path, log_path, board="qemu", sudo=False):
os.chdir(path)
self.log = "{}/{}_{}.log".format(log_path, board, self.start_time)
flag1 = getConfigValue(path, board, core=None, flag="ARCH_CHIP")
if flag1 == "imx6":
self.process = pexpect.spawn(
"bash",
[
"-c",
"qemu-system-arm -semihosting -M sabrelite -m 1024 -smp 4 -kernel ./nuttx -nographic | tee %s"
% self.log,
],
)
if flag1 == "qemu-rv":
flag2 = getConfigValue(path, board, core=None, flag="ARCH_RV64")
options = ""
if flag2:
riscv = "qemu-system-riscv64"
else:
riscv = "qemu-system-riscv32"
fs_flag = getConfigValue(path, board, core=None, flag="DRIVERS_VIRTIO_BLK")
if fs_flag:
os.system("dd if=/dev/zero of=fatfs.img bs=512 count=128K")
os.system("mkfs.fat fatfs.img")
os.system("chmod 777 ./fatfs.img")
options = (
"-drive index=0,id=userdata,if=none,format=raw,file=./fatfs.img "
"-device virtio-blk-device,bus=virtio-mmio-bus.0,drive=userdata"
)
self.process = pexpect.spawn(
"bash",
[
"-c",
"%s -M virt -bios ./nuttx -nographic %s | tee %s"
% (riscv, options, self.log),
],
)
self.process.expect(self.PROMPT)
def runCmd(cmd):
p = subprocess.Popen(
cmd,
shell=True,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout, stderr = p.communicate()
recode = p.returncode
if recode != 0:
print("Debug: run command '%s' failed" % cmd)
return stdout
# find file
def findFile(path, flag, board, core=None, match=True):
fList = []
for root, dirs, files in os.walk(path):
for name in files:
if not match and flag in name:
fList.append(os.path.join(root, name))
if match and name == flag and ".o" not in name:
fList.append(os.path.join(root, name))
if len(fList) != 1:
fList = [x for x in fList if board in x and core + "/" in x]
print(fList)
return fList
# get CONFIG_NSH_PROMPT_STRING value
def getConfigValue(path, board, core, flag):
value = ""
l1 = findFile(path, ".config", board, core=core)
with open(l1[0], "r+") as f:
lines = f.readlines()
f.close()
print(lines)
print(flag)
for line in lines:
if flag + "=" in line:
value = line.split("=")[1]
if '"' in value:
value = value.strip('"').strip()
print(value)
return value
# read log and extract timestamp
def getTimestamp(filename, str1, str2, pattern):
with open(filename, "r") as f:
buff = f.read()
pat = re.compile(str1 + "(.*?)" + str2, re.S)
result = pat.findall(buff)
timestamp = re.findall(pattern, str(result)) # extract timestamp
return timestamp
# calculate the time difference for each time
def getDif(timestamp, rateANDinterance):
lst = []
for i in range(1, len(timestamp)):
difts = int(timestamp[i]) - int(timestamp[i - 1])
float_num = (difts / 1000000) - rateANDinterance
lst.append(float_num)
return lst
# read log
def getLog(filename, str1, str2):
l1 = []
with open(filename, "r", encoding="utf-8", errors="ignore") as f:
buff = f.read()
f.close()
pat = re.compile(str1 + "(.*?)" + str2, re.S)
result = pat.findall(buff)
if "gtest" in str1:
return result[0].strip("\n")
else:
# add line to l1
for i in result[0].strip("\n").split("\n"):
l1.append(i)
return l1
def rmfile(p, core, file):
if p.core == core:
p.sendCommand("rm -r" + file)
else:
p.process.sendline("cu -l /dev/tty" + core.upper())
p.sendCommand("\n", core, flag=core + ">")
p.sendCommand("rm -r " + file, core, flag=core + ">")
if p.core == "ap":
p.sendControlCmd("c")
else:
p.switch_to_original_core(p)

257
tools/ci/testrun/utils/data_model.py
View file

@ -1,257 +0,0 @@
############################################################################
# tools/ci/testrun/utils/data_model.py
#
# SPDX-License-Identifier: Apache-2.0
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership. The
# ASF licenses this file to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance with the
# License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
############################################################################
import re
from datetime import datetime
from typing import Dict, List
"""
cmocka.json
"""
Passed = "Passed"
Failed = "Failed"
Unexecuted = "Unexecuted"
class CaseInfo:
def __init__(self, test_suite_name, test_case_name, status, log=None):
self.test_suite_name = test_suite_name
# case name，e.g. "TestNuttxMm01"
self.test_case_name = test_case_name
# result: Passed or Failed
self.status = status
# log
self.log: List = [""] if log is None else log
class SuiteInfo:
def __init__(self, test_suite_name):
# suite name, e.g. "NuttxMmTestSuites"
self.test_suite_name = test_suite_name
# all test cases in the current test suite
self.test_cases: Dict[str, CaseInfo] = dict()
# number of cases passed in the current test suites
self.passed_count = 0
# number of cases failed in the current test suites
self.failed_count = 0
# case run count
self.run_count = 0
# unexecuted count
self.unexecuted_count = 0
# number of cases in the current test suites
self.cases_count = 0
# suite run flag
self.is_suite_run = False
class CmockaSummary:
def __init__(self, duration=0):
# number of all test suites
self.total_suites_count = 0
# all test cases number
self.total_cases_count = 0
# number of all passed cases
self.total_passed_count = 0
# number of all failed cases
self.total_failed_count = 0
# number of all unknown cases
self.total_unexecuted_count = 0
# duration
self.duration = duration
class CmockaSingleCoreRecord:
def __init__(self, lines, core="", board="", log="", duration=0):
# create time
self.create_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# core
self.core = "" if core is None else core
# board
self.board = "" if board is None else board
# cmocka info
self.test_suites: Dict[str, SuiteInfo] = dict()
# summary
self.summary = CmockaSummary(duration)
# log path
self.log = "" if log is None else log
# bad_case
self.bad_case_tip = ""
suite_pattern = r"\] (?P<test_suite_name>[a-zA-Z]*TestSuites)"
case_pattern = r"\]\s+(?P<test_case_name>TestNuttx\w+)"
lines_iter = iter(lines)
current_suite = None
while True:
try:
line = next(lines_iter)
if (suite_match := re.search(suite_pattern, line)) is not None:
current_suite = suite_match.group("test_suite_name")
elif (
current_suite is not None
and (case_match := re.search(case_pattern, line)) is not None
):
current_case = case_match.group("test_case_name")
self.append(CaseInfo(current_suite, current_case, Unexecuted))
except StopIteration:
break
def append(self, object: CaseInfo):
suite: SuiteInfo = self.test_suites.get(object.test_suite_name)
if suite is None:
suite = SuiteInfo(object.test_suite_name)
self.test_suites.update({object.test_suite_name: suite})
suite.test_cases.update({object.test_case_name: object})
passed_count = 0
failed_count = 0
unexecuted_count = 0
test_case: CaseInfo
for test_case in suite.test_cases.values():
if test_case.status == Passed:
passed_count += 1
elif test_case.status == Failed:
failed_count += 1
else:
unexecuted_count += 1
suite.passed_count = passed_count
suite.failed_count = failed_count
suite.unexecuted_count = unexecuted_count
suite.run_count = passed_count + failed_count
suite.cases_count = passed_count + failed_count + unexecuted_count
if passed_count + failed_count != 0:
suite.is_suite_run = True
total_passed_count = 0
total_failed_count = 0
total_unexecuted_count = 0
total_cases_count = 0
suite: SuiteInfo
for suite in self.test_suites.values():
total_passed_count += suite.passed_count
total_failed_count += suite.failed_count
total_unexecuted_count += suite.unexecuted_count
total_cases_count += suite.cases_count
self.summary.total_passed_count = total_passed_count
self.summary.total_failed_count = total_failed_count
self.summary.total_unexecuted_count = total_unexecuted_count
self.summary.total_cases_count = total_cases_count
self.summary.total_suites_count = len(self.test_suites)
def process(self, lines, err_code):
# regular expression
suite_start_pattern = r"\] (?P<test_suite_name>[a-zA-Z]*TestSuites): Running (?P<cases_count>\d+) test\(s\)"
case_run_pattern = r"\[\s+RUN\s+\] (?P<test_case_name>TestNuttx\w+)"
case_pass_pattern = r"\[\s+OK\s+\] (?P<test_case_name>TestNuttx\w+)"
case_fail_pattern = r"\[\s+FAILED\s+\] (?P<test_case_name>TestNuttx\w+)"
lines_iter = iter(lines)
line = next(lines_iter)
while True:
try:
interrupt_flag = False
# matching new test suites
if (
suite_start_match := re.search(suite_start_pattern, line)
) is not None:
test_suite_name = suite_start_match.group("test_suite_name")
cases_count = int(suite_start_match.group("cases_count"))
suite_end_pattern = r"{}: {} test(s) run.".format(
test_suite_name, cases_count
)
line = next(lines_iter)
while True:
if (
case_run_match := re.search(case_run_pattern, line)
) is not None:
test_case_name = case_run_match.group("test_case_name")
log = [line]
while True:
try:
line = next(lines_iter)
log.append(line)
except StopIteration:
self.append(
CaseInfo(
test_suite_name, test_case_name, Failed, log
)
)
if err_code == -3:
self.bad_case_tip = f"This case was not executed, \
because crash occurred after running '{test_case_name}'."
elif err_code == -4:
self.bad_case_tip = f"This case was not executed, \
because no response for a long time after running '{test_case_name}'."
elif err_code == -1:
self.bad_case_tip = f"This case was not executed, \
because the maximum waiting time has been exceeded \
while running '{test_case_name}'."
else:
self.bad_case_tip = "This case was not executed due to unknown reasons."
raise StopIteration
if re.search(case_pass_pattern, line) is not None:
self.append(
CaseInfo(
test_suite_name, test_case_name, Passed, log
)
)
break
elif re.search(case_fail_pattern, line) is not None:
self.append(
CaseInfo(
test_suite_name, test_case_name, Failed, log
)
)
break
elif re.search(suite_start_pattern, line) is not None:
self.append(
CaseInfo(
test_suite_name, test_case_name, Failed, log
)
)
interrupt_flag = True
break
elif suite_end_pattern in line:
break
if interrupt_flag:
break
line = next(lines_iter)
if interrupt_flag:
continue
line = next(lines_iter)
except StopIteration:
break