From 08d1e2325eb2fdd063277f9c67d146482e779c67 Mon Sep 17 00:00:00 2001 From: Zach Johnson Date: Sun, 1 Mar 2020 21:40:57 -0800 Subject: [PATCH] Migrate connection response to L2capMatchers Test: cert/run --host --test_filter=L2capTest Change-Id: Ia180ec59f94cd65621c19225600fceef455229bc --- gd/cert/matchers.py | 16 ++++++++++++++++ gd/l2cap/classic/cert/l2cap_test.py | 18 +----------------- 2 files changed, 17 insertions(+), 17 deletions(-) diff --git a/gd/cert/matchers.py b/gd/cert/matchers.py index c8b0511d8..09990ed27 100644 --- a/gd/cert/matchers.py +++ b/gd/cert/matchers.py @@ -17,11 +17,16 @@ import bluetooth_packets_python3 as bt_packets from bluetooth_packets_python3 import l2cap_packets from bluetooth_packets_python3.l2cap_packets import CommandCode +from bluetooth_packets_python3.l2cap_packets import ConnectionResponseResult class L2capMatchers(object): @staticmethod + def ConnectionResponse(scid): + return lambda packet: L2capMatchers._is_matching_connection_response(packet, scid) + + @staticmethod def ConnectionRequest(): return lambda packet: L2capMatchers._is_control_frame_with_code(packet, CommandCode.CONNECTION_REQUEST) @@ -61,3 +66,14 @@ class L2capMatchers(object): @staticmethod def _is_control_frame_with_code(packet, code): return L2capMatchers._control_frame_with_code(packet, code) is not None + + @staticmethod + def _is_matching_connection_response(packet, scid): + frame = L2capMatchers._control_frame_with_code( + packet, CommandCode.CONNECTION_RESPONSE) + if frame is None: + return False + response = l2cap_packets.ConnectionResponseView(frame) + return response.GetSourceCid() == scid and response.GetResult( + ) == ConnectionResponseResult.SUCCESS and response.GetDestinationCid( + ) != 0 diff --git a/gd/l2cap/classic/cert/l2cap_test.py b/gd/l2cap/classic/cert/l2cap_test.py index 27c494b90..ac623e495 100644 --- a/gd/l2cap/classic/cert/l2cap_test.py +++ b/gd/l2cap/classic/cert/l2cap_test.py @@ -364,24 +364,8 @@ class L2capTest(GdFacadeOnlyBaseTestClass): 1, l2cap_packets.ConnectionRequestBuilder(signal_id, psm, scid)) self.cert_send_b_frame(open_channel) - def verify_connection_response(packet): - packet_bytes = packet.payload - l2cap_view = l2cap_packets.BasicFrameView( - bt_packets.PacketViewLittleEndian(list(packet_bytes))) - l2cap_control_view = l2cap_packets.ControlView( - l2cap_view.GetPayload()) - if l2cap_control_view.GetCode( - ) != l2cap_packets.CommandCode.CONNECTION_RESPONSE: - return False - connection_response_view = l2cap_packets.ConnectionResponseView( - l2cap_control_view) - return connection_response_view.GetSourceCid( - ) == scid and connection_response_view.GetResult( - ) == l2cap_packets.ConnectionResponseResult.SUCCESS and connection_response_view.GetDestinationCid( - ) != 0 - assertThat(self.cert_acl_manager.get_acl_stream()).emits( - verify_connection_response) + L2capMatchers.ConnectionResponse(scid)) def test_connect_dynamic_channel_and_send_data(self): self._setup_link_from_cert() -- 2.11.0