autotests: notify tests after registerAgent()

This commit is contained in:
Rahul Rahul 2016-07-14 16:43:47 -07:00 committed by Denis Kenzior
parent fc0fd49cbd
commit 6be18cff4e
3 changed files with 34 additions and 13 deletions

View File

@ -7,17 +7,31 @@ import dbus
import time import time
import logging import logging
import os import os
from random import randrange from subprocess import Popen, PIPE, STDOUT
from subprocess import Popen
import sys import sys
sys.path.append('../utility') #needed to import all the utility modules sys.path.append('../utility') #needed to import all the utility modules
import utility import utility
import pty
class TestConnectDisconnect(unittest.TestCase): class TestConnectDisconnect(unittest.TestCase):
def doConnectDisconnect(self): def doConnectDisconnect(self):
objectList = utility.getObjectList(bus) objectList = utility.getObjectList(bus)
networkToConnect = utility.getNetworkToConnectTo(objectList)
# start simpleAgent
master, slave = pty.openpty()
proc = Popen([sys.executable, '../utility/simpleAgent.py'],
stdin=PIPE, stdout=slave, close_fds=True)
stdout_handle = os.fdopen(master)
if stdout_handle.readline().rstrip() == "AGENT_REGISTERED":
logger.debug("Agent Registered")
else:
logger.debug("Agent failed to register")
# close the handles
stdout_handle.close()
os.close(slave)
networkToConnect = utility.getNetworkToConnectTo(objectList)
# check if networkToConnect is not null. If yes, restart program # check if networkToConnect is not null. If yes, restart program
# so that the network list is updated. Alternatively, we can scan # so that the network list is updated. Alternatively, we can scan
# for networks. # for networks.
@ -27,9 +41,6 @@ class TestConnectDisconnect(unittest.TestCase):
os.execl(sys.executable, sys.executable, * sys.argv) os.execl(sys.executable, sys.executable, * sys.argv)
self.assertNotEqual(networkToConnect, "") self.assertNotEqual(networkToConnect, "")
# start simpleAgent
proc = Popen([sys.executable, '../utility/simpleAgent.py'])
time.sleep(2)
network = dbus.Interface(bus.get_object(utility.IWD_SERVICE, network = dbus.Interface(bus.get_object(utility.IWD_SERVICE,
networkToConnect), networkToConnect),
utility.IWD_NETWORK_INTERFACE) utility.IWD_NETWORK_INTERFACE)
@ -69,9 +80,6 @@ class TestConnectDisconnect(unittest.TestCase):
if bus.name_has_owner(utility.IWD_SERVICE) == True: if bus.name_has_owner(utility.IWD_SERVICE) == True:
break break
self.doConnectDisconnect() self.doConnectDisconnect()
#watch doesn't seem to work. So used name_has_owner
#watch = bus.watch_name_owner(utility.IWD_SERVICE,
# self.doConnectDisconnect)
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):

View File

@ -9,11 +9,12 @@ import dbus.mainloop.glib
import time import time
import threading import threading
import logging import logging
from subprocess import Popen from subprocess import Popen, PIPE, STDOUT
import sys import sys
import os import os
sys.path.append('../utility') #needed to import all the utlilty modules sys.path.append('../utility') #needed to import all the utlilty modules
import utility import utility
import pty
def getSecondNetworkToConnect(objectList, firstNetworkName): def getSecondNetworkToConnect(objectList, firstNetworkName):
logger.debug(sys._getframe().f_code.co_name) logger.debug(sys._getframe().f_code.co_name)
@ -32,8 +33,19 @@ class TestTwoNetworks(unittest.TestCase):
# connect to network B. disconnect from network B. # connect to network B. disconnect from network B.
def connectToNetwork(self, networkToConnect): def connectToNetwork(self, networkToConnect):
# start simpleAgent # start simpleAgent
proc = Popen([sys.executable, '../utility/simpleAgent.py']) master, slave = pty.openpty()
time.sleep(2) proc = Popen([sys.executable, '../utility/simpleAgent.py'],
stdin=PIPE, stdout=slave, close_fds=True)
stdout_handle = os.fdopen(master)
if stdout_handle.readline().rstrip() == "AGENT_REGISTERED":
logger.debug("Agent Registered")
else:
logger.debug("Agent failed to register")
# close the handles
stdout_handle.close()
os.close(slave)
network = dbus.Interface(bus.get_object(utility.IWD_SERVICE, network = dbus.Interface(bus.get_object(utility.IWD_SERVICE,
networkToConnect), networkToConnect),
utility.IWD_NETWORK_INTERFACE) utility.IWD_NETWORK_INTERFACE)

View File

@ -9,7 +9,6 @@ import logging
import traceback import traceback
import threading import threading
import time import time
import sys import sys
sys.path.append('../utility') #needed to import all the utlilty modules sys.path.append('../utility') #needed to import all the utlilty modules
import utility import utility
@ -55,6 +54,8 @@ def registerAgent(bus, mainloop):
try: try:
manager.RegisterAgent(pathAgent) manager.RegisterAgent(pathAgent)
logger.debug("Registered iwd agent") logger.debug("Registered iwd agent")
# this will be received by stdout in the parent
print("AGENT_REGISTERED")
except: except:
logger.debug("Error in registering path") logger.debug("Error in registering path")
logger.debug(traceback.print_exc(file=sys.stdout)) logger.debug(traceback.print_exc(file=sys.stdout))