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
(cherry picked from commit 496b06658e)
mr10.3
Victor Seva 4 years ago committed by Víctor Seva
parent 3a008159b2
commit 91ccd53952
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
# read kamailio.proxy.dlgcnt.pair_redis_db
REDIS_DB="$(ngcp-dlgcnt-clean -c 2>/dev/null|| echo 4)"
host="$(ngcp-dlgcnt-clean -C 2>/dev/null|| echo localhost)"
REDIS_DB="$(ngcp-dlgcnt-clean -c fake 2>/dev/null|| echo 4)"
host="$(ngcp-dlgcnt-clean -C fake 2>/dev/null|| echo localhost)"
# redis full list
REDIS_CALLIDS=$(mktemp)

18
scripts/ngcp-dlgcnt-clean
Unescape View File

@ -1,6 +1,5 @@
#!/usr/bin/env lua5.1
--local ut = require 'ngcp.utils'.table
local NGCPDlg = require 'ngcp.dlgcnt'
local config = "/etc/kamailio/proxy/dlgcnt.lua.cfg"
local argparse = require "argparse"
@ -53,16 +52,25 @@ local args = parser:parse()
get_config()
local dlg = NGCPDlg:new()
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)
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)
end
local NGCPDlg = require 'ngcp.dlgcnt'
local dlg = NGCPDlg:new()
if dlg_config then
dlg.config.central.host = dlg_config.central.host
dlg.config.central.port = dlg_config.central.port

216
tests/test_dlgcnt.py
Unescape View File

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

Write Preview
Loading…
Cancel
Save