cellular-infrastructure
/
osmo-python-tests
mirror of https://gerrit.osmocom.org/python/osmo-python-tests
9
0
Fork 0
Code Issues Projects Releases Wiki Activity

Update ctrl command parsing for python3 

* make parse() return command id in addition to variable name and value
* introduce parse_kv() wrapper which ignores that id and use it instead
  of old parse()
* make parse() compatible with python3 where we got bytes, not string
  from the socket so we have to decode it properly before using split()
* expand test_py3.py with simply asyn server which verifies that
  osmo_ctrl.py works properly

Change-Id: I599f9f5a18109929f59386ab4416b8bfd75c74d1
This commit is contained in:
Max 2017-12-21 14:38:39 +01:00
parent 8a02e36575
commit 566f2a7590
5 changed files with 88 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
contrib/jenkins.sh
View File

@ -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

49
osmopy/osmo_ipa.py
View File

@ -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 (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 i, None, v
if s == self.CTRL_GET:
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 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
"""
(s, i, v) = data.split(' ', 2)
if s == self.CTRL_ERR:
return None, v
if s == self.CTRL_GET:
return 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)
(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

4
scripts/osmo_ctrl.py
View File

@ -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)

4
scripts/twisted_ipa.py
View File

@ -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)

50
tests/test_py3.py
View File

@ -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__)