mawei/dragonpilot
1
0
Fork 0
mirror of https://github.com/dragonpilot/dragonpilot.git synced 2026-08-07 09:15:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

athenad: fix test timeouts and comments (#31540)

Browse Source 
* these pass in 0.5s since server sends ping on connect

* comments

* unused

* fix

* fix these too

* check end_event while uploading, throw abort exception if we need to shut down/restart

* Revert "check end_event while uploading, throw abort exception if we need to shut down/restart"

This reverts commit f0b822fca98cd8e2b3d4d6d5ede6b512d3ed3ec0.

* more tol for lte connection
This commit is contained in:
Shane Smiskol
2024-02-22 00:08:10 -06:00
committed by GitHub
parent 9dde727974
commit c0e172e0c0
2 changed files with 8 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
selfdrive/athena/athenad.py
+3
View File
@@ -746,6 +746,9 @@ def ws_manage(ws: WebSocket, end_event: threading.Event) -> None:
onroad_prev = onroad
if sock is not None:
# While not sending data, onroad, we can expect to time out in 7 + (7 * 2) = 21s
# offroad, we can expect to time out in 30 + (10 * 3) = 60s
# FIXME: TCP_USER_TIMEOUT is effectively 2x for some reason (32s), so it's mostly unused
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_USER_TIMEOUT, 16000 if onroad else 0)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 7 if onroad else 30)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 7 if onroad else 10)
selfdrive/athena/tests/test_athenad_ping.py
+5 -3
View File
@@ -12,6 +12,8 @@ from openpilot.selfdrive.athena import athenad
from openpilot.selfdrive.manager.helpers import write_onroad_params
from openpilot.system.hardware import TICI
TIMEOUT_TOLERANCE = 20 # seconds
def wifi_radio(on: bool) -> None:
if not TICI:
@@ -63,7 +65,7 @@ class TestAthenadPing(unittest.TestCase):
mock_create_connection.assert_called_once()
mock_create_connection.reset_mock()
# check normal behaviour
# check normal behaviour, server pings on connection
with self.subTest("Wi-Fi: receives ping"), Timeout(70, "no ping received"):
while not self._received_ping():
time.sleep(0.1)
@@ -92,12 +94,12 @@ class TestAthenadPing(unittest.TestCase):
@unittest.skipIf(not TICI, "only run on desk")
def test_offroad(self) -> None:
write_onroad_params(False, self.params)
self.assertTimeout(100) # expect approx 90s
self.assertTimeout(60 + TIMEOUT_TOLERANCE) # based using TCP keepalive settings
@unittest.skipIf(not TICI, "only run on desk")
def test_onroad(self) -> None:
write_onroad_params(True, self.params)
self.assertTimeout(30) # expect 20-30s
self.assertTimeout(21 + TIMEOUT_TOLERANCE)
if __name__ == "__main__":