Testing and debugging

Here are some general rules and ideas for developing and debugging distributed systems with aiomas:

  • Distributed systems are complex. Always start as simple as possible. Examine and understand the behavior of that system. Start adding a bit more complexity. Repeat.
  • I find using a debugger does not work very well with async., distributed systems, so I tend to add a lot of logging and or print()s to my code for debugging purposes.
  • Read Develop with asyncio.
  • If you enable asyncio’s debug mode, aiomas also falls into debug mode. It gives you better / more detailed exceptions in some cases. This impacts performance, so it isn’t activated always.
  • Write unit and integration tests and run them as often as possible. Also check that your tests will fail if they should.

Testing coroutines and agents with pytest

My preferred testing tool is pytest. The plug-in pytest-asyncio makes testing asyncio based programs a lot easier.

As an introduction, I also suggest reading my articles on testing with asyncio. They are especially helpful if you are using the channel and RPC layers. Testing agent systems is a bit “easier” (in the sense that the tests are easier to setup). You can, of course, also look at aiomas’ test suite itself.

Here is a small example that demonstrate how you could test an agent. In this case, the agent class itself and the tests for it are in the same module. In real life, you would have the agent and its test in separate packages (e.g., exampleagent.py and test_exampleagent.py).

import pytest
import aiomas

# Production code (exampleagent.py)

class ExampleAgent(aiomas.Agent):
    async def run(self, target_addr, num):
        remote_agent = await self.container.connect(target_addr)
        return (await remote_agent.service(num))

    async def service(self, val):
        await self.container.clock.sleep(0.001)
        return val

# Testing code (test_exampleagent.py)

def container(event_loop, unused_tcp_port):
    """This fixture creates a new Container instance for every test and binds
    it to a random port.

    It requires the *event_loop" fixture, so every test will also have a fresh
    event loop.

    # Create container and bind its server socket to a random port:
    c = aiomas.Container.create(('', unused_tcp_port))

    # Yield the container to the test case:
    yield c

    # Clean-up that is run after the test finished:

# The "@pytest.mark.asyncio" decorator allows you do use "await"/"yield from"
# directly within your test case.
# The "container" argument tells pytest to pass the return/yield value of the
# corresponding fixture to your test.
async def test_example_agent(container):
    num = 42
    # Start two agents:
    agents = [ExampleAgent(container) for _ in range(2)]
    # Run the 1st one and let it connect to the 2nd one.  Check the return
    # value of the 1st one's run() task:
    res = await agents[0].run(agents[1].addr, num)
    assert res == num