From 496b06658eb2f6475e77be35c2f01cf6444fb232 Mon Sep 17 00:00:00 2001 From: Victor Seva Date: Fri, 4 Mar 2022 14:45:42 +0100 Subject: [PATCH] TT#146450 scripts: fix ngcp-dlgcnt-clean use for get configs * callid parameter was missing * add some tests for those cases Change-Id: I4456290de22ccf3d3955470ac1cc05ac4fd62c4b --- scripts/ngcp-dlgcnt-check | 4 +- scripts/ngcp-dlgcnt-clean | 18 +++- tests/test_dlgcnt.py | 216 +++++++++++++++++++++++--------------- 3 files changed, 149 insertions(+), 89 deletions(-) diff --git a/scripts/ngcp-dlgcnt-check b/scripts/ngcp-dlgcnt-check index 0fc9d63..3ed6b24 100755 --- a/scripts/ngcp-dlgcnt-check +++ b/scripts/ngcp-dlgcnt-check @@ -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) diff --git a/scripts/ngcp-dlgcnt-clean b/scripts/ngcp-dlgcnt-clean index 001e156..3b83d5c 100755 --- a/scripts/ngcp-dlgcnt-clean +++ b/scripts/ngcp-dlgcnt-clean @@ -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 diff --git a/tests/test_dlgcnt.py b/tests/test_dlgcnt.py index e89c0d4..f289e64 100755 --- a/tests/test_dlgcnt.py +++ b/tests/test_dlgcnt.py @@ -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, + )