Commit a73a3ef0 authored by Shane Kerr's avatar Shane Kerr
Browse files

Working show_processes with test.

parent cf020fa4
......@@ -270,6 +270,14 @@ class BoB:
answer = isc.config.ccsession.create_answer(0)
return answer
def get_processes(self):
pids = list(self.processes.keys())
pids.sort()
process_list = [ ]
for pid in pids:
process_list.append([ pid, self.processes[pid].name ])
return process_list
def command_handler(self, command, args):
if self.verbose:
sys.stdout.write("[bind10] Boss got command: " + command + "\n")
......@@ -280,7 +288,13 @@ class BoB:
if command == "shutdown":
self.runnable = False
answer = isc.config.ccsession.create_answer(0)
elif command == "ping":
answer = isc.config.ccsession.create_answer(0, "pong")
elif command == "show_processes":
answer = isc.config.ccsession.create_answer(0,
self.get_processes())
else:
print("should be doing the unknown command thingy")
answer = isc.config.ccsession.create_answer(1,
"Unknown command")
return answer
......
......@@ -26,6 +26,11 @@
"command_name": "ping",
"command_description": "Ping the boss process",
"command_args": []
},
{
"command_name": "show_processes",
"command_description": "List the running BIND 10 processes",
"command_args": []
}
]
}
......
PYCOVERAGE_RUN = @PYCOVERAGE_RUN@
#PYTESTS = args_test.py bind10_test.py bind10_cmd_test.py
PYTESTS = bind10_test.py bind10_cmd_test_test.py
PYTESTS = bind10_test.py bind10_cmd_test.py
EXTRA_DIST = $(PYTESTS)
# test using command-line arguments, so use check-local target instead of TESTS
......@@ -13,5 +13,6 @@ endif
for pytest in $(PYTESTS) ; do \
echo Running test: $$pytest ; \
env PYTHONPATH=$(abs_top_srcdir)/src/lib/python:$(abs_top_builddir)/src/lib/python:$(abs_top_builddir)/src/bin/bind10 \
$(PYCOVERAGE_RUN) $(abs_srcdir)/$$pytest || exit ; \
BIND10_MSGQ_SOCKET_FILE=$(abs_top_builddir)/msgq_socket \
$(PYCOVERAGE_RUN) $(abs_srcdir)/$$pytest || exit ; \
done
......@@ -10,7 +10,9 @@ import isc.cc
BIND10_EXE="../run_bind10.sh"
TIMEOUT=3
class TestBossBindctl(unittest.TestCase):
# TODO: single setup
class TestBossCmd(unittest.TestCase):
def _waitForString(self, bob, s):
"""Read the input from the Process object until we find the
string we're looking for or we timeout."""
......@@ -27,20 +29,99 @@ class TestBossBindctl(unittest.TestCase):
break
return found_string
def testBasicBindctl(self):
"""Run basic bindctl"""
def testPing(self):
"""Simple aliveness check"""
ping_worked = False
# start bind10
bob = subprocess.Popen(args=(BIND10_EXE,),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
started_ok = self._waitForString(bob, '[bind10] BIND 10 started')
if started_ok:
# wrap everything in a try block so we don't leave
# stray processes around on error
try:
# connect to the command channel
self.cc = isc.cc.Session()
self.cc.group_subscribe('Boss')
# send a ping
cmd = { "command": ['ping']}
seq = self.cc.group_sendmsg(cmd, 'Boss')
# wait for a pong
env, msg = self.cc.recvmsg(False, seq)
if 'result' in msg:
result = msg['result']
if (result[0] == 0) and (result[1] == 'pong'):
ping_worked = True
except:
pass
# shut down
bob.terminate()
bob.wait()
# check that we were able to ping
self.assertEqual(ping_worked, True)
def _check_processes(self, process_list):
# the msgq and cfgmgr are required, everything else is optional
msgq_pid = None
cfgmgr_pid = None
for process in process_list:
if len(process) != 2:
return False
if type(process[0]) != int:
return False
if type(process[1]) != str:
return False
if process[1] == 'b10-msgq':
msgq_pid = process[0]
elif process[1] == 'b10-cfgmgr':
cfgmgr_pid = process[0]
if msgq_pid and cfgmgr_pid:
return True
return False
def testShowServices(self):
"""Get a list of children"""
command_worked = False
# start bind10
bob = subprocess.Popen(args=(BIND10_EXE,),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
started_ok = self._waitForString(bob, '[bind10] BIND 10 started')
# connect to the command channel
self.cc = isc.cc.Session()
if started_ok:
# wrap everything in a try block so we don't leave
# stray processes around on error
try:
# connect to the command channel
self.cc = isc.cc.Session()
self.cc.group_subscribe('Boss')
# send a ping
cmd = { "command": ['show_processes']}
seq = self.cc.group_sendmsg(cmd, 'Boss')
# wait for a pong
env, msg = self.cc.recvmsg(False, seq)
if 'result' in msg:
result = msg['result']
if (result[0] == 0) and self._check_processes(result[1]):
command_worked = True
except:
pass
# shut down
bob.terminate()
bob.wait()
# check that we were able to ping
self.assertEqual(command_worked, True)
if __name__ == '__main__':
unittest.main()
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment