154 lines
8.5 KiB
Python
154 lines
8.5 KiB
Python
"""Real pinned esptool CLI/resources; transport doubles and disposable PTYs only."""
|
|
import importlib.metadata
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
import archy_esptool as wrapper
|
|
from archy_rnodeconf import esptool_v5_args
|
|
|
|
|
|
class EsptoolV5Tests(unittest.TestCase):
|
|
def test_reviewed_version_and_no_ecdsa_dependency(self):
|
|
self.assertEqual(importlib.metadata.version('esptool'), wrapper.ESPTOOL_VERSION)
|
|
self.assertFalse(any(req.lower().startswith('ecdsa') for req in importlib.metadata.requires('esptool')))
|
|
with self.assertRaises(importlib.metadata.PackageNotFoundError):
|
|
importlib.metadata.version('ecdsa')
|
|
|
|
def test_actual_s3_stub_without_serial_open_and_missing_resource_refuses(self):
|
|
import serial
|
|
esptool = wrapper.load_esptool()
|
|
from esptool.loader import StubFlasher
|
|
with patch.object(serial.Serial, 'open', side_effect=AssertionError('no serial access')):
|
|
self.assertEqual(wrapper.check_stub(), '5.6.0')
|
|
with tempfile.TemporaryDirectory() as empty, patch.object(StubFlasher, 'STUB_DIR', empty):
|
|
with self.assertRaises(esptool.FatalError): wrapper.check_stub()
|
|
with patch.object(esptool, '__version__', '4.8.1'):
|
|
with self.assertRaisesRegex(RuntimeError, 'reviewed pin'): wrapper.check_stub()
|
|
|
|
def test_rns_legacy_spelling_keeps_offsets_paths_and_reset_modes(self):
|
|
args = ['--chip', 'esp32s3', '--port', '/dev/tty_fake_name', '--baud', '921600',
|
|
'--before', 'default_reset', '--after', 'hard_reset', 'write_flash', '-z',
|
|
'--flash_mode', 'dio', '--flash_freq', '80m', '--flash_size', '8MB',
|
|
'0x0', '/cache/boot_loader.bin', '0x10000', 'write_flash']
|
|
self.assertEqual(esptool_v5_args(args), [
|
|
'--chip', 'esp32s3', '--port', '/dev/tty_fake_name', '--baud', '921600',
|
|
'--before', 'default-reset', '--after', 'hard-reset', 'write-flash', '-z',
|
|
'--flash-mode', 'dio', '--flash-freq', '80m', '--flash-size', '8MB',
|
|
'0x0', '/cache/boot_loader.bin', '0x10000', 'write_flash'])
|
|
|
|
def test_actual_cli_write_options_reset_and_failure_status(self):
|
|
esptool = wrapper.load_esptool()
|
|
# Real parser, prepare/teardown, reset dispatcher and file handling.
|
|
# Only the hardware operations are replaced.
|
|
chip = Mock(CHIP_NAME='ESP32-S3', secure_download_mode=False)
|
|
chip._port.port = '/dev/fixture-only'
|
|
chip.get_chip_features.return_value = ['Wi-Fi']
|
|
chip.get_usb_mode.return_value = None
|
|
for legacy in (False, True):
|
|
for failure in (False, True):
|
|
with self.subTest(legacy=legacy, failure=failure), tempfile.TemporaryDirectory() as temporary:
|
|
file = Path(temporary) / 'firmware_image.bin'
|
|
file.write_bytes(bytes(range(32)))
|
|
args = ['--chip', 'esp32s3', '--port', '/dev/fixture-only', '--baud', '460800']
|
|
if legacy:
|
|
args += ['--before', 'default_reset', '--after', 'hard_reset',
|
|
'write_flash', '-z', '--flash_mode', 'dio', '--flash_freq', '80m',
|
|
'--flash_size', '8MB', '0x0', str(file)]
|
|
args = esptool_v5_args(args)
|
|
else: args += ['write-flash', '--erase-all', '0x0', str(file)]
|
|
chip.reset_mock()
|
|
observed = []
|
|
def write(target, files, **kwargs):
|
|
observed.append((target, [(offset, f.read()) for offset, f in files], kwargs))
|
|
if failure: raise esptool.FatalError('fixture write failed')
|
|
with patch.object(esptool, 'connect_esp', return_value=chip) as connect, \
|
|
patch.object(esptool, 'run_stub', return_value=chip) as stub, \
|
|
patch.object(esptool, 'read_mac'), patch.object(esptool, 'attach_flash') as attach, \
|
|
patch.object(esptool, 'write_flash', side_effect=write):
|
|
if failure:
|
|
with self.assertRaises(SystemExit) as caught: wrapper.run_esptool(args)
|
|
self.assertEqual(caught.exception.code, 2)
|
|
else: wrapper.run_esptool(args)
|
|
self.assertEqual(connect.call_args.kwargs['before'], 'default-reset')
|
|
self.assertEqual(connect.call_args.kwargs['port'], '/dev/fixture-only')
|
|
stub.assert_called_once_with(chip, plugins=None)
|
|
attach.assert_called_once()
|
|
self.assertEqual(observed[0][1], [(0, file.read_bytes())])
|
|
self.assertEqual(observed[0][2]['erase_all'], not legacy)
|
|
if legacy:
|
|
self.assertEqual(observed[0][2]['flash_size'], '8MB')
|
|
self.assertTrue(observed[0][2]['compress'])
|
|
chip.change_baud.assert_called_once_with(460800)
|
|
chip.hard_reset.assert_called_once()
|
|
chip._port.close.assert_called_once()
|
|
|
|
def test_actual_plain_progress_has_complete_lines_and_decimal_percent(self):
|
|
esptool = wrapper.load_esptool()
|
|
output = io.StringIO()
|
|
esptool.log.set_console_options(file=output)
|
|
try:
|
|
esptool.log.progress_bar(423, 1000, prefix='Writing at 0x00001000 ')
|
|
esptool.log.progress_bar(1000, 1000, prefix='Writing at 0x00002000 ')
|
|
finally: esptool.log.set_console_options(file=None)
|
|
value = output.getvalue()
|
|
self.assertNotIn('\x1b', value)
|
|
self.assertNotIn('\r', value)
|
|
self.assertEqual(len(value.splitlines()), 2)
|
|
self.assertIn('42.3%', value)
|
|
self.assertIn('100.0%', value)
|
|
|
|
def test_owned_import_order_actual_loader_never_reopens_or_enumerates_host(self):
|
|
code = textwrap.dedent('''
|
|
import fcntl, os, termios
|
|
from unittest.mock import patch
|
|
import archy_owned_serial
|
|
master, slave = os.openpty()
|
|
fcntl.ioctl(slave, termios.TIOCEXCL)
|
|
path = '/dev/ttyUSB-fixture-not-present'
|
|
try:
|
|
archy_owned_serial.install(path, slave)
|
|
import archy_esptool
|
|
esptool = archy_esptool.load_esptool()
|
|
import esp_pylib.serial_ports
|
|
from esptool.targets.esp32s3 import ESP32S3ROM
|
|
assert esp_pylib.serial_ports.get_port_names() == [path]
|
|
with patch.object(os, 'open', side_effect=AssertionError('path reopen forbidden')):
|
|
chip = ESP32S3ROM(path)
|
|
assert os.fstat(chip._port.fd).st_ino == os.fstat(slave).st_ino
|
|
chip._port.write(b'fixture')
|
|
assert os.read(master, 7) == b'fixture'
|
|
chip._port.close()
|
|
assert os.isatty(slave)
|
|
try: ESP32S3ROM('/dev/other-device')
|
|
except esptool.FatalError: pass
|
|
else: raise AssertionError('different device accepted')
|
|
archy_esptool.run_esptool(['version'])
|
|
finally:
|
|
os.close(master); os.close(slave)
|
|
''')
|
|
result = subprocess.run([sys.executable, '-c', code], cwd=Path(__file__).parent,
|
|
capture_output=True, text=True, timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertIn('5.6.0', result.stdout)
|
|
|
|
def test_real_entrypoints_plain_version_and_invalid_input_fail(self):
|
|
for script, prefix in [('archy_esptool.py', []), ('archy_rnodeconf.py', ['--archy-esptool'])]:
|
|
for args, success in [(['version'], True), (['write-flash', 'invalid'], False),
|
|
(['--after', 'invalid-reset', 'version'], False)]:
|
|
result = subprocess.run([sys.executable, str(Path(__file__).with_name(script)), *prefix, *args],
|
|
capture_output=True, text=True, timeout=20)
|
|
self.assertEqual(result.returncode == 0, success, result.stdout + result.stderr)
|
|
self.assertNotIn('\x1b', result.stdout + result.stderr)
|
|
if success: self.assertIn('5.6.0', result.stdout)
|
|
else: self.assertTrue(result.stderr)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|