diff options
| -rwxr-xr-x | contrib/jenkins.sh | 4 | ||||
| -rwxr-xr-x | osmopy/osmo_ipa.py | 41 | ||||
| -rwxr-xr-x | scripts/osmo_ctrl.py | 4 | ||||
| -rwxr-xr-x | scripts/twisted_ipa.py | 4 | ||||
| -rw-r--r-- | tests/test_py3.py | 50 |
5 files changed, 84 insertions, 19 deletions
diff --git a/contrib/jenkins.sh b/contrib/jenkins.sh index d18b19d..9734549 100755 --- a/contrib/jenkins.sh +++ b/contrib/jenkins.sh @@ -24,7 +24,7 @@ do $PY3 $COM_FLAGS $f done -cd scripts -./osmo_ctrl.py --help +# Run async server which tests scripts/osmo_ctrl.py interaction +$PY3 tests/test_py3.py # TODO: add more tests diff --git a/osmopy/osmo_ipa.py b/osmopy/osmo_ipa.py index 71cbf45..a1fcaf6 100755 --- a/osmopy/osmo_ipa.py +++ b/osmopy/osmo_ipa.py @@ -28,7 +28,7 @@ class IPA(object): """ Stateless IPA protocol multiplexer: add/remove/parse (extended) header """ - version = "0.0.5" + version = "0.0.6" TCP_PORT_OML = 3002 TCP_PORT_RSL = 3003 # OpenBSC extensions: OSMO, MGCP_OLD @@ -231,23 +231,36 @@ class Ctrl(IPA): return None return d - def parse(self, data, op=None): + def parse(self, raw_data): """ - Parse Ctrl string returning (var, value) pair + Parse Ctrl string returning (id, var, value) tuple var could be None in case of ERROR message value could be None in case of GET message + both could be None in case of TRAP with non-zero id """ + data = self.rem_header(raw_data) + if data == None: + return None, None, None + data = data.decode('utf-8') (s, i, v) = data.split(' ', 2) if s == self.CTRL_ERR: - return None, v + return i, None, v if s == self.CTRL_GET: - return v, None + return i, v, None + if s == self.CTRL_GET + '_' + self.CTRL_REP: + return i, v, None (s, i, var, val) = data.split(' ', 3) if s == self.CTRL_TRAP and i != '0': - return None, '%s with non-zero id %s' % (s, i) - if op is not None and i != op: - if s == self.CTRL_GET + '_' + self.CTRL_REP or s == self.CTRL_SET + '_' + self.CTRL_REP: - return None, '%s with unexpected id %s' % (s, i) + return i, None, None + return i, var, val + + def parse_kv(self, raw_data): + """ + Parse Ctrl string returning (var, value) pair + var could be None in case of ERROR message + value could be None in case of GET message + """ + (i, var, val) = self.parse(raw_data) return var, val def trap(self, var, val): @@ -265,11 +278,19 @@ class Ctrl(IPA): return r, self.add_header("%s %s %s %s" % (self.CTRL_SET, r, var, val)) return r, self.add_header("%s %s %s" % (self.CTRL_GET, r, var)) + def reply(self, op_id, var, val=None): + """ + Make SET/GET command reply: returns assembled message + """ + if val is not None: + return self.add_header("%s_%s %s %s %s" % (self.CTRL_SET, self.CTRL_REP, op_id, var, val)) + return self.add_header("%s_%s %s %s" % (self.CTRL_GET, self.CTRL_REP, op_id, var)) + def verify(self, reply, r, var, val=None): """ Verify reply to SET/GET command: returns (b, v) tuple where v is True/False verification result and v is the variable value """ - (k, v) = self.parse(reply) + (k, v) = self.parse_kv(reply) if k != var or (val is not None and v != val): return False, v return True, v diff --git a/scripts/osmo_ctrl.py b/scripts/osmo_ctrl.py index 8c0608f..ac20050 100755 --- a/scripts/osmo_ctrl.py +++ b/scripts/osmo_ctrl.py @@ -40,8 +40,8 @@ def connect(host, port): def do_set_get(sck, var, value = None): (r, c) = Ctrl().cmd(var, value) sck.send(c) - answer = Ctrl().rem_header(sck.recv(4096)) - return (answer,) + Ctrl().verify(answer, r, var, value) + ret = sck.recv(4096) + return (Ctrl().rem_header(ret),) + Ctrl().verify(ret, r, var, value) def set_var(sck, var, val): (a, _, _) = do_set_get(sck, var, val) diff --git a/scripts/twisted_ipa.py b/scripts/twisted_ipa.py index bb8323d..533bfae 100755 --- a/scripts/twisted_ipa.py +++ b/scripts/twisted_ipa.py @@ -22,7 +22,7 @@ */ """ -__version__ = "0.7.0" # bump this on every non-trivial change +__version__ = "0.7.1" # bump this on every non-trivial change from osmopy.osmo_ipa import Ctrl, IPA from twisted.internet.protocol import ReconnectingClientFactory @@ -243,7 +243,7 @@ class CTRL(IPACommon): OSMO CTRL message dispatcher, lambda default should never happen For basic tests only, appropriate handling routines should be replaced: see CtrlServer for example """ - self.dbg('OSMO CTRL received %s::%s' % Ctrl().parse(data.decode('utf-8'))) + self.dbg('OSMO CTRL received %s::%s' % Ctrl().parse_kv(data)) (cmd, op_id, v) = data.decode('utf-8').split(' ', 2) method = getattr(self, 'ctrl_' + cmd, lambda: "CTRL unknown command") method(data, op_id, v) diff --git a/tests/test_py3.py b/tests/test_py3.py index cac2f93..3a96d9f 100644 --- a/tests/test_py3.py +++ b/tests/test_py3.py @@ -1,7 +1,51 @@ #!/usr/bin/env python3 -# just import a smoke test for osmopy +# just a smoke test for osmopy -import osmopy +import asyncio, random +from osmopy.osmo_ipa import Ctrl +from osmopy import __version__ -print('[Python3] Smoke test PASSED.') +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__) |