Skip to content

PlaygroundNetworkTestQuickstart

Jump to bottom
sethnielson edited this page Apr 1, 2019 · 2 revisions

It's not ideal to test your network protocols "live". It's hard to get the network to behave the way you want it to, especially when doing error testing.

Unittest

You should learn how to use Python's unit test framework. If you looked at the early escape room autograder, it was written using unittest. You can read more about unittest here.

But a quick-start simple example is shown below:

import unittest
import my_adder

class my_adder_tester(unittest.TestCase):
    def setUp(self):
        self.values_to_add = [2,2]
        self.expected_sum = 4

    def tearDown(self):
        # pass (nothing to do here, could leave out)

    def test_two_plus_two(self):
        sum = 0
        for value in self.values_to_add:
             sum += value
        self.assertEqual(sum, self.expected_sum)

if __name__=="__main__":
    unittest.main()

The assertEqual will cause the test case to fail if sum and self.expected_sum are not the same value. There is also an assertTrue that can check arbitrary conditions as well as other test methods. For the full list, refer to the documentation.

Note that the setUp and tearDown methods are called before and after every test method (every method in the class that begins with test).

Transport Mock

We can use unittest to test a protocol. But we need a "fake" transport that will help us out. The playground framework includes a mock transport that sends all written data to a "sink" variable. The sink attribute can be any object that also has a write method, so it could be an io.BytesIO object or even another transport.

In the example below, the fake transport will produce a series of packets that can be popped off and read.

import my_protocol
import unittest
from playground.network.testing.mock import MockTransportToStorageStream as MockTransport

class PacketStream:
    def __init__(self):
        self.packets = []
    
    def write(self, data):
        deserializer = PIMPPacket.Deserializer()
        deserializer.update(data)
        self.packets += list(deserializer.nextPackets())

class my_protocol_tester(unittest.TestCase):
    def setUp(self):
        self.protocol = my_protocol()
        self.transport = MockTransport(PacketStream())
        # all data written to self.transport will be written
        # to the PacketStream. Can be accessed as 
        # self.transport.sink

    def test_protocol_start(self):
        self.protocol.connection_made(self.transport)

        # Assuming this protocol sends data on connection made
        self.assertEqual(len(self.transport.sink.packets),1)
        sent_packet = self.transport.sink.packets.pop(0)
        self.assertTrue(some_test_on(sent_packet))

    def test_some_other_test(self):
        # some other test here...

Ok, please pay close attention to what is happening here.

  1. In the setup, we create a protocol and a transport. This will happen before every test, so we can start each test assuming that we have a newly created transport and a newly created protocol
  2. We created the protocol directly. We are not using factories, connections, or even playground. And there's no asyncio.
  3. We call the connection_made method on our protocol manually and pass in our mock transport
  4. The test above assumes that the protocol starts writing as soon as connection is made and writes only one packet. Obviously, adjust this to your protocol's spec

But how do you send data to the protocol? Well, you have to custom build packets within the test and send them to the protocol's data_received method.

    def test_some_state(self):
        self.protocol.connection_made(self.transport)
        # assume we need to send some packet
        packet = MyPacket()
        packet.value1 = 3.14159
        packet.value2 = "hello world"
        self.protocol.data_received(packet.__serialize__())

Using this mock transport and hard-coded packets, you can test a wide variety of network communication without a network. It's very effective and very powerful.

Higher Protocol Mock

The unittest framework already has a great mock system for generic objects, and you can use this for general objects that get called. The MagicMock for example will automatically accept any function call and store whatever data is passed to it.

So, if you're creating a StackingProtocol you want a higher protocol to receive data. The MagicMock is an out-of-the box miracle.

from unittest.mock import MagicMock

class some_test_case(unittest.TestCase):
    def setUp(self):
        self.protocol = my_protocol()
        self.transport = MockTransport()
        self.protocol.setHigherProtocol(MagicMock())

    def test_higher_protocol(self):
        self.protocol.connection_made(self.transport)
        
        # assume that the protocol call's higher protocol's
        # connection_made after receiving packet
        first_packet = MyPacket()
        
        self.protocol.data_received(first_packet.__serialize__())
        
        # at this point, higher protocol's connection_made
        # should have been called. Here's how we test
        self.protocol.higherProtocol().connection_made.assert_called()

        # you can also see what it was called with. connection_made
        # is called with a transport. If you want to see what was
        # passed up do this:
        args = self.protocol.higherProtocol().connection_made.call_args[0]
        app_transport = args[0]

For more details, refer to the MagicMock documentation.

Loop Mock

But what if you have a timer? What if your protocol sends something after a certain amount of time? The playground framework also has a simple loop substitute that can help. But you need to be using asyncio.get_event_loop().call_later or similar method for this to help. If you're using threads or some other kind of mechanism, this won't work.

The mock is called TestLoopEx. It has a number of features, but for today, we will only look at its ability to move the internal clock. The basic idea is to create this test loop, set it as asyncio's main loop, and then use it to advance time.

from playground.asyncio_lib.testing import TestLoopEx
# not all imports shown

class some_test_case(unittest.TestCase):
    def setUp(self):
        self.protocol = my_protocol()
        self.transport = MockTransport(PacketStream())
        self.loop = TestLoopEx()
        asyncio.set_event_loop(self.loop)

    def test_delayed_send(self):
        # suppose the protocol sends a message 30 seconds
        # after connection made
        self.protocol.connection_made(self.transport)

        self.loop.advanceClock(31)

        # all call_later functions scheduled within 30
        # seconds of connection_made will fire
        # assume it sent a packet now
        self.assertEqual(len(self.transport.sink.packets), 1)

Example

Hopefully this will get you started. You can look at the reliable unittest that is included as an example.

Clone this wiki locally