# See the License for the specific language governing permissions and
# limitations under the License.
+import logging
import bluetooth_packets_python3 as bt_packets
import logging
@staticmethod
def LoopbackOf(packet):
- data = bytes(hci_packets.LoopbackCommandBuilder(packet).Serialize())
+ return HciMatchers.Exactly(hci_packets.LoopbackCommandBuilder(packet))
+
+ @staticmethod
+ def Exactly(packet):
+ data = bytes(packet.Serialize())
return lambda event: data == event.payload
from cert.event_stream import EventStream
from cert.truth import assertThat
from cert.py_hal import PyHal
+from cert.matchers import HciMatchers
from google.protobuf import empty_pb2
from facade import rootservice_pb2 as facade_rootservice_pb2
from hal import facade_pb2 as hal_facade_pb2
self.cert_hal.close()
super().teardown_test()
- def test_fetch_hci_event(self):
+ def test_stream_events(self):
self.dut_hal.send_hci_command(
hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))
- event = hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)
- assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: bytes(event.Serialize()) in packet.payload)
+ assertThat(self.dut_hal.get_hci_event_stream()).emits(
+ HciMatchers.Exactly(hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)))
def test_loopback_hci_command(self):
self.dut_hal.send_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))
'0C:05:04:03:02:01')
self.dut_hal.send_hci_command(command)
- assertThat(
- self.dut_hal.get_hci_event_stream()).emits(lambda packet: bytes(command.Serialize()) in packet.payload)
+ assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LoopbackOf(command))
def test_inquiry_from_dut(self):
self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))