Max | c35bb47 | 2017-12-15 11:56:05 +0100 | [diff] [blame] | 1 | #!/usr/bin/env python3 |
| 2 | |
Max | 566f2a7 | 2017-12-21 14:38:39 +0100 | [diff] [blame] | 3 | # just a smoke test for osmopy |
Max | c35bb47 | 2017-12-15 11:56:05 +0100 | [diff] [blame] | 4 | |
Max | cd5a6e4 | 2018-01-14 18:27:31 +0100 | [diff] [blame] | 5 | import asyncio, random, sys, os |
| 6 | |
| 7 | # we have to use this ugly hack to workaroundbrokenrelative imports in py3: |
| 8 | # from ..osmopy.osmo_ipa import Ctrl |
| 9 | # does not work as expected |
Max | 901f5eb | 2018-01-15 14:24:15 +0100 | [diff] [blame] | 10 | sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.realpath(__file__)))) |
Max | 566f2a7 | 2017-12-21 14:38:39 +0100 | [diff] [blame] | 11 | from osmopy.osmo_ipa import Ctrl |
| 12 | from osmopy import __version__ |
Max | c35bb47 | 2017-12-15 11:56:05 +0100 | [diff] [blame] | 13 | |
Max | 566f2a7 | 2017-12-21 14:38:39 +0100 | [diff] [blame] | 14 | class CtrlProtocol(asyncio.Protocol): |
| 15 | def connection_made(self, transport): |
| 16 | peername = transport.get_extra_info('peername') |
| 17 | print('Connection from {}'.format(peername)) |
| 18 | self.transport = transport |
| 19 | |
| 20 | def data_received(self, data): |
| 21 | (i, v, k) = Ctrl().parse(data) |
| 22 | if not k: |
| 23 | print('Ctrl GET received: %s' % v) |
| 24 | else: |
| 25 | print('Ctrl SET received: %s :: %s' % (v, k)) |
| 26 | |
| 27 | message = Ctrl().reply(i, v, k) |
| 28 | self.transport.write(message) |
| 29 | |
| 30 | self.transport.close() |
| 31 | # quit the loop gracefully |
| 32 | print('Closing the loop...') |
| 33 | loop.stop() |
| 34 | |
| 35 | |
| 36 | if __name__ == '__main__': |
| 37 | loop = asyncio.get_event_loop() |
| 38 | test_host = '127.0.0.5' |
| 39 | test_port = str(random.randint(1025, 60000)) |
| 40 | |
Max | c41b751 | 2018-01-15 14:08:54 +0100 | [diff] [blame] | 41 | print('Testing v%s on %s:%s' % (__version__, test_host, test_port)) |
| 42 | |
Max | 566f2a7 | 2017-12-21 14:38:39 +0100 | [diff] [blame] | 43 | # Each client connection will create a new protocol instance |
| 44 | server = loop.run_until_complete(loop.create_server(CtrlProtocol, test_host, test_port)) |
| 45 | |
| 46 | print('Serving on {}...'.format(server.sockets[0].getsockname())) |
| 47 | |
| 48 | # Async client running in the subprocess plugged to the same event loop |
| 49 | loop.run_until_complete(asyncio.gather(asyncio.create_subprocess_exec('./scripts/osmo_ctrl.py', '-g', 'mnc', '-d', test_host, '-p', test_port), loop = loop)) |
| 50 | |
| 51 | loop.run_forever() |
| 52 | |
| 53 | # Cleanup after loop is finished |
| 54 | server.close() |
| 55 | loop.run_until_complete(server.wait_closed()) |
| 56 | loop.close() |
| 57 | |
| 58 | print('[Python3] Smoke test PASSED for v%s' % __version__) |