Skip to content

Commit

Permalink
peer tests: Clean up SABM handling in connection helper
Browse files Browse the repository at this point in the history
  • Loading branch information
@sjlongland
sjlongland committed May 7, 2024
1 parent a93f869 commit 1b844b9
Showing 1 changed file with 5 additions and 277 deletions.
282 changes: 5 additions & 277 deletions tests/test_peer/test_peerconnection.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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():
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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

0 comments on commit 1b844b9

Please sign in to comment.