forked from arti/vt1200
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""Tests for the announcement replayer."""
|
|
import os
|
|
import struct
|
|
import sys
|
|
import unittest
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
import announce_replay as ar # noqa: E402
|
|
import extract_stream as ex # noqa: E402
|
|
|
|
PCAP = os.path.join(HERE, "..", "local", "filter-test2.pcap")
|
|
|
|
|
|
def ann_frame(dst=ar.LAT_MULTICAST, mtype=ar.MSG_TYPE_SERVICE, payload=b"hello"):
|
|
lat = bytearray()
|
|
lat.append((mtype << 2) & 0xFC)
|
|
lat.append(1)
|
|
lat += struct.pack("<H", 0) + struct.pack("<H", 0)
|
|
lat += bytes([0, 0])
|
|
lat += bytes([0, 0, len(payload), 0x0E]) + payload
|
|
f = bytearray(dst) + ex.mac_bytes(ex.HOST_MAC)
|
|
f += struct.pack(">H", 0x6004) + lat
|
|
if len(f) < 60:
|
|
f += b"\x00" * (60 - len(f))
|
|
return bytes(f)
|
|
|
|
|
|
class LoadTest(unittest.TestCase):
|
|
def test_real_capture_has_announcements(self):
|
|
if not os.path.exists(PCAP):
|
|
self.skipTest("no capture")
|
|
frames = ar.load_announcements(PCAP)
|
|
self.assertGreater(len(frames), 50)
|
|
# all must be the LAT multicast address and type 0xA
|
|
for f in frames:
|
|
self.assertEqual(f[0:6], ar.LAT_MULTICAST)
|
|
self.assertEqual((f[14] >> 2) & 0x3F, ar.MSG_TYPE_SERVICE)
|
|
|
|
|
|
class ReplayTest(unittest.TestCase):
|
|
def test_sends_in_order_and_honours_loops(self):
|
|
got = []
|
|
n = ar.replay([b"A", b"B"], got.append, interval=0, loops=3)
|
|
self.assertEqual(n, 6)
|
|
self.assertEqual(got, [b"A", b"B"] * 3)
|
|
|
|
def test_interval_sleeps_once_per_frame(self):
|
|
slept = []
|
|
ar.replay([b"A", b"B"], lambda f: None, interval=2.5, loops=2,
|
|
sleeper=slept.append)
|
|
self.assertEqual(slept, [2.5] * 4)
|
|
|
|
def test_empty_frames_terminates(self):
|
|
self.assertEqual(ar.replay([], lambda f: None, interval=0, loops=5), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|