mirror of
https://github.com/meshtastic/python.git
synced 2026-04-19 22:37:21 -04:00
add duplicate check
This commit is contained in:
@@ -6,6 +6,8 @@ import platform
|
|||||||
import subprocess
|
import subprocess
|
||||||
import re
|
import re
|
||||||
|
|
||||||
|
import meshtastic.util
|
||||||
|
|
||||||
# Goal is to detect which device and port to use from the supported devices
|
# Goal is to detect which device and port to use from the supported devices
|
||||||
# without installing any libraries that are not currently in the python meshtastic library
|
# without installing any libraries that are not currently in the python meshtastic library
|
||||||
|
|
||||||
@@ -109,7 +111,7 @@ def get_devices_with_vendor_id(vid):
|
|||||||
sd.add(d)
|
sd.add(d)
|
||||||
return sd
|
return sd
|
||||||
|
|
||||||
def active_ports_on_supported_devices(sds):
|
def active_ports_on_supported_devices(sds, eliminate_duplicates=False):
|
||||||
"""Return a set of active ports based on the supplied supported devices"""
|
"""Return a set of active ports based on the supplied supported devices"""
|
||||||
ports = set()
|
ports = set()
|
||||||
baseports = set()
|
baseports = set()
|
||||||
@@ -170,6 +172,10 @@ def active_ports_on_supported_devices(sds):
|
|||||||
# add all ports
|
# add all ports
|
||||||
for com_port in com_ports:
|
for com_port in com_ports:
|
||||||
ports.add(com_port)
|
ports.add(com_port)
|
||||||
|
if eliminate_duplicates:
|
||||||
|
ports = meshtastic.util.eliminate_duplicate_port(list(ports))
|
||||||
|
ports.sort()
|
||||||
|
ports = set(ports)
|
||||||
return ports
|
return ports
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
82
meshtastic/tests/test_supported_device.py
Normal file
82
meshtastic/tests/test_supported_device.py
Normal file
@@ -0,0 +1,82 @@
|
|||||||
|
"""Meshtastic unit tests for supported_device.py"""
|
||||||
|
|
||||||
|
|
||||||
|
from unittest.mock import patch
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from meshtastic.supported_device import active_ports_on_supported_devices, SupportedDevice
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
@patch('platform.system', return_value='Linux')
|
||||||
|
def test_active_ports_on_supported_devices_empty(mock_platform):
|
||||||
|
"""Test active_ports_on_supported_devices()"""
|
||||||
|
sds = set()
|
||||||
|
assert active_ports_on_supported_devices(sds) == set()
|
||||||
|
mock_platform.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
@patch('subprocess.getstatusoutput')
|
||||||
|
@patch('platform.system', return_value='Linux')
|
||||||
|
def test_active_ports_on_supported_devices_linux(mock_platform, mock_sp):
|
||||||
|
"""Test active_ports_on_supported_devices()"""
|
||||||
|
mock_sp.return_value = (None, 'crw-rw-rw- 1 root wheel 0x9000000 Feb 8 22:22 /dev/ttyUSBfake')
|
||||||
|
fake_device = SupportedDevice(name='a', for_firmware='heltec-v2.1', baseport_on_linux='ttyUSB')
|
||||||
|
fake_supported_devices = [fake_device]
|
||||||
|
assert active_ports_on_supported_devices(fake_supported_devices) == {'/dev/ttyUSBfake'}
|
||||||
|
mock_platform.assert_called()
|
||||||
|
mock_sp.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
@patch('subprocess.getstatusoutput')
|
||||||
|
@patch('platform.system', return_value='Darwin')
|
||||||
|
def test_active_ports_on_supported_devices_mac(mock_platform, mock_sp):
|
||||||
|
"""Test active_ports_on_supported_devices()"""
|
||||||
|
mock_sp.return_value = (None, 'crw-rw-rw- 1 root wheel 0x9000000 Feb 8 22:22 /dev/cu.usbserial-foo')
|
||||||
|
fake_device = SupportedDevice(name='a', for_firmware='heltec-v2.1', baseport_on_linux='cu.usbserial-')
|
||||||
|
fake_supported_devices = [fake_device]
|
||||||
|
assert active_ports_on_supported_devices(fake_supported_devices) == {'/dev/cu.usbserial-foo'}
|
||||||
|
mock_platform.assert_called()
|
||||||
|
mock_sp.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
@patch('meshtastic.supported_device.detect_windows_port', return_value={'COM2'})
|
||||||
|
@patch('platform.system', return_value='Windows')
|
||||||
|
def test_active_ports_on_supported_devices_win(mock_platform, mock_dwp):
|
||||||
|
"""Test active_ports_on_supported_devices()"""
|
||||||
|
fake_device = SupportedDevice(name='a', for_firmware='heltec-v2.1')
|
||||||
|
fake_supported_devices = [fake_device]
|
||||||
|
assert active_ports_on_supported_devices(fake_supported_devices) == {'COM2'}
|
||||||
|
mock_platform.assert_called()
|
||||||
|
mock_dwp.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
@patch('subprocess.getstatusoutput')
|
||||||
|
@patch('platform.system', return_value='Darwin')
|
||||||
|
def test_active_ports_on_supported_devices_mac_no_duplicates_check(mock_platform, mock_sp):
|
||||||
|
"""Test active_ports_on_supported_devices()"""
|
||||||
|
mock_sp.return_value = (None, ('crw-rw-rw- 1 root wheel 0x9000005 Mar 8 10:05 /dev/cu.usbmodem53230051441\n'
|
||||||
|
'crw-rw-rw- 1 root wheel 0x9000003 Mar 8 10:06 /dev/cu.wchusbserial53230051441'))
|
||||||
|
fake_device = SupportedDevice(name='a', for_firmware='tbeam', baseport_on_mac='cu.usbmodem')
|
||||||
|
fake_supported_devices = [fake_device]
|
||||||
|
assert active_ports_on_supported_devices(fake_supported_devices, False) == {'/dev/cu.usbmodem53230051441', '/dev/cu.wchusbserial53230051441'}
|
||||||
|
mock_platform.assert_called()
|
||||||
|
mock_sp.assert_called()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
@patch('subprocess.getstatusoutput')
|
||||||
|
@patch('platform.system', return_value='Darwin')
|
||||||
|
def test_active_ports_on_supported_devices_mac_duplicates_check(mock_platform, mock_sp):
|
||||||
|
"""Test active_ports_on_supported_devices()"""
|
||||||
|
mock_sp.return_value = (None, ('crw-rw-rw- 1 root wheel 0x9000005 Mar 8 10:05 /dev/cu.usbmodem53230051441\n'
|
||||||
|
'crw-rw-rw- 1 root wheel 0x9000003 Mar 8 10:06 /dev/cu.wchusbserial53230051441'))
|
||||||
|
fake_device = SupportedDevice(name='a', for_firmware='tbeam', baseport_on_mac='cu.usbmodem')
|
||||||
|
fake_supported_devices = [fake_device]
|
||||||
|
assert active_ports_on_supported_devices(fake_supported_devices, True) == {'/dev/cu.wchusbserial53230051441'}
|
||||||
|
mock_platform.assert_called()
|
||||||
|
mock_sp.assert_called()
|
||||||
@@ -336,6 +336,8 @@ def test_eliminate_duplicate_port():
|
|||||||
assert eliminate_duplicate_port(['/dev/cu.usbserial-0001', '/dev/cu.SLAB_USBtoUART']) == ['/dev/cu.usbserial-0001']
|
assert eliminate_duplicate_port(['/dev/cu.usbserial-0001', '/dev/cu.SLAB_USBtoUART']) == ['/dev/cu.usbserial-0001']
|
||||||
assert eliminate_duplicate_port(['/dev/cu.usbmodem11301', '/dev/cu.wchusbserial11301']) == ['/dev/cu.wchusbserial11301']
|
assert eliminate_duplicate_port(['/dev/cu.usbmodem11301', '/dev/cu.wchusbserial11301']) == ['/dev/cu.wchusbserial11301']
|
||||||
assert eliminate_duplicate_port(['/dev/cu.wchusbserial11301', '/dev/cu.usbmodem11301']) == ['/dev/cu.wchusbserial11301']
|
assert eliminate_duplicate_port(['/dev/cu.wchusbserial11301', '/dev/cu.usbmodem11301']) == ['/dev/cu.wchusbserial11301']
|
||||||
|
assert eliminate_duplicate_port(['/dev/cu.usbmodem53230051441', '/dev/cu.wchusbserial53230051441']) == ['/dev/cu.wchusbserial53230051441']
|
||||||
|
assert eliminate_duplicate_port(['/dev/cu.wchusbserial53230051441', '/dev/cu.usbmodem53230051441']) == ['/dev/cu.wchusbserial53230051441']
|
||||||
|
|
||||||
@patch('platform.version', return_value='10.0.22000.194')
|
@patch('platform.version', return_value='10.0.22000.194')
|
||||||
@patch('platform.release', return_value='10')
|
@patch('platform.release', return_value='10')
|
||||||
|
|||||||
Reference in New Issue
Block a user