From 1b844b947a40c6d92031435193d131db795ba752 Mon Sep 17 00:00:00 2001 From: Stuart Longland Date: Tue, 7 May 2024 17:01:38 +1000 Subject: [PATCH] peer tests: Clean up SABM handling in connection helper --- tests/test_peer/test_peerconnection.py | 282 +------------------------ 1 file changed, 5 insertions(+), 277 deletions(-) diff --git a/tests/test_peer/test_peerconnection.py b/tests/test_peer/test_peerconnection.py index 01a621e..f9a7e09 100644 --- a/tests/test_peer/test_peerconnection.py +++ b/tests/test_peer/test_peerconnection.py @@ -159,31 +159,6 @@ def test_peerconn_on_negotiated_failed(): assert done_evts == [{"response": "whoopsie"}] -def test_peerconn_on_negotiated_sabmframe_handler(): - """ - Test _on_negotiated refuses to run if another SABM frame handler is hooked. - """ - station = DummyStation(AX25Address("VK4MSL", ssid=1)) - peer = DummyPeer(station, AX25Address("VK4MSL")) - helper = AX25PeerConnectionHandler(peer) - - # Nothing should be set up - assert helper._timeout_handle is None - assert not helper._done - assert peer.transmit_calls == [] - - # Hook the SABM handler - peer._sabmframe_handler = lambda *a, **kwa: None - - # Hook the done signal - done_evts = [] - helper.done_sig.connect(lambda **kw: done_evts.append(kw)) - - # Try to connect - helper._on_negotiated("xid") - assert done_evts == [{"response": "station_busy"}] - - def test_peerconn_on_negotiated_uaframe_handler(): """ Test _on_negotiated refuses to run if another UA frame handler is hooked. @@ -287,241 +262,28 @@ def test_peerconn_on_negotiated_xid(): assert callback is None -def test_peerconn_check_connection_init(): - """ - Test _check_connection_init finalises connection if both SABMs ACKed - """ - station = DummyStation(AX25Address("VK4MSL", ssid=1)) - peer = DummyPeer(station, AX25Address("VK4MSL")) - helper = AX25PeerConnectionHandler(peer) - - # Assume we're using modulo-8 mode - peer._modulo128 = False - - # Mark SABMs ACKed - helper._our_sabm_acked = True - helper._their_sabm_acked = True - - # Nothing should be set up - assert helper._timeout_handle is None - assert not helper._done - - # Stub peer _init_connection - count = dict(init=0) - - def _init_connection(extended): - assert extended is False, "Should be in Modulo-8 mode" - count["init"] += 1 - - peer._init_connection = _init_connection - - # Hook the done signal - done_evts = [] - helper.done_sig.connect(lambda **kw: done_evts.append(kw)) - - # Call _check_connection_init - helper._check_connection_init() - - # We should have initialised the connection - assert count == dict(init=1) - - # See that the helper finished - assert helper._done is True - assert done_evts == [{"response": "ack"}] - - -def test_peerconn_check_connection_init_mod128(): - """ - Test _check_connection_init finalises mod128 connections too - """ - station = DummyStation(AX25Address("VK4MSL", ssid=1)) - peer = DummyPeer(station, AX25Address("VK4MSL")) - helper = AX25PeerConnectionHandler(peer) - - # Assume we're using modulo-128 mode - peer._modulo128 = True - - # Mark SABMs ACKed - helper._our_sabm_acked = True - helper._their_sabm_acked = True - - # Nothing should be set up - assert helper._timeout_handle is None - assert not helper._done - - # Stub peer _init_connection - count = dict(init=0) - - def _init_connection(extended): - assert extended is True, "Should be in Modulo-128 mode" - count["init"] += 1 - - peer._init_connection = _init_connection - - # Hook the done signal - done_evts = [] - helper.done_sig.connect(lambda **kw: done_evts.append(kw)) - - # Call _check_connection_init - helper._check_connection_init() - - # We should have initialised the connection - assert count == dict(init=1) - - # See that the helper finished - assert helper._done is True - assert done_evts == [{"response": "ack"}] - - -def test_peerconn_check_connection_init_notoursabm(): - """ - Test _check_connection_init does nothing if our SABM not ACKed - """ - station = DummyStation(AX25Address("VK4MSL", ssid=1)) - peer = DummyPeer(station, AX25Address("VK4MSL")) - helper = AX25PeerConnectionHandler(peer) - - # Assume we're using modulo-8 mode - peer._modulo128 = False - - # Mark their SABM ACKed, but not ours - helper._our_sabm_acked = False - helper._their_sabm_acked = True - - # Nothing should be set up - assert helper._timeout_handle is None - assert not helper._done - - # Stub peer _init_connection - count = dict(init=0) - - def _init_connection(extended): - assert extended is False, "Should be in Modulo-8 mode" - count["init"] += 1 - - peer._init_connection = _init_connection - - # Hook the done signal - done_evts = [] - helper.done_sig.connect(lambda **kw: done_evts.append(kw)) - - # Call _check_connection_init - helper._check_connection_init() - - # We should NOT have initialised the connection - assert count == dict(init=0) - - # See that the helper is NOT finished - assert helper._done is False - assert done_evts == [] - - -def test_peerconn_check_connection_init_nottheirsabm(): +def test_peerconn_receive_ua(): """ - Test _check_connection_init does nothing if their SABM not ACKed + Test _on_receive_ua marks the SABM as ACKed """ station = DummyStation(AX25Address("VK4MSL", ssid=1)) peer = DummyPeer(station, AX25Address("VK4MSL")) helper = AX25PeerConnectionHandler(peer) - # Assume we're using modulo-8 mode - peer._modulo128 = False - - # Mark our SABM ACKed, but not theirs - helper._our_sabm_acked = True - helper._their_sabm_acked = False - - # Nothing should be set up - assert helper._timeout_handle is None - assert not helper._done - - # Stub peer _init_connection - count = dict(init=0) - - def _init_connection(extended): - assert extended is False, "Should be in Modulo-8 mode" - count["init"] += 1 - - peer._init_connection = _init_connection - # Hook the done signal done_evts = [] helper.done_sig.connect(lambda **kw: done_evts.append(kw)) - # Call _check_connection_init - helper._check_connection_init() - - # We should NOT have initialised the connection - assert count == dict(init=0) - - # See that the helper is NOT finished - assert helper._done is False - assert done_evts == [] - - -def test_peerconn_receive_ua(): - """ - Test _on_receive_ua marks the SABM as ACKed - """ - station = DummyStation(AX25Address("VK4MSL", ssid=1)) - peer = DummyPeer(station, AX25Address("VK4MSL")) - helper = AX25PeerConnectionHandler(peer) - # Nothing should be set up assert helper._timeout_handle is None assert not helper._done - # Stub helper _check_connection_init - count = dict(check=0) - - def _check_connection_init(): - count["check"] += 1 - - helper._check_connection_init = _check_connection_init - - assert helper._our_sabm_acked is False - # Call _on_receive_ua helper._on_receive_ua() - # Our SABM should be marked as ACKed - assert helper._our_sabm_acked is True - - # We should have checked the ACK status - assert count == dict(check=1) - - -def test_peerconn_receive_sabm(): - """ - Test _on_receive_sabm ends the helper - """ - station = DummyStation(AX25Address("VK4MSL", ssid=1)) - peer = DummyPeer(station, AX25Address("VK4MSL")) - helper = AX25PeerConnectionHandler(peer) - - # Nothing should be set up - assert helper._timeout_handle is None - assert not helper._done - - # Stub peer _send_ua - count = dict(send_ua=0, check=0) - - def _send_ua(): - count["send_ua"] += 1 - - peer._send_ua = _send_ua - - # Stub helper _check_connection_init - def _check_connection_init(): - count["check"] += 1 - - helper._check_connection_init = _check_connection_init - - # Call _on_receive_sabm - helper._on_receive_sabm() - - # We should have ACKed the SABM and checked the connection status - assert count == dict(send_ua=1, check=1) + # We should be connected + assert helper._done is True + assert done_evts == [{"response": "ack"}] def test_peerconn_receive_frmr(): @@ -607,7 +369,6 @@ def test_peerconn_on_timeout_first(): assert helper._retries == 1 # Helper should have hooked the handler events - assert peer._sabmframe_handler == helper._on_receive_sabm assert peer._uaframe_handler == helper._on_receive_ua assert peer._frmrframe_handler == helper._on_receive_frmr assert peer._dmframe_handler == helper._on_receive_dm @@ -638,7 +399,6 @@ def test_peerconn_on_timeout_last(): helper._retries = 0 # Pretend we're hooked up - peer._sabmframe_handler = helper._on_receive_sabm peer._uaframe_handler = helper._on_receive_ua peer._frmrframe_handler = helper._on_receive_frmr peer._dmframe_handler = helper._on_receive_dm @@ -654,7 +414,6 @@ def test_peerconn_on_timeout_last(): assert helper._timeout_handle is None # Helper should have unhooked the handler events - assert peer._sabmframe_handler is None assert peer._uaframe_handler is None assert peer._frmrframe_handler is None assert peer._dmframe_handler is None @@ -677,7 +436,6 @@ def test_peerconn_finish_disconnect_ua(): # Pretend we're hooked up dummy_uaframe_handler = lambda *a, **kw: None - peer._sabmframe_handler = helper._on_receive_sabm peer._uaframe_handler = dummy_uaframe_handler peer._frmrframe_handler = helper._on_receive_frmr peer._dmframe_handler = helper._on_receive_dm @@ -686,37 +444,11 @@ def test_peerconn_finish_disconnect_ua(): helper._finish() # All except UA (which is not ours) should be disconnected - assert peer._sabmframe_handler is None assert peer._uaframe_handler == dummy_uaframe_handler assert peer._frmrframe_handler is None assert peer._dmframe_handler is None -def test_peerconn_finish_disconnect_sabm(): - """ - Test _finish leaves other SABM hooks intact - """ - station = DummyStation(AX25Address("VK4MSL", ssid=1)) - peer = DummyPeer(station, AX25Address("VK4MSL")) - helper = AX25PeerConnectionHandler(peer) - - # Pretend we're hooked up - dummy_sabmframe_handler = lambda *a, **kw: None - peer._sabmframe_handler = dummy_sabmframe_handler - peer._uaframe_handler = helper._on_receive_ua - peer._frmrframe_handler = helper._on_receive_frmr - peer._dmframe_handler = helper._on_receive_dm - - # Call the finish routine - helper._finish() - - # All except SABM (which is not ours) should be disconnected - assert peer._sabmframe_handler == dummy_sabmframe_handler - assert peer._uaframe_handler is None - assert peer._frmrframe_handler is None - assert peer._dmframe_handler is None - - def test_peerconn_finish_disconnect_frmr(): """ Test _finish leaves other FRMR hooks intact @@ -727,7 +459,6 @@ def test_peerconn_finish_disconnect_frmr(): # Pretend we're hooked up dummy_frmrframe_handler = lambda *a, **kw: None - peer._sabmframe_handler = helper._on_receive_sabm peer._uaframe_handler = helper._on_receive_ua peer._frmrframe_handler = dummy_frmrframe_handler peer._dmframe_handler = helper._on_receive_dm @@ -736,7 +467,6 @@ def test_peerconn_finish_disconnect_frmr(): helper._finish() # All except FRMR (which is not ours) should be disconnected - assert peer._sabmframe_handler is None assert peer._uaframe_handler is None assert peer._frmrframe_handler == dummy_frmrframe_handler assert peer._dmframe_handler is None @@ -752,7 +482,6 @@ def test_peerconn_finish_disconnect_dm(): # Pretend we're hooked up dummy_dmframe_handler = lambda *a, **kw: None - peer._sabmframe_handler = helper._on_receive_sabm peer._uaframe_handler = helper._on_receive_ua peer._frmrframe_handler = helper._on_receive_frmr peer._dmframe_handler = dummy_dmframe_handler @@ -761,7 +490,6 @@ def test_peerconn_finish_disconnect_dm(): helper._finish() # All except DM (which is not ours) should be disconnected - assert peer._sabmframe_handler is None assert peer._uaframe_handler is None assert peer._frmrframe_handler is None assert peer._dmframe_handler == dummy_dmframe_handler