aiden
/
appmotor
mirror of https://github.com/cutefishos/appmotor
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Changes: Replace time.sleep in invoker test with loops waiting for some condition.

Browse Source 
RevBy: Thyagarajan Balakrishnan <ext-thyagarajan.1.balakrishnan@nokia.com>
pull/1/head
Marek Ruszczak 14 years ago
parent e96d67e8ca
commit e4146389fa
2 changed files with 144 additions and 139 deletions
Show Stats Download Patch File Download Diff File

267
tests/harmattan/testscripts/test-invoker.py
Unescape Escape View File

@ -16,7 +16,7 @@
# of this file. # of this file.
""" """
This program tests the functionalities of invoker This program tests the functionalities of invoker
Authors: nimika.1.keshri@nokia.com Authors: nimika.1.keshri@nokia.com
""" """
@ -31,7 +31,7 @@ from subprocess import Popen
from utils import * from utils import *
from os.path import basename from os.path import basename
class InvokerTests(unittest.TestCase): class InvokerTests(CustomTestCase):
def setUp(self): def setUp(self):
debug("setUp") debug("setUp")
if daemons_running(): if daemons_running():
@ -51,7 +51,7 @@ class InvokerTests(unittest.TestCase):
if self.START_DAEMONS_AT_TEARDOWN: if self.START_DAEMONS_AT_TEARDOWN:
start_daemons() start_daemons()
def sighup_applauncherd(self): def sighup_applauncherd(self):
same_pid, booster_status = send_sighup_to_applauncherd() same_pid, booster_status = send_sighup_to_applauncherd()
self.assert_(same_pid, "Applauncherd has new pid after SIGHUP") self.assert_(same_pid, "Applauncherd has new pid after SIGHUP")
self.assert_(booster_status, "Atleast one of the boosters is not restarted") self.assert_(booster_status, "Atleast one of the boosters is not restarted")
@ -75,34 +75,37 @@ class InvokerTests(unittest.TestCase):
self.sighup_applauncherd() self.sighup_applauncherd()
self.test_wait_term_e(False) self.test_wait_term_e(False)
def _test_wait_term(self, app_path, app_name, btype): def _test_wait_term(self, app_path, app_name, btype, directlyKillInvoker=True):
""" """
When calling invoker with --wait-term and killing invoker, When calling invoker with --wait-term and killing invoker,
the launched application should die too. the launched application should die too.
""" """
kill_process(app_name)
# Launch the app with invoker using --wait-term # Launch the app with invoker using --wait-term
p = run_app_as_user_with_invoker(app_path,booster = btype, arg = '--wait-term' ) p = run_app_as_user_with_invoker(app_path,booster = btype, arg = '--wait-term' )
time.sleep(4)
# Retrieve their pids # Retrieve their pids
invoker_pid = wait_for_app('invoker')
app_pid = wait_for_app(app_name) app_pid = wait_for_app(app_name)
# Make sure that both apps started # Make sure that both apps started
self.assert_(invoker_pid != None, "invoker not executed?")
self.assert_(app_pid != None, "%s not launched by invoker?" % app_path) self.assert_(app_pid != None, "%s not launched by invoker?" % app_path)
# Send SIGTERM to invoker, the launched app should die # Send SIGTERM to invoker, the launched app should die
debug(" Send SIGTERM to invoker, the launched app should die") if directlyKillInvoker :
kill_process(None, invoker_pid, 15) debug(" Send SIGTERM to invoker, the launched app should die")
kill_process('invoker', None, 15)
self.waitForAsertEqual(lambda: p.poll(), 143, "invoker was not killed.", timeout = 3)
else :
debug(" Send SIGTERM to shell which runs invoker, the launched app should die")
kill_process(None, p.pid, 15)
self.waitForAsertEqual(lambda: p.poll(), -15, "invoker was not killed.", timeout = 3)
time.sleep(4) debug("Invoker return code: %s" %(p.poll()))
# This should be None self.waitForAsertEqual(lambda: get_pid(app_name), None,
app_pid2 = get_pid(app_name) "Application '%s' was not killed!" % app_path,
self.assert_(app_pid2 == None, "%s was not killed" % app_path) timeout = 8)
def test_launch_wo_applauncherd(self): def test_launch_wo_applauncherd(self):
""" """
@ -114,7 +117,7 @@ class InvokerTests(unittest.TestCase):
stop_applauncherd() stop_applauncherd()
if get_pid(PREFERED_APP) != None: if get_pid(PREFERED_APP) != None:
kill_process(PREFERED_APP) kill_process(PREFERED_APP)
p = run_app_as_user_with_invoker(PREFERED_APP, booster = 'm') p = run_app_as_user_with_invoker(PREFERED_APP, booster = 'm')
pid1 = wait_for_app(PREFERED_APP) pid1 = wait_for_app(PREFERED_APP)
self.assert_(pid1 != None, "Application was executed") self.assert_(pid1 != None, "Application was executed")
@ -144,40 +147,32 @@ class InvokerTests(unittest.TestCase):
#Test for m-booster #Test for m-booster
debug("Test for m-booster") debug("Test for m-booster")
st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_m.py") st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_m.py")
time.sleep(3)
debug("The Invoker killed by : <%s>" %op.split ('\n')[-1]) debug("The Invoker killed by : <%s>" %op.split ('\n')[-1])
self.assert_(op.split('\n')[-1].startswith('Segmentation fault') == True, "The invoker(m-booster) was not killed by the same signal") self.assert_(op.split('\n')[-1].startswith('Segmentation fault') == True, "The invoker(m-booster) was not killed by the same signal")
time.sleep(2)
#Test for d-booster #Test for d-booster
debug("Test for d-booster") debug("Test for d-booster")
st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_d.py") st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_d.py")
time.sleep(3)
debug("The Invoker killed by : %s" % op.split('\n')[-1]) debug("The Invoker killed by : %s" % op.split('\n')[-1])
self.assert_(op.split('\n')[-1].startswith('Terminated') == True, "The invoker(d-booster) was not killed by the same signal") self.assert_(op.split('\n')[-1].startswith('Terminated') == True, "The invoker(d-booster) was not killed by the same signal")
time.sleep(2)
#Test for e-booster #Test for e-booster
debug("Test for e-booster") debug("Test for e-booster")
st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_e.py") st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_e.py")
time.sleep(3)
debug("The Invoker killed by : %s" % op.split('\n')[-1]) debug("The Invoker killed by : %s" % op.split('\n')[-1])
self.assert_(op.split('\n')[-1].startswith('Terminated') == True, "The invoker(e-booster) was not killed by the same signal") self.assert_(op.split('\n')[-1].startswith('Terminated') == True, "The invoker(e-booster) was not killed by the same signal")
time.sleep(2)
#This case is launching the application in user mode #This case is launching the application in user mode
#Test for q-booster #Test for q-booster
debug("Test for q-booster") debug("Test for q-booster")
st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_qt.py") st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_sf_qt.py")
time.sleep(3)
debug("The Invoker killed by : %s" %op.split('\n')[-1]) debug("The Invoker killed by : %s" %op.split('\n')[-1])
self.assert_(op.split('\n')[-1].startswith('Aborted') == True, "The invoker(q-booster) was not killed by the same signal") self.assert_(op.split('\n')[-1].startswith('Aborted') == True, "The invoker(q-booster) was not killed by the same signal")
time.sleep(2)
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
self.test_signal_forwarding(False) self.test_signal_forwarding(False)
@ -188,32 +183,39 @@ class InvokerTests(unittest.TestCase):
Test the --delay parameter of the invoker. Test the --delay parameter of the invoker.
""" """
invokerDelay = 10
# launch an app with invoker --delay n # launch an app with invoker --delay n
debug("launching fala_ft_hello ...") debug("launching fala_ft_hello ...")
p = Popen(['/usr/bin/invoker', '--delay', '10', '--type=m', '--no-wait', p = Popen(['/usr/bin/invoker', '--delay', str(invokerDelay), '--type=m', '--no-wait',
'/usr/bin/fala_ft_hello'], '/usr/bin/fala_ft_hello'],
shell=False, shell=False,
stdout=DEV_NULL, stderr=DEV_NULL) stdout=DEV_NULL, stderr=DEV_NULL)
# wait a little pStartTime = time.time()
debug("waiting ...") appPid = wait_for_app('fala_ft_hello', timeout=7)
time.sleep(5) try:
falaStartTime = time.time()
success = True
# wait a little
if p.poll() == None: debug("Waiting for invoker to terminate after %.1fs..." %(invokerDelay))
debug("NOT DEAD") # time out loop, if invoker runs longer then invokerDelay + 1 this means that something is wrong
else: while time.time()<pStartTime + invokerDelay + 1 and p.poll()==None :
debug("DEAD") time.sleep(0.5)
success = False
endTime = time.time()
debug("waiting for invoker to terminate ...") # check that invoker has terminated before timeout
p.wait() if p.poll()==None :
p.terminate()
debug("terminating fala_ft_hello ...") self.assert_(False, "invoker didn't terminate on time. "
Popen(['pkill', 'fala_ft_hello']).wait() "It was running for: %.1fs (%.1fs was expected)."
%(endTime - pStartTime, invokerDelay))
self.assert_(success, "invoker terminated before delay elapsed") debug("Invoker has terminated after %.1fs." %(endTime - pStartTime))
self.assert_(endTime - pStartTime > invokerDelay - 1,
"invoker has terminated to soon: %.1fs (%.1fs was expected)" %(endTime - pStartTime, invokerDelay))
self.assertEqual(get_pid('fala_ft_hello'), appPid, "Test application shouldn't be killed.")
finally:
kill_process('fala_ft_hello')
debug("Ok.")
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
@ -228,19 +230,19 @@ class InvokerTests(unittest.TestCase):
st, op = commands.getstatusoutput('/usr/bin/fala_status') st, op = commands.getstatusoutput('/usr/bin/fala_status')
app_st_wo_inv = os.WEXITSTATUS(st) app_st_wo_inv = os.WEXITSTATUS(st)
debug("The exit status of app without invoker is : %d" %app_st_wo_inv) debug("The exit status of app without invoker is : %d" %app_st_wo_inv)
#Run application with invoker and get the exit status - booster-m case #Run application with invoker and get the exit status - booster-m case
debug("Run application with invoker and get the exit status") debug("Run application with invoker and get the exit status")
st, op = commands.getstatusoutput('invoker --type=m --wait-term /usr/bin/fala_status') st, op = commands.getstatusoutput('invoker --type=m --wait-term /usr/bin/fala_status')
app_st_w_inv = os.WEXITSTATUS(st) app_st_w_inv = os.WEXITSTATUS(st)
debug("The exit status of app with invoker (booster-m) is : %d" %app_st_w_inv) debug("The exit status of app with invoker (booster-m) is : %d" %app_st_w_inv)
#Run application with invoker and get the exit status - booster-e case #Run application with invoker and get the exit status - booster-e case
debug("Run application with invoker and get the exit status") debug("Run application with invoker and get the exit status")
st, op = commands.getstatusoutput('invoker --type=e --wait-term /usr/bin/fala_status') st, op = commands.getstatusoutput('invoker --type=e --wait-term /usr/bin/fala_status')
app_st_we_inv = os.WEXITSTATUS(st) app_st_we_inv = os.WEXITSTATUS(st)
debug("The exit status of app with invoker (booster-e) is : %d" %app_st_we_inv) debug("The exit status of app with invoker (booster-e) is : %d" %app_st_we_inv)
self.assert_(app_st_wo_inv == app_st_w_inv, "The invoker returns a wrong exit status for booster-m") self.assert_(app_st_wo_inv == app_st_w_inv, "The invoker returns a wrong exit status for booster-m")
self.assert_(app_st_wo_inv == app_st_we_inv, "The invoker returns a wrong exit status for booster-e") self.assert_(app_st_wo_inv == app_st_we_inv, "The invoker returns a wrong exit status for booster-e")
@ -257,35 +259,40 @@ class InvokerTests(unittest.TestCase):
# invoker searches PATH for the executable # invoker searches PATH for the executable
p = run_app_as_user_with_invoker(PREFERED_APP, booster = 'm', arg = '--no-wait') p = run_app_as_user_with_invoker(PREFERED_APP, booster = 'm', arg = '--no-wait')
self.assert_(p.wait() == 0, "Couldn't launch fala_ft_hello") try:
time.sleep(2) self.waitForAsertEqual(lambda: p.poll(), 0, "Couldn't launch fala_ft_hello")
kill_process('fala_ft_hello') finally:
wait_for_app('fala_ft_hello')
kill_process('fala_ft_hello')
# launch with relative path # launch with relative path
p = run_cmd_as_user('cd /usr/share/; invoker --type=m --no-wait ' + p = run_cmd_as_user('cd /usr/share/; invoker --type=m --no-wait ' +
"../bin/fala_ft_hello") "../bin/fala_ft_hello")
self.assert_(p.wait() == 0, "Couldnt launch fala_ft_hello" + try:
" with relative path") self.waitForAsertEqual(lambda: p.poll(), 0, "Couldn't launch fala_ft_hello" +
time.sleep(2) " with relative path or invoker didn't terminate")
kill_process('fala_ft_hello') finally:
wait_for_app('fala_ft_hello')
kill_process('fala_ft_hello')
# and finally, try to launch something that doesn't exist # and finally, try to launch something that doesn't exist
p = run_cmd_as_user('invoker --type=m --no-wait spam_cake') p = run_cmd_as_user('invoker --type=m --no-wait spam_cake')
self.assert_(p.wait() != 127, "Found spam_cakefor some reason") try :
time.sleep(2) self.waitForAsertEqual(lambda: p.poll(), 1, "spam_cake has been found!")
kill_process('spam_cake') finally:
kill_process('spam_cake')
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
self.test_invoker_search_prog(False) self.test_invoker_search_prog(False)
def test_invoker_gid_uid(self, sighup = True): def test_invoker_gid_uid(self, sighup = True):
""" """
To Test that the set gid and uid is passed from invoker process to launcher To Test that the set gid and uid is passed from invoker process to launcher
""" """
debug("This test uses a test application that returns the uid and gid and exits") debug("This test uses a test application that returns the uid and gid and exits")
#get the id in user mode #get the id in user mode
debug("Get the system's uid and gid in User Mode \n") debug("Get the system's uid and gid in User Mode \n")
st, op = commands.getstatusoutput('su user -c ' "id") st, op = commands.getstatusoutput('su user -c ' "id")
usr_id1 = op.split(' ')[0].split('(')[0] usr_id1 = op.split(' ')[0].split('(')[0]
@ -294,12 +301,12 @@ class InvokerTests(unittest.TestCase):
#get invokers's uid and gid by running the application using invoker in user mode #get invokers's uid and gid by running the application using invoker in user mode
debug("get invoker's uid and gid by running the application using invoker in user mode") debug("get invoker's uid and gid by running the application using invoker in user mode")
app = "invoker --type=m --no-wait /usr/bin/fala_status" app = "invoker --type=m --no-wait /usr/bin/fala_status"
st, op = commands.getstatusoutput('su user -c "%s"' %app ); st, op = commands.getstatusoutput('su user -c "%s"' %app );
usr_id = op.split('\n')[1] usr_id = op.split('\n')[1]
grp_id = op.split('\n')[2] grp_id = op.split('\n')[2]
debug("Invoker %s \tInvoker %s" %(usr_id, grp_id)) debug("Invoker %s \tInvoker %s" %(usr_id, grp_id))
#get application's uid and gid by running the application without invoker in user mode #get application's uid and gid by running the application without invoker in user mode
debug("get application's uid and gid by running the application without invoker in user mode") debug("get application's uid and gid by running the application without invoker in user mode")
app = "/usr/bin/fala_status" app = "/usr/bin/fala_status"
@ -314,7 +321,7 @@ class InvokerTests(unittest.TestCase):
self.assert_(usr_id == usr_id2, "The correct UID is not passed by invoker") self.assert_(usr_id == usr_id2, "The correct UID is not passed by invoker")
self.assert_(grp_id == grp_id2, "The correct GID is not passed by invoker") self.assert_(grp_id == grp_id2, "The correct GID is not passed by invoker")
#get the id in root mode #get the id in root mode
debug("Get the Sysem's uid and gid in Root Mode \n") debug("Get the Sysem's uid and gid in Root Mode \n")
st, op = commands.getstatusoutput("id") st, op = commands.getstatusoutput("id")
usr_id1 = op.split(' ')[0].split('(')[0] usr_id1 = op.split(' ')[0].split('(')[0]
@ -323,12 +330,12 @@ class InvokerTests(unittest.TestCase):
#get id by running the application using invoker in root mode #get id by running the application using invoker in root mode
debug("get invoker's uid and gid by running the application using invoker in root mode") debug("get invoker's uid and gid by running the application using invoker in root mode")
app = "invoker --type=m --no-wait /usr/bin/fala_status" app = "invoker --type=m --no-wait /usr/bin/fala_status"
st, op = commands.getstatusoutput("%s" %app ); st, op = commands.getstatusoutput("%s" %app );
usr_id = op.split('\n')[1] usr_id = op.split('\n')[1]
grp_id = op.split('\n')[2] grp_id = op.split('\n')[2]
debug("Invoker %s \tInvoker %s" %(usr_id, grp_id)) debug("Invoker %s \tInvoker %s" %(usr_id, grp_id))
#get id by running the application without invoker in root mode #get id by running the application without invoker in root mode
debug("get application's uid and gid by running the application without invoker in root mode") debug("get application's uid and gid by running the application without invoker in root mode")
app = "/usr/bin/fala_status" app = "/usr/bin/fala_status"
@ -346,7 +353,7 @@ class InvokerTests(unittest.TestCase):
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
self.test_invoker_gid_uid(False) self.test_invoker_gid_uid(False)
def test_invoker_param_creds(self): def test_invoker_param_creds(self):
p = run_cmd_as_user('invoker --creds') p = run_cmd_as_user('invoker --creds')
@ -354,20 +361,18 @@ class InvokerTests(unittest.TestCase):
def test_invoker_param_respawn_delay(self, sighup = True): def test_invoker_param_respawn_delay(self, sighup = True):
p = run_cmd_as_user('invoker --respawn 10 --type=q --no-wait fala_ft_hello') p = run_cmd_as_user('invoker --respawn 10 --type=q --no-wait fala_ft_hello')
try :
time.sleep(7) startTime = time.time()
wait_for_app('fala_ft_hello')
pid = get_pid('booster-q') pid = wait_for_app('booster-q', timeout = startTime - time.time() + 12)
self.assert_(pid == None, "'booster-q' was respawned too soon") endTime = time.time()
self.assert_(endTime-startTime>8, "'booster-q' was respawned too soon: %.1fs" %(endTime-startTime))
time.sleep(7) self.assert_(pid != None, "'booster-q' was not respawned on time.")
pid = get_pid('booster-q') p = run_cmd_as_user('invoker --respawn 256 --type=q --no-wait fala_ft_hello')
self.assert_(pid != None, "'booster-q' was not respawned in time") self.assert_(p.wait() != 0, "invoker didn't die with too big respawn delay")
finally:
p = run_cmd_as_user('invoker --respawn 256 --type=q --no-wait fala_ft_hello') kill_process('fala_ft_hello')
self.assert_(p.wait() != 0, "invoker didn't die with too big respawn delay")
kill_process('fala_ft_hello')
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
@ -382,15 +387,13 @@ class InvokerTests(unittest.TestCase):
def test_invoker_signal_forward(self, sighup = True): def test_invoker_signal_forward(self, sighup = True):
""" """
Test that UNIX signals are forwarded from Test that UNIX signals are forwarded from
invoker to the invoked process invoker to the invoked process
""" """
st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_inv_sf.py") st, op = commands.getstatusoutput("/usr/share/applauncherd-testscripts/signal-forward/fala_inv_sf.py")
time.sleep(3)
debug("The Invoker killed by : <%s>" %op) debug("The Invoker killed by : <%s>" %op)
self.assert_(op.startswith('Segmentation fault') == True, "The invoker(m-booster) was not killed by the same signal") self.assert_(op.startswith('Segmentation fault') == True, "The invoker(m-booster) was not killed by the same signal")
time.sleep(2)
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
@ -409,7 +412,7 @@ class InvokerTests(unittest.TestCase):
def test_unknown_parameter(self): def test_unknown_parameter(self):
""" """
Test that help is printed if unknown parameter Test that help is printed if unknown parameter
is passed to invoker is passed to invoker
""" """
cmd = '/usr/bin/invoker --type=m --x /usr/bin/fala_wl' cmd = '/usr/bin/invoker --type=m --x /usr/bin/fala_wl'
@ -419,7 +422,7 @@ class InvokerTests(unittest.TestCase):
def test_non_existing_binary_launch(self): def test_non_existing_binary_launch(self):
""" """
Test that invoker gives error when it tries to launch Test that invoker gives error when it tries to launch
a binary that does not exist a binary that does not exist
""" """
cmd = '/usr/bin/invoker --type=m /usr/bin/fala_foo' cmd = '/usr/bin/invoker --type=m /usr/bin/fala_foo'
@ -460,8 +463,7 @@ class InvokerTests(unittest.TestCase):
os.system("ln -s /usr/bin/fala_wl /usr/bin/fala_link") os.system("ln -s /usr/bin/fala_wl /usr/bin/fala_link")
cmd = "/usr/bin/invoker --type=m /usr/bin/fala_link" cmd = "/usr/bin/invoker --type=m /usr/bin/fala_link"
os.system('su - user -c "%s"&' %cmd) os.system('su - user -c "%s"&' %cmd)
time.sleep(3) pid = wait_for_app("fala_link")
pid = get_pid("fala_link")
os.system("rm /usr/bin/fala_link") os.system("rm /usr/bin/fala_link")
kill_process("fala_link") kill_process("fala_link")
self.assert_(pid != None ,"The application was not launched") self.assert_(pid != None ,"The application was not launched")
@ -477,16 +479,18 @@ class InvokerTests(unittest.TestCase):
-D = INVOKER_MSG_MAGIC_OPTION_DLOPEN_DEEP -D = INVOKER_MSG_MAGIC_OPTION_DLOPEN_DEEP
""" """
p = run_app_as_user_with_invoker("/usr/bin/fala_wl", booster='m', arg = '-G') p = run_app_as_user_with_invoker("/usr/bin/fala_wl", booster='m', arg = '-G')
time.sleep(3) wait_for_app('fala_wl')
pid = p.pid returnCode = p.poll()
kill_process("fala_wl") kill_process("fala_wl")
self.assert_(pid != None ,"The application was not launched") p.wait()
self.assert_(returnCode == None ,"The application was not launched")
p = run_app_as_user_with_invoker("/usr/bin/fala_wl", booster='m', arg = '-D') p = run_app_as_user_with_invoker("/usr/bin/fala_wl", booster='m', arg = '-D')
time.sleep(3) wait_for_app('fala_wl')
pid = p.pid returnCode = p.poll()
kill_process("fala_wl") kill_process("fala_wl")
self.assert_(pid != None ,"The application was not launched") p.wait()
self.assert_(returnCode == None ,"The application was not launched")
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
@ -494,12 +498,11 @@ class InvokerTests(unittest.TestCase):
def test_app_directory(self, sighup = True): def test_app_directory(self, sighup = True):
""" """
Test that invoker is unable to launch a application which is a directory Test that invoker is unable to launch a application which is a directory
""" """
#Test for a directory #Test for a directory
os.system("mkdir /usr/bin/fala_dir") os.system("mkdir /usr/bin/fala_dir")
st, op = commands.getstatusoutput("/usr/bin/invoker --type=m /usr/bin/fala_dir") st, op = commands.getstatusoutput("/usr/bin/invoker --type=m /usr/bin/fala_dir")
time.sleep(3)
os.system("rm -rf /usr/bin/fala_dir") os.system("rm -rf /usr/bin/fala_dir")
self.assert_(st != 0 ,"The application was not launched") self.assert_(st != 0 ,"The application was not launched")
@ -522,31 +525,36 @@ class InvokerTests(unittest.TestCase):
Check that application is started and invoker is waiting termination before exit Check that application is started and invoker is waiting termination before exit
""" """
kill_process('fala_wait')
# launch an app with invoker --wait-term # launch an app with invoker --wait-term
debug("launching fala_wait ...") debug("launching fala_wait ...")
p = Popen(['/usr/bin/invoker', '--type=m', '--wait-term', p = Popen(['/usr/bin/invoker', '--type=m', '--wait-term',
'/usr/bin/fala_wait'], '/usr/bin/fala_wait'],
shell=False, shell=False,
stdout=DEV_NULL, stderr=DEV_NULL) stdout=DEV_NULL, stderr=DEV_NULL)
# wait a little # wait a little
debug("waiting ...") debug("waiting ...")
time.sleep(5) wait_for_app('fala_wait', sleep = 0.3)
# fala_wait is quits after 6 seconds
endTime = time.time() + 6
success = True success = True
while time.time()<endTime-1 :
if p.poll() == None: if p.poll() != None:
debug("NOT DEAD") debug("invoker has end working before it was expected")
else: success = False
debug("DEAD") break
success = False time.sleep(0.5)
debug("waiting for invoker to terminate ...") if success :
p.wait() debug("waiting for invoker to terminate ...")
self.waitForAsert(lambda : p.poll() != None,
debug("terminating fala_wait ...") "invoker didn't terminate")
debug("...invoker has been terminated.")
self.assert_(success, "invoker terminated before delay elapsed") self.assertEqual(get_pid('fala_wait'), None, "invoker died before fala_wait application.")
self.assert_(success, "invoker has ended execution before expected time")
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
@ -573,15 +581,11 @@ class InvokerTests(unittest.TestCase):
p = run_app_as_user_with_invoker(PREFERED_APP, booster = 'm', p = run_app_as_user_with_invoker(PREFERED_APP, booster = 'm',
arg = '-o') arg = '-o')
time.sleep(4) pid = wait_for_app(PREFERED_APP)
pid = get_pid(PREFERED_APP)
st, op = commands.getstatusoutput('cat /proc/%s/oom_adj' % pid) st, op = commands.getstatusoutput('cat /proc/%s/oom_adj' % pid)
kill_process(PREFERED_APP)
self.assert_(op == '-2', "oom.adj of the launched process is not -1") self.assert_(op == '-2', "oom.adj of the launched process is not -2."
"cat return code: %s, oom_adj contains: '%s'" %(st, op))
kill_process(PREFERED_APP)
if(sighup): if(sighup):
self.sighup_applauncherd() self.sighup_applauncherd()
@ -604,12 +608,11 @@ class InvokerTests(unittest.TestCase):
self.assert_(pid != None, "The application was not launched") self.assert_(pid != None, "The application was not launched")
stop_applauncherd() stop_applauncherd()
time.sleep(20) #wait app to be terminated and invoker to exit
pid = wait_for_app(PREFERED_APP, timeout = 2) #don't need to wait long since the app supposed not running self.assert_(wait_for_process_end('applauncherd'), "applauncherd was not stoped.")
if pid != None: self.waitForAsert(lambda:p.poll() != None, "Invoker didn't terminate.")
kill_process(PREFERED_APP) self.assert_(wait_for_process_end(PREFERED_APP), "invoker didn't kill application after applauncherd termination.")
self.assert_(pid == None, "The application is still running!")
# main # main
if __name__ == '__main__': if __name__ == '__main__':

16
tests/harmattan/testscripts/utils.py
Unescape Escape View File

@ -5,6 +5,7 @@ import time
import sys import sys
import re import re
import unittest import unittest
import types
from subprocess import Popen from subprocess import Popen
from os.path import basename from os.path import basename
@ -26,28 +27,29 @@ class CustomTestCase(unittest.TestCase) :
type(periodicCheck) == types.LambdaType) type(periodicCheck) == types.LambdaType)
start = time.time() start = time.time()
stop = start + timeout stop = start + timeout
newValue = None
debug("Waiting for '%s' to return True value in time of %.1fs." %(periodicCheck.__doc__, timeout)) debug("Waiting for '%s' to return True value in time of %.1fs." %(periodicCheck.__name__, timeout))
while stop > time.time() : while stop > time.time() :
if periodicCheck() : if periodicCheck() :
debug("'%s' has returned True after %.1fs." %(periodicCheck.__doc__, time.time()-start)) debug("'%s' has returned True after %.1fs." %(periodicCheck.__name__, time.time()-start))
return return
time.sleep(sleep) time.sleep(sleep)
debug("waitForAsert has timed out after %ss." %timeout) debug("waitForAsert has timed out after %ss." %timeout)
self.assert_(periodicCheck(), msg) self.assert_(periodicCheck(), msg)
return return
def waitForAsertEqual(self, periodickCheck, expectedValue, msg="", timeout=20, sleep=1) : def waitForAsertEqual(self, periodicCheck, expectedValue, msg="", timeout=20, sleep=1) :
assert(type(periodicCheck) == types.FunctionType or assert(type(periodicCheck) == types.FunctionType or
type(periodicCheck) == types.LambdaType) type(periodicCheck) == types.LambdaType)
start = time.time() start = time.time()
stop = start + timeout stop = start + timeout
debug("Waiting for '%s' to return True value in time of %.1fs." %(periodicCheck.__doc__, timeout)) debug("Waiting for '%s' to return expected value: '%s' in time of %.1fs."
%(periodicCheck.__name__, expectedValue, timeout))
while stop > time.time() : while stop > time.time() :
if periodicCheck() == expectedValue : if periodicCheck() == expectedValue :
debug("'%s' has returned expected value: '%s' after %.1fs." debug("'%s' has returned expected value: '%s' after %.1fs."
%(periodicCheck.__doc__, expectedValue, time.time()-start)) %(periodicCheck.__name__, expectedValue, time.time()-start))
return return
time.sleep(sleep) time.sleep(sleep)
debug("waitForAsertEqual has timed out after %ss." %timeout) debug("waitForAsertEqual has timed out after %ss." %timeout)
@ -250,7 +252,7 @@ def wait_for_app(app = None, timeout = 40, sleep = 1):
pid = None pid = None
start = time.time() start = time.time()
debug("Waiting for '%s' to startup in %ss time" %(app, timeout)) debug("Waiting for '%s' to startup in %.1fs time" %(app, timeout))
while pid == None and time.time() < start + timeout: while pid == None and time.time() < start + timeout:
p = subprocess.Popen(['pgrep', '-n', app], shell = False, p = subprocess.Popen(['pgrep', '-n', app], shell = False,
stdout = subprocess.PIPE, stderr = DEV_NULL) stdout = subprocess.PIPE, stderr = DEV_NULL)

Write Preview
Loading…
Cancel
Save