| % Regression tests for TestSocket |
| |
| + Configuration |
| ~ conf |
| |
| = Imports |
| |
| from test.testsocket import TestSocket, cleanup_testsockets |
| |
| = Create Dummy Packet |
| |
| class TestPacket(Packet): |
| fields_desc = [ |
| IntField("identifier", 0), |
| StrField("data", b"") |
| ] |
| def answers(self, other): |
| if other.__class__ != self.__class__: |
| return False |
| if self.identifier % 2: |
| return False |
| if self.identifier == (other.identifier + 1): |
| return True |
| return False |
| def hashret(self): |
| return struct.pack('I', self.identifier + (self.identifier % 2)) |
| |
| |
| = Create Sockets |
| |
| sender = TestSocket(TestPacket) |
| receiver = TestSocket(TestPacket) |
| sender.pair(receiver) |
| |
| + Basic tests |
| |
| = Simple ping pong |
| |
| def create_answer(p): |
| ans = TestPacket(identifier=p.identifier + 1, data=p.data + b"_answer") |
| receiver.send(ans) |
| |
| t = AsyncSniffer(timeout=50, prn=create_answer, opened_socket=receiver) |
| t.start() |
| |
| pks = PacketList() |
| |
| for i in range(1, 2000, 2): |
| txp = TestPacket(identifier=i, data=b"hello"*i) |
| rxp = sender.sr1(txp, verbose=False, timeout=0.5) |
| pks.append(txp) |
| pks.append(rxp) |
| |
| t.stop(join=True) |
| convs = pks.sr() |
| |
| sender.close() |
| receiver.close() |
| |
| assert len(t.results) == 1000 |
| assert len(pks) == 2000 |
| assert len(convs[0]) == 1000 |
| |
| = Simple ping pong with sr with packet generator 500 |
| |
| testlen = 500 |
| |
| sender = TestSocket(TestPacket) |
| receiver = TestSocket(TestPacket) |
| sender.pair(receiver) |
| |
| t = AsyncSniffer(timeout=10, prn=create_answer, opened_socket=receiver) |
| t.start() |
| |
| txp = TestPacket(identifier=range(1, testlen * 2, 2), data=b"test1") |
| rxp = sender.sr(txp, timeout=10, verbose=False, prebuild=True) |
| t.stop(join=True) |
| |
| print(rxp) |
| print(rxp[0].summary()) |
| |
| sender.close() |
| receiver.close() |
| |
| assert len(t.results) == testlen |
| assert len(rxp[0]) == testlen |
| |
| = Simple ping pong with sr with generated packets |
| |
| sender = TestSocket(TestPacket) |
| receiver = TestSocket(TestPacket) |
| sender.pair(receiver) |
| |
| t = AsyncSniffer(timeout=10, prn=create_answer, opened_socket=receiver) |
| t.start() |
| |
| txp = [TestPacket(identifier=i, data=b"hello") for i in range(1, 2000, 2)] |
| rxp = sender.sr(txp, timeout=10, verbose=False) |
| t.stop(join=True) |
| |
| print(rxp) |
| assert len(t.results) == 1000 |
| assert len(rxp[0]) == 1000 |
| |
| + Cleanup |
| |
| = Delete TestSockets |
| |
| cleanup_testsockets() |
