systemtest: First run at a simple systemtest
This commit is contained in:
parent
91ef01ca73
commit
d1a00d9797
|
@ -0,0 +1,97 @@
|
|||
#!/usr/bin/env python2
|
||||
|
||||
import subprocess
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
import random
|
||||
import struct
|
||||
import timeout_decorator
|
||||
|
||||
verbose = True
|
||||
|
||||
def prefix_ipa_ctrl_header(data):
|
||||
return struct.pack(">HBB", len(data)+1, 0xee, 0) + data
|
||||
|
||||
def ipa_ctrl_header(header):
|
||||
(plen, ipa_proto, osmo_proto) = struct.unpack(">HBB", header)
|
||||
return None if (ipa_proto != 0xee or osmo_proto != 0) else plen
|
||||
|
||||
def remove_ipa_ctrl_header(data):
|
||||
if (len(data) < 4):
|
||||
raise BaseException("Answer too short!")
|
||||
plen = ipa_ctrl_header(data[:4])
|
||||
if (None == plen):
|
||||
raise BaseException("Wrong protocol in answer!")
|
||||
if (plen + 3 > len(data)):
|
||||
print "Warning: Wrong payload length (expected %i, got %i)" % (plen, len(data) - 3)
|
||||
return data[4:plen+3], data[plen+3:]
|
||||
|
||||
def connect(host, port):
|
||||
if verbose:
|
||||
print "Connecting to host %s:%i" % (host, port)
|
||||
|
||||
sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
sck.setblocking(1)
|
||||
sck.connect((host, port))
|
||||
return sck
|
||||
|
||||
def send(sck, data):
|
||||
if verbose:
|
||||
print "Sending \"%s\"" %(data)
|
||||
data = prefix_ipa_ctrl_header(data)
|
||||
sck.send(data)
|
||||
|
||||
def do_set(var, value, op_id, sck):
|
||||
setmsg = "SET %s %s %s" %(op_id, var, value)
|
||||
send(sck, setmsg)
|
||||
|
||||
def do_get(var, op_id, sck):
|
||||
getmsg = "GET %s %s" %(op_id, var)
|
||||
send(sck, getmsg)
|
||||
|
||||
def do_set_get(sck, var, value = None):
|
||||
r = random.randint(1, sys.maxint)
|
||||
if (value != None):
|
||||
s = 'SET_REPLY'
|
||||
do_set(var, value, r, sck)
|
||||
else:
|
||||
s = 'GET_REPLY'
|
||||
do_get(var, r, sck)
|
||||
(answer, data) = remove_ipa_ctrl_header(sck.recv(4096))
|
||||
x = answer.split()
|
||||
if (s == x[0] and str(r) == x[1] and var == x[2]):
|
||||
return None if ('SET_REPLY' == s and value != x[3]) else x[3]
|
||||
return None
|
||||
|
||||
def set_var(sck, var, val):
|
||||
return do_set_get(sck, var, val)
|
||||
|
||||
@timeout_decorator.timeout(20)
|
||||
def run_ussd_test():
|
||||
gst = subprocess.Popen(["gst", "ussd/USSDTest.st"])
|
||||
res = gst.wait()
|
||||
print(res)
|
||||
if res != 23:
|
||||
raise Exception("Non success error code")
|
||||
|
||||
# Start OpenBSCand give it some time to start
|
||||
obsc = subprocess.Popen(["../openbsc/openbsc/src/osmo-nitb/osmo-nitb", "-c", "../openbsc/openbsc/doc/examples/osmo-nitb/nanobts/openbsc.cfg", "-e", "1"])
|
||||
time.sleep(10)
|
||||
|
||||
# try to create and authorize subscriber
|
||||
sck = connect("localhost", 4249)
|
||||
res = set_var(sck, b"subscriber-modify-v1", b"901010000001111,445567")
|
||||
sck.close()
|
||||
|
||||
|
||||
# Okay now.. execute the tests..
|
||||
run_ussd_test()
|
||||
|
||||
|
||||
|
||||
obsc.terminate()
|
||||
obsc.wait()
|
||||
res = proc.wait()
|
||||
print(res)
|
||||
sys.exit(1)
|
|
@ -0,0 +1,11 @@
|
|||
#!/bin/bash
|
||||
|
||||
set -ex
|
||||
|
||||
cd openbsc
|
||||
#./contrib/jenkins.sh
|
||||
export LD_LIBRARY_PATH=$PWD/deps/install/lib/
|
||||
|
||||
cd ../systemtest
|
||||
rm -f hlr.sqlite3
|
||||
./systemtest.py
|
Reference in New Issue