Skip to content

Commit

Permalink
Check for telnet BrokenPipeError, better match on bad telnet cmd
Browse files Browse the repository at this point in the history
coverage multiprocessing concurrency
  • Loading branch information
julianneswinoga committed May 20, 2024
1 parent ee5efb3 commit cba1777
Show file tree
Hide file tree
Showing 3 changed files with 31 additions and 3 deletions.
7 changes: 5 additions & 2 deletions flightgear_python/fg_if.py
Original file line number Diff line number Diff line change
Expand Up @@ -423,7 +423,10 @@ def _telnet_str(in_str: str) -> ByteString:
return f'{in_str}\r\n'.encode()

def _send_cmd_get_resp(self, cmd_str: str, buflen: int = 512) -> str:
self.sock.sendall(self._telnet_str(cmd_str))
try:
self.sock.sendall(self._telnet_str(cmd_str))
except BrokenPipeError as e:
raise FGCommunicationError('Failed to send data. Did you call .connect()?') from e

# FG telnet always ends with a prompt (`cwd`> ), and since we always
# operate relative to the root directory, it should always be the same prompt
Expand All @@ -438,7 +441,7 @@ def _send_cmd_get_resp(self, cmd_str: str, buflen: int = 512) -> str:
resp_bytes = strip_end(resp_bytes, b'\r\n' + ending_bytes) # trim the prompt

resp_str = resp_bytes.decode()
if resp_str.startswith('-ERR'):
if resp_str.startswith('-ERR') or resp_str.startswith('Valid commands are'):
raise FGCommunicationError(f'Bad telnet command "{cmd_str}". Response: "{resp_str}"')
return resp_str

Expand Down
5 changes: 5 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -58,3 +58,8 @@ addopts = '-m "not fg_integration"'
filterwarnings = [
'ignore::DeprecationWarning:setuptools:2',
]

[tool.coverage.run]
branch = true
timid = true
concurrency = ['multiprocessing']
22 changes: 21 additions & 1 deletion tests/test_integration_FG_telnet.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import pytest

from flightgear_python.fg_if import TelnetConnection
from flightgear_python.fg_util import FGConnectionError
from flightgear_python.fg_util import FGConnectionError, FGCommunicationError


pytestmark = pytest.mark.fg_integration
Expand Down Expand Up @@ -49,3 +49,23 @@ def test_telnet_wrong_port():
t_con = TelnetConnection('localhost', 123)
with pytest.raises(FGConnectionError):
t_con.connect()


def test_telnet_no_connect():
t_con = TelnetConnection('localhost', 5500)
with pytest.raises(FGCommunicationError):
t_con.get_prop('/controls/flight/rudder')


def test_telnet_bad_path_cmd():
t_con = TelnetConnection('localhost', 5500)
t_con.connect()
with pytest.raises(FGCommunicationError):
t_con.list_props('/aaa/bbb/ccc')


def test_telnet_bad_internal_cmd():
t_con = TelnetConnection('localhost', 5500)
t_con.connect()
with pytest.raises(FGCommunicationError):
t_con._send_cmd_get_resp('what a bad command')

0 comments on commit cba1777

Please sign in to comment.