sipwise
/
lua-ngcp-kamailio
mirror of https://github.com/sipwise/lua-ngcp-kamailio.git
1
0
Fork 0
Code Issues Releases Wiki Activity

TT#146450 scripts: fix ngcp-dlgcnt-clean use for get configs

Browse Source 
* callid parameter was missing
* add some tests for those cases

Change-Id: I4456290de22ccf3d3955470ac1cc05ac4fd62c4b
mr10.4
Victor Seva 3 years ago
parent 6f08b673d0
commit 496b06658e
3 changed files with 149 additions and 89 deletions
Show Stats Download Patch File Download Diff File

4
scripts/ngcp-dlgcnt-check
Unescape View File

@ -31,8 +31,8 @@ if ! ngcp-check-active -q ; then
fi fi
# read kamailio.proxy.dlgcnt.pair_redis_db # read kamailio.proxy.dlgcnt.pair_redis_db
REDIS_DB="$(ngcp-dlgcnt-clean -c 2>/dev/null|| echo 4)" REDIS_DB="$(ngcp-dlgcnt-clean -c fake 2>/dev/null|| echo 4)"
host="$(ngcp-dlgcnt-clean -C 2>/dev/null|| echo localhost)" host="$(ngcp-dlgcnt-clean -C fake 2>/dev/null|| echo localhost)"
# redis full list # redis full list
REDIS_CALLIDS=$(mktemp) REDIS_CALLIDS=$(mktemp)

18
scripts/ngcp-dlgcnt-clean
Unescape View File

@ -1,6 +1,5 @@
#!/usr/bin/env lua5.1 #!/usr/bin/env lua5.1
--local ut = require 'ngcp.utils'.table --local ut = require 'ngcp.utils'.table
local NGCPDlg = require 'ngcp.dlgcnt'
local config = "/etc/kamailio/proxy/dlgcnt.lua.cfg" local config = "/etc/kamailio/proxy/dlgcnt.lua.cfg"
local argparse = require "argparse" local argparse = require "argparse"
@ -53,16 +52,25 @@ local args = parser:parse()
get_config() get_config()
local dlg = NGCPDlg:new()
if args.config_db then if args.config_db then
print(tostring(dlg_config.pair.db)) local val = 4
if dlg_config then
val = dlg_config.pair.db
end
print(tostring(val))
os.exit(0) os.exit(0)
elseif args.config_host then elseif args.config_host then
print(tostring(dlg_config.pair.host)) local val = "127.0.0.1"
if dlg_config then
val = dlg_config.pair.host
end
print(tostring(val))
os.exit(0) os.exit(0)
end end
local NGCPDlg = require 'ngcp.dlgcnt'
local dlg = NGCPDlg:new()
if dlg_config then if dlg_config then
dlg.config.central.host = dlg_config.central.host dlg.config.central.host = dlg_config.central.host
dlg.config.central.port = dlg_config.central.port dlg.config.central.port = dlg_config.central.port

212
tests/test_dlgcnt.py
Unescape View File

@ -8,17 +8,21 @@ import io
import shutil import shutil
import copy import copy
WORKSPACE = os.path.abspath(os.getenv('WORKSPACE', '/tmp')) WORKSPACE = os.path.abspath(os.getenv("WORKSPACE", "/tmp"))
FIXTURES = os.path.join(os.getcwd(), "tests/fixtures") FIXTURES = os.path.join(os.getcwd(), "tests/fixtures")
FAKE_CMD_LIST = os.path.join(WORKSPACE, "cmd_list") FAKE_CMD_LIST = os.path.join(WORKSPACE, "cmd_list")
FAKE_BIN = os.path.join(WORKSPACE, "bin") FAKE_BIN = os.path.join(WORKSPACE, "bin")
FAKE_PATH = "%s:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin" % FAKE_BIN FAKE_PATH = "%s:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin" % FAKE_BIN
def executeAndReturnOutput(command): def executeAndReturnOutput(command, env={}):
p = subprocess.Popen(command, encoding="utf-8", p = subprocess.Popen(
command,
encoding="utf-8",
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE) stderr=subprocess.PIPE,
env=env,
)
stdoutdata, stderrdata = p.communicate() stdoutdata, stderrdata = p.communicate()
# print(stdoutdata, file=sys.stdout) # print(stdoutdata, file=sys.stdout)
print(stderrdata, file=sys.stderr) print(stderrdata, file=sys.stderr)
@ -31,24 +35,23 @@ def create_prog(filename, command):
:param unicode filename: destination filename :param unicode filename: destination filename
:param unicode command: command to write to test program :param unicode command: command to write to test program
""" """
with io.open(filename, 'w', encoding='utf-8') as fp: with io.open(filename, "w", encoding="utf-8") as fp:
fp.write("#!/bin/bash\n%s\n" % (command,)) fp.write("#!/bin/bash\n%s\n" % (command,))
os.fchmod(fp.fileno(), 0o755) os.fchmod(fp.fileno(), 0o755)
def setnode(active=True): def setnode(active=True):
if active: if active:
mode = 'true' mode = "true"
else: else:
mode = 'false' mode = "false"
create_prog(os.path.join(FAKE_BIN, 'ngcp-check-active'), mode) create_prog(os.path.join(FAKE_BIN, "ngcp-check-active"), mode)
command = [
"env",
"PATH=%s" % FAKE_PATH,
"./scripts/ngcp-dlgcnt-check", "-r"]
FAKE_DLG_CLEAN = """ command = ["env", "PATH=%s" % FAKE_PATH, "./scripts/ngcp-dlgcnt-check", "-r"]
FAKE_DLG_CLEAN = (
"""
if [ $1 = -c ] ; then if [ $1 = -c ] ; then
echo 4 echo 4
exit 0 exit 0
@ -57,7 +60,9 @@ if [ $1 = -C ] ; then
echo localhost echo localhost
exit 0 exit 0
fi fi
echo "ngcp-dlgcnt-clean $*">> %s""" % FAKE_CMD_LIST echo "ngcp-dlgcnt-clean $*">> %s"""
% FAKE_CMD_LIST
)
FAKE_REDIS_HELPER = """ FAKE_REDIS_HELPER = """
if [ $1 != -h ] && [ $2 != localhost ] ; then if [ $1 != -h ] && [ $2 != localhost ] ; then
@ -70,28 +75,28 @@ if [ $3 != -n ] && [ $4 != 4 ] ; then
fi""" fi"""
class TestDlgCnt(unittest.TestCase): class Test(unittest.TestCase):
def checkNotCmd(self): def checkNotCmd(self):
self.assertFalse(os.path.exists(FAKE_CMD_LIST), self.assertFalse(os.path.exists(FAKE_CMD_LIST), "%s found" % FAKE_CMD_LIST)
"%s found" % FAKE_CMD_LIST)
def checkCmd(self, f, f2): def checkCmd(self, f, f2):
self.assertTrue(os.path.exists(f), "%s not found" % f) self.assertTrue(os.path.exists(f), "%s not found" % f)
self.assertTrue(os.path.exists(f2), "%s not found" % f2) self.assertTrue(os.path.exists(f2), "%s not found" % f2)
res = executeAndReturnOutput( res = executeAndReturnOutput(["diff", "-uNb", f, f2])
['diff', '-uNb', f, f2])
self.assertEqual(res[0], 0, res[1]) self.assertEqual(res[0], 0, res[1])
class TestDlgCnt(Test):
def setUp(self): def setUp(self):
self.command = copy.deepcopy(command) self.command = copy.deepcopy(command)
if not os.path.exists(FAKE_BIN): if not os.path.exists(FAKE_BIN):
os.makedirs(FAKE_BIN) os.makedirs(FAKE_BIN)
setnode(True) setnode(True)
create_prog(os.path.join(FAKE_BIN, 'ngcp-dlgcnt-clean'), create_prog(os.path.join(FAKE_BIN, "ngcp-dlgcnt-clean"), FAKE_DLG_CLEAN)
FAKE_DLG_CLEAN) create_prog(
create_prog(os.path.join(FAKE_BIN, 'ngcp-dlglist-clean'), os.path.join(FAKE_BIN, "ngcp-dlglist-clean"),
'echo "ngcp-dlglist-clean $*">> %s' % FAKE_CMD_LIST) 'echo "ngcp-dlglist-clean $*">> %s' % FAKE_CMD_LIST,
)
def tearDown(self): def tearDown(self):
shutil.rmtree(WORKSPACE) shutil.rmtree(WORKSPACE)
@ -100,14 +105,14 @@ class TestDlgCnt(unittest.TestCase):
self.command = ["./scripts/ngcp-dlgcnt-check", "-k"] self.command = ["./scripts/ngcp-dlgcnt-check", "-k"]
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 1, res[2]) self.assertEqual(res[0], 1, res[2])
self.assertRegex(res[2], 'illegal option') self.assertRegex(res[2], "illegal option")
self.checkNotCmd() self.checkNotCmd()
def test_help(self): def test_help(self):
self.command = ["./scripts/ngcp-dlgcnt-check", "-h"] self.command = ["./scripts/ngcp-dlgcnt-check", "-h"]
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[2]) self.assertEqual(res[0], 0, res[2])
self.assertRegex(res[1], '\toptions\n') self.assertRegex(res[1], "\toptions\n")
self.checkNotCmd() self.checkNotCmd()
def test_inactive(self): def test_inactive(self):
@ -117,96 +122,143 @@ class TestDlgCnt(unittest.TestCase):
self.checkNotCmd() self.checkNotCmd()
def test_noredisconf(self): def test_noredisconf(self):
create_prog(os.path.join(FAKE_BIN, 'ngcp-dlgcnt-clean'), create_prog(
'echo "error" >&2; false') os.path.join(FAKE_BIN, "ngcp-dlgcnt-clean"), 'echo "error" >&2; false'
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), )
"true") create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "true")
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'), create_prog(
"%s; true" % FAKE_REDIS_HELPER) os.path.join(FAKE_BIN, "ngcp-redis-helper"), "%s; true" % FAKE_REDIS_HELPER
)
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[1]) self.assertEqual(res[0], 0, res[1])
self.checkNotCmd() self.checkNotCmd()
def test_redisconf(self): def test_redisconf(self):
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "true")
"true") create_prog(
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'), os.path.join(FAKE_BIN, "ngcp-redis-helper"), "%s; true" % FAKE_REDIS_HELPER
"%s; true" % FAKE_REDIS_HELPER) )
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[1]) self.assertEqual(res[0], 0, res[1])
self.checkNotCmd() self.checkNotCmd()
def test_empty(self): def test_empty(self):
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "true")
"true") create_prog(os.path.join(FAKE_BIN, "ngcp-redis-helper"), "true")
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'),
"true")
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[1]) self.assertEqual(res[0], 0, res[1])
self.checkNotCmd() self.checkNotCmd()
def test_empty_line(self): def test_empty_line(self):
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "echo")
"echo") create_prog(os.path.join(FAKE_BIN, "ngcp-redis-helper"), "echo")
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'),
"echo")
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[1]) self.assertEqual(res[0], 0, res[1])
self.checkNotCmd() self.checkNotCmd()
def test_okredis(self): def test_okredis(self):
FAKE_DLG = os.path.join(FIXTURES, 'okredis.dlg') FAKE_DLG = os.path.join(FIXTURES, "okredis.dlg")
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "cat %s" % (FAKE_DLG))
"cat %s" % (FAKE_DLG)) FAKE_REDIS = os.path.join(FIXTURES, "okredis.redis")
FAKE_REDIS = os.path.join(FIXTURES, 'okredis.redis') create_prog(
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'), os.path.join(FAKE_BIN, "ngcp-redis-helper"), "cat %s" % (FAKE_REDIS)
"cat %s" % (FAKE_REDIS)) )
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[2]) self.assertEqual(res[0], 0, res[2])
self.checkNotCmd() self.checkNotCmd()
def test_koredis(self): def test_koredis(self):
FAKE_DLG = os.path.join(FIXTURES, 'koredis.dlg') FAKE_DLG = os.path.join(FIXTURES, "koredis.dlg")
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "cat %s" % (FAKE_DLG))
"cat %s" % (FAKE_DLG)) FAKE_REDIS = os.path.join(FIXTURES, "koredis.redis")
FAKE_REDIS = os.path.join(FIXTURES, 'koredis.redis') create_prog(
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'), os.path.join(FAKE_BIN, "ngcp-redis-helper"), "cat %s" % (FAKE_REDIS)
"cat %s" % (FAKE_REDIS)) )
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[2]) self.assertEqual(res[0], 0, res[2])
self.checkCmd(os.path.join(FIXTURES, 'koredis.cmd'), self.checkCmd(os.path.join(FIXTURES, "koredis.cmd"), FAKE_CMD_LIST)
FAKE_CMD_LIST)
def test_kodlgclean(self): def test_kodlgclean(self):
FAKE_DLG = os.path.join(FIXTURES, 'koredis.dlg') FAKE_DLG = os.path.join(FIXTURES, "koredis.dlg")
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "cat %s" % (FAKE_DLG))
"cat %s" % (FAKE_DLG)) FAKE_REDIS = os.path.join(FIXTURES, "koredis.redis")
FAKE_REDIS = os.path.join(FIXTURES, 'koredis.redis') create_prog(
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'), os.path.join(FAKE_BIN, "ngcp-redis-helper"), "cat %s" % (FAKE_REDIS)
"cat %s" % (FAKE_REDIS)) )
create_prog(os.path.join(FAKE_BIN, 'ngcp-dlgcnt-clean'), create_prog(
'%s; false' % FAKE_DLG_CLEAN) os.path.join(FAKE_BIN, "ngcp-dlgcnt-clean"), "%s; false" % FAKE_DLG_CLEAN
)
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[2]) self.assertEqual(res[0], 0, res[2])
self.checkCmd(os.path.join(FIXTURES, 'koredis.cmd'), self.checkCmd(os.path.join(FIXTURES, "koredis.cmd"), FAKE_CMD_LIST)
FAKE_CMD_LIST)
def test_kolistclean(self): def test_kolistclean(self):
FAKE_DLG = os.path.join(FIXTURES, 'koredis.dlg') FAKE_DLG = os.path.join(FIXTURES, "koredis.dlg")
create_prog(os.path.join(FAKE_BIN, 'ngcp-kamctl'), create_prog(os.path.join(FAKE_BIN, "ngcp-kamctl"), "cat %s" % (FAKE_DLG))
"cat %s" % (FAKE_DLG)) FAKE_REDIS = os.path.join(FIXTURES, "koredis.redis")
FAKE_REDIS = os.path.join(FIXTURES, 'koredis.redis') create_prog(
create_prog(os.path.join(FAKE_BIN, 'ngcp-redis-helper'), os.path.join(FAKE_BIN, "ngcp-redis-helper"), "cat %s" % (FAKE_REDIS)
"cat %s" % (FAKE_REDIS)) )
create_prog(os.path.join(FAKE_BIN, 'ngcp-dlglist-clean'), create_prog(
'echo "ngcp-dlglist-clean $*">> %s; false' % FAKE_CMD_LIST) os.path.join(FAKE_BIN, "ngcp-dlglist-clean"),
'echo "ngcp-dlglist-clean $*">> %s; false' % FAKE_CMD_LIST,
)
res = executeAndReturnOutput(self.command) res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 0, res[2]) self.assertEqual(res[0], 0, res[2])
self.checkCmd(os.path.join(FIXTURES, 'koredis.cmd'), self.checkCmd(os.path.join(FIXTURES, "koredis.cmd"), FAKE_CMD_LIST)
FAKE_CMD_LIST)
class TestDlgCntClean(Test):
env = {"DLG_CONFIG": os.path.join(FIXTURES, "dlgcnt.lua.cfg")}
def setUp(self):
self.command = copy.deepcopy(command)
def test_wrong_option(self):
self.command = ["./scripts/ngcp-dlgcnt-clean"]
res = executeAndReturnOutput(self.command)
self.assertEqual(res[0], 1, res[2])
self.assertRegex(res[2], "missing argument 'callid'")
self.checkNotCmd()
def test_config_defaults_c(self):
self.command = ["./scripts/ngcp-dlgcnt-clean", "-c", "fake"]
res = executeAndReturnOutput(self.command)
self.assertRegex(res[2], "using defaults")
self.assertEqual(res[0], 0, res[2])
self.assertRegex(res[1], "^4\n")
self.checkNotCmd()
def test_config_defaults_C(self):
self.command = ["./scripts/ngcp-dlgcnt-clean", "-C", "fake"]
res = executeAndReturnOutput(self.command)
self.assertRegex(res[2], "using defaults")
self.assertEqual(res[0], 0, res[2])
self.assertRegex(res[1], "^127.0.0.1\n")
self.checkNotCmd()
def test_config_c(self):
self.command = ["./scripts/ngcp-dlgcnt-clean", "-c", "fake"]
res = executeAndReturnOutput(self.command, self.env)
self.assertNotRegex(res[2], "using defaults")
self.assertEqual(res[0], 0, res[2])
self.assertRegex(res[1], "^4\n")
self.checkNotCmd()
def test_config_C(self):
self.command = ["./scripts/ngcp-dlgcnt-clean", "-C", "fake"]
res = executeAndReturnOutput(self.command, self.env)
self.assertNotRegex(res[2], "using defaults")
self.assertEqual(res[0], 0, res[2])
self.assertRegex(res[1], "^127.2.0.1\n")
self.checkNotCmd()
if __name__ == '__main__': if __name__ == "__main__":
unittest.main( unittest.main(
# these make sure that some options that are not applicable # these make sure that some options that are not applicable
# remain hidden from the help menu. # remain hidden from the help menu.
failfast=False, buffer=False, catchbreak=False) failfast=False,
buffer=False,
catchbreak=False,
)

Write Preview
Loading…
Cancel
Save