mirror of
https://github.com/meshtastic/python.git
synced 2026-01-19 21:28:07 -05:00
cover a few more lines
This commit is contained in:
@@ -62,8 +62,7 @@ class StreamInterface(MeshInterface):
|
||||
# because we want to ensure it is looking for START1)
|
||||
p = bytearray([START2] * 32)
|
||||
self._writeBytes(p)
|
||||
if not self.noProto:
|
||||
time.sleep(0.1) # wait 100ms to give device time to start running
|
||||
time.sleep(0.1) # wait 100ms to give device time to start running
|
||||
|
||||
self._rxThread.start()
|
||||
|
||||
@@ -90,8 +89,7 @@ class StreamInterface(MeshInterface):
|
||||
self.stream.write(b)
|
||||
self.stream.flush()
|
||||
# we sleep here to give the TBeam a chance to work
|
||||
if not self.noProto:
|
||||
time.sleep(0.1)
|
||||
time.sleep(0.1)
|
||||
|
||||
def _readBytes(self, length):
|
||||
"""Read an array of bytes from our stream"""
|
||||
|
||||
@@ -33,6 +33,12 @@ class TCPInterface(StreamInterface):
|
||||
StreamInterface.__init__(self, debugOut=debugOut, noProto=noProto,
|
||||
connectNow=connectNow)
|
||||
|
||||
def _socket_shutdown(self):
|
||||
"""Shutdown the socket.
|
||||
Note: Broke out this line so the exception could be unit tested.
|
||||
"""
|
||||
self.socket.shutdown(socket.SHUT_RDWR)
|
||||
|
||||
def myConnect(self):
|
||||
"""Connect to socket"""
|
||||
server_address = (self.hostname, self.portNumber)
|
||||
@@ -48,7 +54,7 @@ class TCPInterface(StreamInterface):
|
||||
self._wantExit = True
|
||||
if not self.socket is None:
|
||||
try:
|
||||
self.socket.shutdown(socket.SHUT_RDWR)
|
||||
self._socket_shutdown()
|
||||
except:
|
||||
pass # Ignore errors in shutdown, because we might have a race with the server
|
||||
self.socket.close()
|
||||
|
||||
@@ -27,6 +27,23 @@ def test_TCPInterface(capsys):
|
||||
iface.close()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_TCPInterface_exception():
|
||||
"""Test that we can instantiate a TCPInterface"""
|
||||
|
||||
def throw_an_exception():
|
||||
raise ValueError("Fake exception.")
|
||||
|
||||
with patch('meshtastic.tcp_interface.TCPInterface._socket_shutdown') as mock_shutdown:
|
||||
mock_shutdown.side_effect = throw_an_exception
|
||||
with patch('socket.socket') as mock_socket:
|
||||
iface = TCPInterface(hostname='localhost', noProto=True)
|
||||
iface.myConnect()
|
||||
iface.close()
|
||||
assert mock_socket.called
|
||||
assert mock_shutdown.called
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_TCPInterface_without_connecting():
|
||||
"""Test that we can instantiate a TCPInterface with connectNow as false"""
|
||||
|
||||
Reference in New Issue
Block a user