tests/test_py3.py

#!/usr/bin/env python3

# just a smoke test for osmopy

import asyncio, random
from osmopy.osmo_ipa import Ctrl
from osmopy import __version__

class CtrlProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        (i, v, k) = Ctrl().parse(data)
        if not k:
            print('Ctrl GET received: %s' % v)
        else:
            print('Ctrl SET received: %s :: %s' % (v, k))

        message = Ctrl().reply(i, v, k)
        self.transport.write(message)

        self.transport.close()
        # quit the loop gracefully
        print('Closing the loop...')
        loop.stop()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    test_host = '127.0.0.5'
    test_port = str(random.randint(1025, 60000))

    # Each client connection will create a new protocol instance
    server = loop.run_until_complete(loop.create_server(CtrlProtocol, test_host, test_port))

    print('Serving on {}...'.format(server.sockets[0].getsockname()))

    # Async client running in the subprocess plugged to the same event loop
    loop.run_until_complete(asyncio.gather(asyncio.create_subprocess_exec('./scripts/osmo_ctrl.py', '-g', 'mnc', '-d', test_host, '-p', test_port), loop = loop))

    loop.run_forever()

    # Cleanup after loop is finished
    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

    print('[Python3] Smoke test PASSED for v%s' % __version__)