Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ OS:
- micropython-nostr: log background relay connection failures at INFO instead of ERROR to avoid REPL pollution breaking file transfers

Development:
- mpos_controller: give --serial-port to mpremote, so file transfers, screenshots and widget trees use the selected device instead of the first device that mpremote finds
- Add Python-level line coverage via sys.settrace (mpcov build variant)
- Add `--coverage` flag to test_runner.py for collecting per-file line coverage
- Add `make build-mpos-unix-coverage` target
Expand Down
97 changes: 71 additions & 26 deletions scripts/mpos_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,45 @@ def _resolve_cwd():
return os.path.normpath(os.path.join(d, "..", "internal_filesystem"))


def _mpremote_cmd(port=None, *rest):
"""
Build an mpremote command, with the port when the port is known.

If the command has no `connect`, mpremote uses `connect auto`. That
selects sorted(comports())[0], which is the first USB serial device by
name. If the host has two devices, that device can be the incorrect one.
Because of this, each caller that knows the port must give the port here.
"""
script_dir = os.path.dirname(os.path.abspath(__file__))
mpremote = os.path.normpath(os.path.join(
script_dir, "..",
"lvgl_micropython/lib/micropython/tools/mpremote/mpremote.py"))
cmd = ["python3", mpremote]
if port:
cmd += ["connect", port]
return cmd + list(rest)


def _count_usb_serial_devices():
"""
Count the attached USB serial devices, or give None if the count is unknown.

The count is unknown when pyserial is missing from *this* interpreter.
That does not mean the host has no devices: _mpremote_cmd() starts
mpremote with `python3`, which can be a different interpreter that does
have pyserial. A missing pyserial therefore gives None, and not 0, so
that the caller does not read it as "the host has no devices".
"""
try:
import serial.tools.list_ports
except ImportError:
return None
return sum(
1 for p in serial.tools.list_ports.comports()
if p.vid is not None and p.pid is not None
)


def _build_test_code(test_path, tests_dir=None):
with open(test_path) as f:
test_content = f.read()
Expand Down Expand Up @@ -1076,11 +1115,8 @@ def _read_remote_file(self, path):
import subprocess, tempfile, os as _os
with tempfile.NamedTemporaryFile(suffix=".raw", delete=False) as tmp:
tmppath = tmp.name
script_dir = _os.path.dirname(_os.path.abspath(__file__))
mpremote = _os.path.join(script_dir, "..",
"lvgl_micropython/lib/micropython/tools/mpremote/mpremote.py")
subprocess.run(
["python3", mpremote, "cp", ":{}".format(path), tmppath],
_mpremote_cmd(self.port, "cp", ":{}".format(path), tmppath),
capture_output=True, timeout=60
)
with open(tmppath, "rb") as f:
Expand All @@ -1094,11 +1130,8 @@ def write_remote_file(self, path, data):
tmppath = tmp.name
with open(tmppath, "wb") as f:
f.write(data)
script_dir = _os.path.dirname(_os.path.abspath(__file__))
mpremote = _os.path.join(script_dir, "..",
"lvgl_micropython/lib/micropython/tools/mpremote/mpremote.py")
subprocess.run(
["python3", mpremote, "cp", tmppath, ":{}".format(path)],
_mpremote_cmd(self.port, "cp", tmppath, ":{}".format(path)),
capture_output=True, timeout=60
)
_os.unlink(tmppath)
Expand Down Expand Up @@ -1222,11 +1255,8 @@ def get_widget_tree(self):
import subprocess, json as _json, tempfile, os as _os
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as tmp:
tmppath = tmp.name
script_dir = _os.path.dirname(_os.path.abspath(__file__))
mpremote = _os.path.join(script_dir, "..",
"lvgl_micropython/lib/micropython/tools/mpremote/mpremote.py")
subprocess.run(
["python3", mpremote, "cp", ":/_mpos_tree.json", tmppath],
_mpremote_cmd(self.port, "cp", ":/_mpos_tree.json", tmppath),
capture_output=True, timeout=15
)
with open(tmppath) as f:
Expand Down Expand Up @@ -1269,28 +1299,23 @@ def display_size(self):
def run_test_file(self, test_path, tests_dir=None, timeout=300):
import subprocess, re
code = _build_test_code(test_path, tests_dir)
mpremote = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "..",
"lvgl_micropython/lib/micropython/tools/mpremote/mpremote.py",
)
host_test_dir = os.path.dirname(os.path.abspath(test_path))
mpk_names = set(re.findall(r"\.\./tests/(com\.micropythonos\.ziptest_[^\"]+\.mpk)", code))
if mpk_names:
subprocess.run(
["python3", mpremote, "connect", self.port, "exec",
"import os; os.mkdir('tests')"],
_mpremote_cmd(self.port, "exec", "import os; os.mkdir('tests')"),
capture_output=True, timeout=15,
)
for name in sorted(mpk_names):
host_mpk = os.path.join(host_test_dir, name)
subprocess.run(
["python3", mpremote, "connect", self.port, "cp",
host_mpk, ":tests/{}".format(name)],
_mpremote_cmd(self.port, "cp",
host_mpk, ":tests/{}".format(name)),
capture_output=True, timeout=60,
)
code = code.replace("../tests/", "tests/")
result = subprocess.run(
["python3", mpremote, "connect", self.port, "exec", code],
_mpremote_cmd(self.port, "exec", code),
capture_output=True, timeout=timeout + 60,
)
out = result.stdout
Expand Down Expand Up @@ -1558,12 +1583,32 @@ def main():
return 1
apppath = args.args[0]
import subprocess, os
script_dir = os.path.dirname(os.path.abspath(__file__))
mpremote = os.path.join(script_dir, "..",
"lvgl_micropython/lib/micropython/tools/mpremote/mpremote.py")
subprocess.run(["python3", mpremote, "mkdir", ":/apps"], capture_output=True)
if not args.serial_port:
device_count = _count_usb_serial_devices()
if device_count is None:
print(
"warning: pyserial is not available here, so the number of "
"USB serial devices is unknown.\n"
" mpremote selects the first device. Give "
"--serial-port to select a device.",
file=sys.stderr,
)
elif device_count > 1:
print(
"error: the host has more than one USB serial device and "
"you gave no --serial-port.\n"
" mpremote would select the first device, which can "
"be the incorrect one.\n"
" Run 'mpremote connect list' and give --serial-port.",
file=sys.stderr,
)
return 1
print("Installing to {}".format(args.serial_port or "the first device"))
subprocess.run(
_mpremote_cmd(args.serial_port, "mkdir", ":/apps"), capture_output=True
)
result = subprocess.run(
["python3", mpremote, "fs", "cp", "-r", apppath, ":/apps/"],
_mpremote_cmd(args.serial_port, "fs", "cp", "-r", apppath, ":/apps/"),
capture_output=True, timeout=60
)
if result.returncode != 0:
Expand Down
68 changes: 61 additions & 7 deletions tests/cpython_mpos_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,11 @@
import subprocess

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from scripts.mpos_controller import MPOSController
from scripts.mpos_controller import (
MPOSController,
_count_usb_serial_devices,
_mpremote_cmd,
)


PASS = 0
Expand Down Expand Up @@ -69,6 +73,7 @@ def run_tests(mpos, only=None, is_serial=False, cli_binary=None, serial_port=Non
"navigation": test_app_navigation,
"appmanagement": test_app_management,
"helpers": test_controller_helpers,
"mpremoteport": test_mpremote_port,
}
if only:
names = [s.strip() for s in only.split(",")]
Expand All @@ -85,6 +90,55 @@ def run_tests(mpos, only=None, is_serial=False, cli_binary=None, serial_port=Non
fn(mpos, is_serial=is_serial, cli_binary=cli_binary, serial_port=serial_port)


def test_mpremote_port(mpos, is_serial=False, cli_binary=None, serial_port=None):
# This test does not use a device, because _mpremote_cmd() only builds a
# list of arguments. It runs with the desktop backend and with a device.
section("mpremote port selection")

cmd = _mpremote_cmd("/dev/ttyACM7", "fs", "cp", "app.py", ":/")
has_connect = "connect" in cmd
check(has_connect, "with a port: the command contains 'connect'")
check(
has_connect and cmd[cmd.index("connect") + 1] == "/dev/ttyACM7",
"with a port: the port comes after 'connect'",
)
check(
cmd[-4:] == ["fs", "cp", "app.py", ":/"],
"with a port: the command keeps the other arguments",
)

# With no port, mpremote connects to the first device that it finds. The
# command must not contain 'connect', because there is no port to give.
cmd = _mpremote_cmd(None, "fs", "cp", "app.py", ":/")
check("connect" not in cmd, "with no port: the command contains no 'connect'")
check(
cmd[-4:] == ["fs", "cp", "app.py", ":/"],
"with no port: the command keeps the other arguments",
)

# A missing pyserial must give None, and not 0. The installapp guard reads
# 0 as "the host has one device or no device", and then does not warn.
# mpremote runs with `python3`, which can be a different interpreter that
# does have pyserial, so 0 would hide a real multi-device condition.
import builtins

real_import = builtins.__import__

def no_serial(name, *a, **k):
if name.split(".")[0] == "serial":
raise ImportError("blocked by the test")
return real_import(name, *a, **k)

builtins.__import__ = no_serial
try:
check(
_count_usb_serial_devices() is None,
"no pyserial: the device count is unknown, and not 0",
)
finally:
builtins.__import__ = real_import


def test_basic(mpos, is_serial=False, cli_binary=None, serial_port=None):
section("Basic exec / eval / multiline")

Expand Down Expand Up @@ -358,13 +412,13 @@ def test_app_management(mpos, is_serial=False, cli_binary=None, serial_port=None
"internal_filesystem/apps", appname)
)

script_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
mpremote = os.path.join(script_dir,
"lvgl_micropython/lib/micropython/tools/mpremote/mpremote.py")

subprocess.run(["python3", mpremote, "mkdir", ":/apps"], capture_output=True)
# Use the port that the caller gave. If mpremote gets no port, it connects
# to the first serial device that it finds, which can be a different device.
subprocess.run(
_mpremote_cmd(serial_port, "mkdir", ":/apps"), capture_output=True
)
result = subprocess.run(
["python3", mpremote, "fs", "cp", "-r", apppath, ":/apps/"],
_mpremote_cmd(serial_port, "fs", "cp", "-r", apppath, ":/apps/"),
capture_output=True, timeout=60
)
check(result.returncode == 0, f"installapp: cp exit code {result.returncode}")
Expand Down