KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use send_ok method in fMBT
Best Python code snippet using fMBT_python
MCLServer.py
Source:MCLServer.py 
...51 self.host = host52 self.port = port53 self.debug = debug54 self.log = log55 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):56 print ("MCLServer:send_ok(self,%s,timeout=%s,error=%s)"57 % (command, timeout, error))58 def set_logFile(self, log):59 self.send_ok('setLog(%s)' % log)60 def simple(self):61 # 1. Set the name of the MCL connection62 self.send_ok('initialize(simple)')63 # 2. Set the name of the MCL ontology to use64 self.send_ok('ontology(simple, basic)')65 # 3. Set the name of the MCL domain66 self.send_ok('configure(simple, simple_domain)')67 # 4. Create location, time, and fuel sensors68 self.send_ok('declareObservableSelf(simple, location, 0)')69 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')70 self.send_ok('declareObservableSelf(simple, fuel, 100)')71 # 5. Set the three sensors' characteristics72 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')73 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')74 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')75 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')76 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')77 # 6. Set some global properties - sensors can fail78 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL,PC_YES)')79 # 7. Create a pair of Expectation Groups80 self.send_ok('declareEG(simple,1)')81 self.send_ok('declareEG(simple,2,1,0)')82 # 8. Create expectation for the plan and action83 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')84 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')85 # 9. Do Monitor86 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')87 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')88 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')89 self.send_ok('EGcomplete(simple, 2, {location=0, elasped_time=3, fuel=35})')90 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')91 # 10. Exit92 self.send_ok('terminate(simple)')93 # 999. All is well94 return True95 def good_sensors(self):96 # 1. Set the name of the MCL connection97 self.send_ok('initialize(simple)')98 # 2. Set the name of the MCL ontology to use99 self.send_ok('ontology(simple, basic)')100 # 3. Set the name of the MCL domain101 self.send_ok('configure(simple, simple_domain)')102 # 4. Create location, time, and fuel sensors103 self.send_ok('declareObservableSelf(simple, location, 0)')104 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')105 self.send_ok('declareObservableSelf(simple, fuel, 100)')106 # 5. Set the three sensors' characteristics107 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')108 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')109 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')110 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')111 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')112 # 6. Set some global properties - sensors can't fail113 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL,PC_NO)')114 # 7. Create a pair of Expectation Groups115 self.send_ok('declareEG(simple,1)')116 self.send_ok('declareEG(simple,2,1,0)')117 # 8. Create expectation for the plan and action118 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')119 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')120 # 9. Do Monitor121 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')122 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')123 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')124 self.send_ok('EGcomplete(simple,2,{location=0, elasped_time=3, fuel=35})')125 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')126 # 999. All is well127 return True128 def no_sensor_diag(self):129 # 1. Set the name of the MCL connection130 self.send_ok('initialize(simple)')131 # 2. Set the name of the MCL ontology to use132 self.send_ok('ontology(simple, basic)')133 # 3. Set the name of the MCL domain134 self.send_ok('configure(simple, simple_domain)')135 # 4. Create location, time, and fuel sensors136 self.send_ok('declareObservableSelf(simple, location, 0)')137 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')138 self.send_ok('declareObservableSelf(simple, fuel, 100)')139 # 5. Set the three sensors' characteristics140 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')141 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')142 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')143 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')144 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')145 # 6. Set some global properties - sensors can't fail146 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL, PC_NO)')147 self.send_ok('setPropertyDefault(simple, CRC_SENSOR_DIAG, PC_NO)')148 # 7. Create a pair of Expectation Groups149 self.send_ok('declareEG(simple,1)')150 self.send_ok('declareEG(simple,2,1,0)')151 # 8. Create expectation for the plan and action152 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')153 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')154 # 9. Do Monitor155 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')156 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')157 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')158 self.send_ok('EGcomplete(simple,2,{location=0, elasped_time=3, fuel=35})')159 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')160 print "hellow!"161 # 999. All is well162 return True163# ======================================================164# MCLServerTCP165# ======================================================166class MCLServerTCP(MCLServer):167 def __init__(self, host=MCL_HOST,168 port=MCL_PORT,169 debug=MCL_DEBUG,170 log=MCL_LOG):171 # 1. Initialize our parent172 MCLServer.__init__(self, host=host,173 port=port,174 debug=debug,175 log=log)176 # 2. Open telnet port177 self.telnet = telnetlib.Telnet(host, port)178 # 3. Set debugging level179 if debug and self.telnet:180 self.telnet.set_debuglevel(debug)181 # 4. Last command and result182 self.command = None183 self.response = None184 # 5. If logging to file, start it up185 if log:186 self.set_logFile(log)187 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):188 if not self.telnet:189 return False190 self.command = command191 self.telnet.write(command)192 self.response = self.telnet.read_until('\n', timeout).strip()193 if not self.response.startswith('ok'):194 if error:195 raise Exception('%s --> %s' %196 (self.command, self.response))197 return False198 return True199 def monitor(self, command, timeout=MCL_TIMEOUT, error=True):200 self.send_ok(command, timeout=timeout, error=error)201 return MCLResponse.MCLResponse.decode_monitor_response(self.response[3:-1])202 def set_debuglevel(self, level):203 if self.telnet:204 self.telnet.set_debuglevel(level)205 def close(self):206 if self.telnet:207 try:208 self.set_logFile('stdout')209 except Exception:210 pass211 self.telnet.close()212 self.telnet = None213# ======================================================214# MCLServerFake215# ======================================================216class MCLServerFake(MCLServer):217 def __init__(self, host=MCL_HOST,218 port=MCL_FAKE,219 debug=MCL_DEBUG,220 log=MCL_LOG):221 # 1. Initialize our parent222 MCLServer.__init__(self, host=host,223 port=port,224 debug=debug,225 log=log)226 # 2. Last command and result227 self.command = None228 self.response = None229 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):230 self.command = command231 self.response = 'ok()'232 if timeout: pass233 if error: pass234 def monitor(self, command, timeout=MCL_TIMEOUT, error=True):235 self.command = command236 self.response = 'ok([])'237 if timeout: pass238 if error: pass239 return []240 def set_debuglevel(self, level):241 self.debug = level242 def close(self):243 pass244# ======================================================245# start_mcl_server246# ======================================================247def start_mcl_server(name=SERVER_NAME,248 host=MCL_HOST, port=MCL_PORT,249 debug=MCL_DEBUG, verbose=False,250 verify=False):251 # 1. Fake servers are always started252 if port == MCL_FAKE:253 return True254 # 2. Stop it if we think it may be running255 if port in SERVERS:256 kill_off_server(SERVERS[port])257 del SERVERS[port]258 # 3. Start a real server and record the pid id259 pid = os.spawnlp(os.P_NOWAIT,260 SERVER_NAME,261 SERVER_NAME,262 SERVER_OPT,263 str(port))264 SERVERS[port] = pid265 time.sleep(SERVER_WAIT)266 # 4. Verify that it is started (if requested)267 if verify:268 if verbose:269 print ("%s: connecting to MCL at %s port %s"270 % (name, host, port))271 try:272 mcl = MCLServerTCP(host=host, port=port, debug=debug)273 # 5. Return False if it is not274 except Exception, e:275 print ("%s: unable to connect to MCL at %s port %s"276 % (name, host, port))277 print e278 kill_off_server(pid)279 return False280 # 6. Return True if it is281 mcl.close()282 # 7. Return success283 return True284# ======================================================285# stop_mcl_server286# ======================================================287def stop_mcl_server(name=SERVER_NAME,288 host=MCL_HOST, port=MCL_PORT,289 debug=MCL_DEBUG, verbose=False):290 # 1. Fake servers need no stopping291 if port == MCL_FAKE:292 return True293 # 2. Connect to the server294 if verbose:295 print ("%s: connecting to MCL at %s port %s"296 % (name, host, port))297 try:298 mcl = MCLServerTCP(host=host, port=port, debug=debug)299 # 3. Return False if we can't connect300 except Exception, e:301 print ("%s: unable to connect to MCL at %s port %s"302 % (name, host, port))303 print e304 if port in SERVERS:305 kill_off_server(SERVERS[port])306 del SERVERS[port]307 return False308 # 4. Send a terminate request309 mcl.send_ok('initialize(%s)' % name)310 mcl.send_ok('terminateWithExtremePrejudice(%s)' % name)311 mcl.close()312 time.sleep(KILL_WAIT)313 # 5. We ask nicely only once314 if port in SERVERS:315 kill_off_server(SERVERS[port])316 del SERVERS[port]317 # 6. Return success318 return True319# ======================================================320# kill_off_server321# ======================================================322def kill_off_server(process_id):323 # 1. Ask nicely324 try:325 os.kill(process_id, signal.SIGHUP)326 except OSError:327 return True328 # 2. Give it a chance329 time.sleep(KILL_WAIT)330 # 3. Really make it go away331 try:332 os.kill(process_id, signal.SIGKILL)333 except OSError:334 return True335 # 4. Give it a chance336 time.sleep(KILL_WAIT)337 # 5. Really, really make it go away338 for _ in range(KILL_RETRY):339 try:340 os.kill(process_id, signal.SIGKILL)341 except OSError:342 return True343 time.sleep(KILL_WAIT)344 # 6. We gave it our best shot345 return False346# ======================================================347# experiment348# ======================================================349def experiment(name, host=MCL_HOST, port=MCL_PORT,350 debug=MCL_DEBUG, verbose=True):351 # 1. Connect to the MCL server and set debuging level352 if verbose:353 print ("%s: connecting to MCL at %s port %s"354 % (name, host, port))355 try:356 if port == MCL_FAKE:357 mcl = MCLServerFake(host=host, port=port,358 debug=debug)359 else:360 mcl = MCLServerTCP(host=host, port=port,361 debug=debug)362 except Exception, e:363 print ("%s: unable to connect to MCL at %s port %s"364 % (name, host, port))365 print e366 return False367 # 2. Carry out the experiment368 result = False369 try:370 if callable(getattr(mcl,name)):371 result = getattr(mcl,name)()372 # 3. Close the connection373 finally:374 mcl.close()375 # 4. Return the experiment's result376 print "oy"377 print result378 return result379# ======================================================380# test_mcl_server381# ======================================================382def test_mcl_server(host=MCL_HOST, port=MCL_FAKE,383 debug=MCL_DEBUG, verbose=False):384 experiment('simple', host=host, port=port,385 debug=debug, verbose=verbose)386 experiment('good_sensors', host=host, port=port,387 debug=debug, verbose=verbose)388 experiment('no_sensor_diag', host=host, port=port,389 debug=debug, verbose=verbose)390# ======================================================391# unittest392# ======================================================393class Test_MCLServer(unittest.TestCase):394 def test_MCLServer_empty(self):395 s = MCLServer()396 self.assertNotEqual(s, None)397# ======================================================398# parseCommandLine399# ======================================================400def parseCommandLine():401 # 1. Create the command line parser402 parser = OptionParser()403 parser.add_option("-u", "--unittest", default=False,404 action="store_true", dest="unittest",405 help="Execute unittests")406 parser.add_option("-v", "--verbose", default=False,407 action="store_true", dest="verbose",408 help="Print status messages to stdout")409 parser.add_option("-o", "--port", default=MCL_PORT,410 action="store", dest="port",411 help="MCL server port number")412 parser.add_option("-d", "--debug", default=MCL_DEBUG,413 action="count", dest="debug",414 help="Telnet debug level: default=%s" % MCL_DEBUG)415 # 2. Parse the command line416 options, args = parser.parse_args()417 options.explain = []418 # 3. Validate mcl port419 try:420 options.port = int(options.port)421 except ValueError:422 parser.error('Invalid MCL port number')423 # 4. Check the arguments (if any)424 if args:425 parser.error('No arguments allowed')426 # 5. Return the options427 return options428#=============================================429#in-between-parser430#=============================================431MCL_NAME = "test_name"432def send_command(server, command, params):433 params.insert(0, MCL_NAME)434 str = command + '(' + string.join(params, ',') + ')'435 436 if(server.send_ok(str)):437 print str + ' executed successfully'438 else:439 print 'error executing ' + str440 441def execute_commands(saveFile = None, autoConfig = True):442 try:443 mcl = MCLServerTCP(host=MCL_HOST, port=MCL_PORT, debug=MCL_DEBUG)444 except Exception, e:445 print ("%s: unable to connect to MCL at %s port %s"446 % (name, host, port))447 print e448 return False449 450 try:...
TestServer.py
Source:TestServer.py
...51 self.host = host52 self.port = port53 self.debug = debug54 self.log = log55 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):56 print ("MCLServer:send_ok(self,%s,timeout=%s,error=%s)"57 % (command, timeout, error))58 def set_logFile(self, log):59 self.send_ok('setLog(%s)' % log)60 def simple(self):61 # 1. Set the name of the MCL connection62 self.send_ok('initialize(simple)')63 # 2. Set the name of the MCL ontology to use64 self.send_ok('ontology(simple, basic)')65 # 3. Set the name of the MCL domain66 self.send_ok('configure(simple, simple_domain)')67 # 4. Create location, time, and fuel sensors68 self.send_ok('declareObservableSelf(simple, location, 0)')69 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')70 self.send_ok('declareObservableSelf(simple, fuel, 100)')71 # 5. Set the three sensors' characteristics72 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')73 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')74 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')75 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')76 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')77 # 6. Set some global properties - sensors can fail78 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL,PC_YES)')79 # 7. Create a pair of Expectation Groups80 self.send_ok('declareEG(simple,1)')81 self.send_ok('declareEG(simple,2,1,0)')82 # 8. Create expectation for the plan and action83 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')84 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')85 # 9. Do Monitor86 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')87 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')88 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')89 self.send_ok('EGcomplete(simple, 2, {location=0, elasped_time=3, fuel=35})')90 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')91 # 10. Exit92 self.send_ok('terminate(simple)')93 # 999. All is well94 return True95 def good_sensors(self):96 # 1. Set the name of the MCL connection97 self.send_ok('initialize(simple)')98 # 2. Set the name of the MCL ontology to use99 self.send_ok('ontology(simple, basic)')100 # 3. Set the name of the MCL domain101 self.send_ok('configure(simple, simple_domain)')102 # 4. Create location, time, and fuel sensors103 self.send_ok('declareObservableSelf(simple, location, 0)')104 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')105 self.send_ok('declareObservableSelf(simple, fuel, 100)')106 # 5. Set the three sensors' characteristics107 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')108 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')109 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')110 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')111 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')112 # 6. Set some global properties - sensors can't fail113 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL,PC_NO)')114 # 7. Create a pair of Expectation Groups115 self.send_ok('declareEG(simple,1)')116 self.send_ok('declareEG(simple,2,1,0)')117 # 8. Create expectation for the plan and action118 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')119 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')120 # 9. Do Monitor121 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')122 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')123 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')124 self.send_ok('EGcomplete(simple,2,{location=0, elasped_time=3, fuel=35})')125 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')126 # 999. All is well127 return True128 def no_sensor_diag(self):129 # 1. Set the name of the MCL connection130 self.send_ok('initialize(simple)')131 # 2. Set the name of the MCL ontology to use132 self.send_ok('ontology(simple, basic)')133 # 3. Set the name of the MCL domain134 self.send_ok('configure(simple, simple_domain)')135 # 4. Create location, time, and fuel sensors136 self.send_ok('declareObservableSelf(simple, location, 0)')137 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')138 self.send_ok('declareObservableSelf(simple, fuel, 100)')139 # 5. Set the three sensors' characteristics140 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')141 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')142 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')143 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')144 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')145 # 6. Set some global properties - sensors can't fail146 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL, PC_NO)')147 self.send_ok('setPropertyDefault(simple, CRC_SENSOR_DIAG, PC_NO)')148 # 7. Create a pair of Expectation Groups149 self.send_ok('declareEG(simple,1)')150 self.send_ok('declareEG(simple,2,1,0)')151 # 8. Create expectation for the plan and action152 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')153 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')154 # 9. Do Monitor155 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')156 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')157 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')158 self.send_ok('EGcomplete(simple,2,{location=0, elasped_time=3, fuel=35})')159 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')160 # 999. All is well161 return True162# ======================================================163# MCLServerTCP164# ======================================================165class MCLServerTCP(MCLServer):166 def __init__(self, host=MCL_HOST,167 port=MCL_PORT,168 debug=MCL_DEBUG,169 log=MCL_LOG):170 # 1. Initialize our parent171 MCLServer.__init__(self, host=host,172 port=port,173 debug=debug,174 log=log)175 # 2. Open telnet port176 self.telnet = telnetlib.Telnet(host, port)177 # 3. Set debugging level178 if debug and self.telnet:179 self.telnet.set_debuglevel(debug)180 # 4. Last command and result181 self.command = None182 self.response = None183 # 5. If logging to file, start it up184 if log:185 self.set_logFile(log)186 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):187 if not self.telnet:188 return False189 self.command = command190 self.telnet.write(command)191 self.response = self.telnet.read_until('\n', timeout).strip()192 if not self.response.startswith('ok'):193 if error:194 raise Exception('%s --> %s' %195 (self.command, self.response))196 return False197 return True198 def monitor(self, command, timeout=MCL_TIMEOUT, error=True):199 self.send_ok(command, timeout=timeout, error=error)200 return MCLResponse.MCLResponse.decode_monitor_response(self.response[3:-1])201 def set_debuglevel(self, level):202 if self.telnet:203 self.telnet.set_debuglevel(level)204 def close(self):205 if self.telnet:206 try:207 self.set_logFile('stdout')208 except Exception:209 pass210 self.telnet.close()211 self.telnet = None212# ======================================================213# MCLServerFake214# ======================================================215class MCLServerFake(MCLServer):216 def __init__(self, host=MCL_HOST,217 port=MCL_FAKE,218 debug=MCL_DEBUG,219 log=MCL_LOG):220 # 1. Initialize our parent221 MCLServer.__init__(self, host=host,222 port=port,223 debug=debug,224 log=log)225 # 2. Last command and result226 self.command = None227 self.response = None228 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):229 self.command = command230 self.response = 'ok()'231 if timeout: pass232 if error: pass233 def monitor(self, command, timeout=MCL_TIMEOUT, error=True):234 self.command = command235 self.response = 'ok([])'236 if timeout: pass237 if error: pass238 return []239 def set_debuglevel(self, level):240 self.debug = level241 def close(self):242 pass243# ======================================================244# start_mcl_server245# ======================================================246def start_mcl_server(name=SERVER_NAME,247 host=MCL_HOST, port=MCL_PORT,248 debug=MCL_DEBUG, verbose=False,249 verify=False):250 # 1. Fake servers are always started251 if port == MCL_FAKE:252 return True253 # 2. Stop it if we think it may be running254 if port in SERVERS:255 kill_off_server(SERVERS[port])256 del SERVERS[port]257 # 3. Start a real server and record the pid id258 pid = os.spawnlp(os.P_NOWAIT,259 SERVER_NAME,260 SERVER_NAME,261 SERVER_OPT,262 str(port))263 SERVERS[port] = pid264 time.sleep(SERVER_WAIT)265 # 4. Verify that it is started (if requested)266 if verify:267 if verbose:268 print ("%s: connecting to MCL at %s port %s"269 % (name, host, port))270 try:271 mcl = MCLServerTCP(host=host, port=port, debug=debug)272 # 5. Return False if it is not273 except Exception, e:274 print ("%s: unable to connect to MCL at %s port %s"275 % (name, host, port))276 print e277 kill_off_server(pid)278 return False279 # 6. Return True if it is280 mcl.close()281 # 7. Return success282 return True283# ======================================================284# stop_mcl_server285# ======================================================286def stop_mcl_server(name=SERVER_NAME,287 host=MCL_HOST, port=MCL_PORT,288 debug=MCL_DEBUG, verbose=False):289 # 1. Fake servers need no stopping290 if port == MCL_FAKE:291 return True292 # 2. Connect to the server293 if verbose:294 print ("%s: connecting to MCL at %s port %s"295 % (name, host, port))296 try:297 mcl = MCLServerTCP(host=host, port=port, debug=debug)298 # 3. Return False if we can't connect299 except Exception, e:300 print ("%s: unable to connect to MCL at %s port %s"301 % (name, host, port))302 print e303 if port in SERVERS:304 kill_off_server(SERVERS[port])305 del SERVERS[port]306 return False307 # 4. Send a terminate request308 mcl.send_ok('initialize(%s)' % name)309 mcl.send_ok('terminateWithExtremePrejudice(%s)' % name)310 mcl.close()311 time.sleep(KILL_WAIT)312 # 5. We ask nicely only once313 if port in SERVERS:314 kill_off_server(SERVERS[port])315 del SERVERS[port]316 # 6. Return success317 return True318# ======================================================319# kill_off_server320# ======================================================321def kill_off_server(process_id):322 # 1. Ask nicely323 try:324 os.kill(process_id, signal.SIGHUP)325 except OSError:326 return True327 # 2. Give it a chance328 time.sleep(KILL_WAIT)329 # 3. Really make it go away330 try:331 os.kill(process_id, signal.SIGKILL)332 except OSError:333 return True334 # 4. Give it a chance335 time.sleep(KILL_WAIT)336 # 5. Really, really make it go away337 for _ in range(KILL_RETRY):338 try:339 os.kill(process_id, signal.SIGKILL)340 except OSError:341 return True342 time.sleep(KILL_WAIT)343 # 6. We gave it our best shot344 return False345# ======================================================346# experiment347# ======================================================348def experiment(name, host=MCL_HOST, port=MCL_PORT,349 debug=MCL_DEBUG, verbose=False):350 # 1. Connect to the MCL server and set debuging level351 if verbose:352 print ("%s: connecting to MCL at %s port %s"353 % (name, host, port))354 try:355 if port == MCL_FAKE:356 mcl = MCLServerFake(host=host, port=port,357 debug=debug)358 else:359 mcl = MCLServerTCP(host=host, port=port,360 debug=debug)361 except Exception, e:362 print ("%s: unable to connect to MCL at %s port %s"363 % (name, host, port))364 print e365 return False366 # 2. Carry out the experiment367 result = False368 try:369 if callable(getattr(mcl,name)):370 result = getattr(mcl,name)()371 # 3. Close the connection372 finally:373 mcl.close()374 # 4. Return the experiment's result375 return result376# ======================================================377# test_mcl_server378# ======================================================379def test_mcl_server(host=MCL_HOST, port=MCL_FAKE,380 debug=MCL_DEBUG, verbose=False):381 experiment('simple', host=host, port=port,382 debug=debug, verbose=verbose)383 experiment('good_sensors', host=host, port=port,384 debug=debug, verbose=verbose)385 experiment('no_sensor_diag', host=host, port=port,386 debug=debug, verbose=verbose)387# ======================================================388# unittest389# ======================================================390class Test_MCLServer(unittest.TestCase):391 def test_MCLServer_empty(self):392 s = MCLServer()393 self.assertNotEqual(s, None)394# ======================================================395# parseCommandLine396# ======================================================397def parseCommandLine():398 # 1. Create the command line parser399 parser = OptionParser()400 parser.add_option("-u", "--unittest", default=False,401 action="store_true", dest="unittest",402 help="Execute unittests")403 parser.add_option("-v", "--verbose", default=False,404 action="store_true", dest="verbose",405 help="Print status messages to stdout")406 parser.add_option("-o", "--port", default=MCL_PORT,407 action="store", dest="port",408 help="MCL server port number")409 parser.add_option("-d", "--debug", default=MCL_DEBUG,410 action="count", dest="debug",411 help="Telnet debug level: default=%s" % MCL_DEBUG)412 # 2. Parse the command line413 options, args = parser.parse_args()414 options.explain = []415 # 3. Validate mcl port416 try:417 options.port = int(options.port)418 except ValueError:419 parser.error('Invalid MCL port number')420 # 4. Check the arguments (if any)421 if args:422 parser.error('No arguments allowed')423 # 5. Return the options424 return options425#=============================================426#in-between-parser427#=============================================428MCL_NAME = "test_name"429def send_command(server, command, params):430 params.insert(0, MCL_NAME)431 str = command + '(' + string.join(params, ',') + ')'432 433 if(server.send_ok(str)):434 print str + ' executed successfully'435 else:436 print 'error executing ' + str437 438def execute_commands(saveFile = None, autoConfig = True):439 try:440 mcl = MCLServerTCP(host=MCL_HOST, port=MCL_PORT, debug=MCL_DEBUG)441 except Exception, e:442 print ("%s: unable to connect to MCL at %s port %s"443 % (MCL_NAME, MCL_HOST, MCL_PORT))444 return False445 446 try:447 print "oy"...
MCLServer.py~
Source:MCLServer.py~
...51 self.host = host52 self.port = port53 self.debug = debug54 self.log = log55 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):56 print ("MCLServer:send_ok(self,%s,timeout=%s,error=%s)"57 % (command, timeout, error))58 def set_logFile(self, log):59 self.send_ok('setLog(%s)' % log)60 def simple(self):61 # 1. Set the name of the MCL connection62 self.send_ok('initialize(simple)')63 # 2. Set the name of the MCL ontology to use64 self.send_ok('ontology(simple, basic)')65 # 3. Set the name of the MCL domain66 self.send_ok('configure(simple, simple_domain)')67 # 4. Create location, time, and fuel sensors68 self.send_ok('declareObservableSelf(simple, location, 0)')69 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')70 self.send_ok('declareObservableSelf(simple, fuel, 100)')71 # 5. Set the three sensors' characteristics72 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')73 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')74 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')75 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')76 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')77 # 6. Set some global properties - sensors can fail78 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL,PC_YES)')79 # 7. Create a pair of Expectation Groups80 self.send_ok('declareEG(simple,1)')81 self.send_ok('declareEG(simple,2,1,0)')82 # 8. Create expectation for the plan and action83 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')84 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')85 # 9. Do Monitor86 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')87 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')88 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')89 self.send_ok('EGcomplete(simple, 2, {location=0, elasped_time=3, fuel=35})')90 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')91 # 10. Exit92 self.send_ok('terminate(simple)')93 # 999. All is well94 return True95 def good_sensors(self):96 # 1. Set the name of the MCL connection97 self.send_ok('initialize(simple)')98 # 2. Set the name of the MCL ontology to use99 self.send_ok('ontology(simple, basic)')100 # 3. Set the name of the MCL domain101 self.send_ok('configure(simple, simple_domain)')102 # 4. Create location, time, and fuel sensors103 self.send_ok('declareObservableSelf(simple, location, 0)')104 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')105 self.send_ok('declareObservableSelf(simple, fuel, 100)')106 # 5. Set the three sensors' characteristics107 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')108 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')109 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')110 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')111 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')112 # 6. Set some global properties - sensors can't fail113 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL,PC_NO)')114 # 7. Create a pair of Expectation Groups115 self.send_ok('declareEG(simple,1)')116 self.send_ok('declareEG(simple,2,1,0)')117 # 8. Create expectation for the plan and action118 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')119 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')120 # 9. Do Monitor121 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')122 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')123 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')124 self.send_ok('EGcomplete(simple,2,{location=0, elasped_time=3, fuel=35})')125 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')126 # 999. All is well127 return True128 def no_sensor_diag(self):129 # 1. Set the name of the MCL connection130 self.send_ok('initialize(simple)')131 # 2. Set the name of the MCL ontology to use132 self.send_ok('ontology(simple, basic)')133 # 3. Set the name of the MCL domain134 self.send_ok('configure(simple, simple_domain)')135 # 4. Create location, time, and fuel sensors136 self.send_ok('declareObservableSelf(simple, location, 0)')137 self.send_ok('declareObservableSelf(simple, elasped_time, 0)')138 self.send_ok('declareObservableSelf(simple, fuel, 100)')139 # 5. Set the three sensors' characteristics140 self.send_ok('setObsPropSelf(simple, location, PROP_SCLASS, SC_SPATIAL)')141 self.send_ok('setObsLegalRangeSelf(simple, location, 0, 100)')142 self.send_ok('setObsPropSelf(simple, elasped_time, PROP_SCLASS, SC_TEMPORAL)')143 self.send_ok('setObsPropSelf(simple, fuel, PROP_SCLASS, SC_RESOURCE)')144 self.send_ok('setObsLegalRangeSelf(simple, fuel, 0, 40)')145 # 6. Set some global properties - sensors can't fail146 self.send_ok('setPropertyDefault(simple, PCI_SENSORS_CAN_FAIL, PC_NO)')147 self.send_ok('setPropertyDefault(simple, CRC_SENSOR_DIAG, PC_NO)')148 # 7. Create a pair of Expectation Groups149 self.send_ok('declareEG(simple,1)')150 self.send_ok('declareEG(simple,2,1,0)')151 # 8. Create expectation for the plan and action152 self.send_ok('declareSelfExp(simple,1,location,ec_take_value,3)')153 self.send_ok('declareSelfExp(simple,2,location,ec_take_value,1)')154 # 9. Do Monitor155 self.send_ok('monitor(simple,{location=0, elasped_time=0, fuel=40})')156 self.send_ok('monitor(simple,{location=0, elasped_time=1, fuel=50})')157 self.send_ok('monitor(simple,{location=0, elasped_time=2, fuel=40})')158 self.send_ok('EGcomplete(simple,2,{location=0, elasped_time=3, fuel=35})')159 self.send_ok('monitor(simple,{location=0, elasped_time=4, fuel=35})')160 # 999. All is well161 return True162# ======================================================163# MCLServerTCP164# ======================================================165class MCLServerTCP(MCLServer):166 def __init__(self, host=MCL_HOST,167 port=MCL_PORT,168 debug=MCL_DEBUG,169 log=MCL_LOG):170 # 1. Initialize our parent171 MCLServer.__init__(self, host=host,172 port=port,173 debug=debug,174 log=log)175 # 2. Open telnet port176 self.telnet = telnetlib.Telnet(host, port)177 # 3. Set debugging level178 if debug and self.telnet:179 self.telnet.set_debuglevel(debug)180 # 4. Last command and result181 self.command = None182 self.response = None183 # 5. If logging to file, start it up184 if log:185 self.set_logFile(log)186 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):187 if not self.telnet:188 return False189 self.command = command190 self.telnet.write(command)191 self.response = self.telnet.read_until('\n', timeout).strip()192 if not self.response.startswith('ok'):193 if error:194 raise Exception('%s --> %s' %195 (self.command, self.response))196 return False197 return True198 def monitor(self, command, timeout=MCL_TIMEOUT, error=True):199 self.send_ok(command, timeout=timeout, error=error)200 return MCLResponse.MCLResponse.decode_monitor_response(self.response[3:-1])201 def set_debuglevel(self, level):202 if self.telnet:203 self.telnet.set_debuglevel(level)204 def close(self):205 if self.telnet:206 try:207 self.set_logFile('stdout')208 except Exception:209 pass210 self.telnet.close()211 self.telnet = None212# ======================================================213# MCLServerFake214# ======================================================215class MCLServerFake(MCLServer):216 def __init__(self, host=MCL_HOST,217 port=MCL_FAKE,218 debug=MCL_DEBUG,219 log=MCL_LOG):220 # 1. Initialize our parent221 MCLServer.__init__(self, host=host,222 port=port,223 debug=debug,224 log=log)225 # 2. Last command and result226 self.command = None227 self.response = None228 def send_ok(self, command, timeout=MCL_TIMEOUT, error=True):229 self.command = command230 self.response = 'ok()'231 if timeout: pass232 if error: pass233 def monitor(self, command, timeout=MCL_TIMEOUT, error=True):234 self.command = command235 self.response = 'ok([])'236 if timeout: pass237 if error: pass238 return []239 def set_debuglevel(self, level):240 self.debug = level241 def close(self):242 pass243# ======================================================244# start_mcl_server245# ======================================================246def start_mcl_server(name=SERVER_NAME,247 host=MCL_HOST, port=MCL_PORT,248 debug=MCL_DEBUG, verbose=False,249 verify=False):250 # 1. Fake servers are always started251 if port == MCL_FAKE:252 return True253 # 2. Stop it if we think it may be running254 if port in SERVERS:255 kill_off_server(SERVERS[port])256 del SERVERS[port]257 # 3. Start a real server and record the pid id258 pid = os.spawnlp(os.P_NOWAIT,259 SERVER_NAME,260 SERVER_NAME,261 SERVER_OPT,262 str(port))263 SERVERS[port] = pid264 time.sleep(SERVER_WAIT)265 # 4. Verify that it is started (if requested)266 if verify:267 if verbose:268 print ("%s: connecting to MCL at %s port %s"269 % (name, host, port))270 try:271 mcl = MCLServerTCP(host=host, port=port, debug=debug)272 # 5. Return False if it is not273 except Exception, e:274 print ("%s: unable to connect to MCL at %s port %s"275 % (name, host, port))276 print e277 kill_off_server(pid)278 return False279 # 6. Return True if it is280 mcl.close()281 # 7. Return success282 return True283# ======================================================284# stop_mcl_server285# ======================================================286def stop_mcl_server(name=SERVER_NAME,287 host=MCL_HOST, port=MCL_PORT,288 debug=MCL_DEBUG, verbose=False):289 # 1. Fake servers need no stopping290 if port == MCL_FAKE:291 return True292 # 2. Connect to the server293 if verbose:294 print ("%s: connecting to MCL at %s port %s"295 % (name, host, port))296 try:297 mcl = MCLServerTCP(host=host, port=port, debug=debug)298 # 3. Return False if we can't connect299 except Exception, e:300 print ("%s: unable to connect to MCL at %s port %s"301 % (name, host, port))302 print e303 if port in SERVERS:304 kill_off_server(SERVERS[port])305 del SERVERS[port]306 return False307 # 4. Send a terminate request308 mcl.send_ok('initialize(%s)' % name)309 mcl.send_ok('terminateWithExtremePrejudice(%s)' % name)310 mcl.close()311 time.sleep(KILL_WAIT)312 # 5. We ask nicely only once313 if port in SERVERS:314 kill_off_server(SERVERS[port])315 del SERVERS[port]316 # 6. Return success317 return True318# ======================================================319# kill_off_server320# ======================================================321def kill_off_server(process_id):322 # 1. Ask nicely323 try:324 os.kill(process_id, signal.SIGHUP)325 except OSError:326 return True327 # 2. Give it a chance328 time.sleep(KILL_WAIT)329 # 3. Really make it go away330 try:331 os.kill(process_id, signal.SIGKILL)332 except OSError:333 return True334 # 4. Give it a chance335 time.sleep(KILL_WAIT)336 # 5. Really, really make it go away337 for _ in range(KILL_RETRY):338 try:339 os.kill(process_id, signal.SIGKILL)340 except OSError:341 return True342 time.sleep(KILL_WAIT)343 # 6. We gave it our best shot344 return False345# ======================================================346# experiment347# ======================================================348def experiment(name, host=MCL_HOST, port=MCL_PORT,349 debug=MCL_DEBUG, verbose=False):350 # 1. Connect to the MCL server and set debuging level351 if verbose:352 print ("%s: connecting to MCL at %s port %s"353 % (name, host, port))354 try:355 if port == MCL_FAKE:356 mcl = MCLServerFake(host=host, port=port,357 debug=debug)358 else:359 mcl = MCLServerTCP(host=host, port=port,360 debug=debug)361 except Exception, e:362 print ("%s: unable to connect to MCL at %s port %s"363 % (name, host, port))364 print e365 return False366 # 2. Carry out the experiment367 result = False368 try:369 if callable(getattr(mcl,name)):370 result = getattr(mcl,name)()371 # 3. Close the connection372 finally:373 mcl.close()374 # 4. Return the experiment's result375 return result376# ======================================================377# test_mcl_server378# ======================================================379def test_mcl_server(host=MCL_HOST, port=MCL_FAKE,380 debug=MCL_DEBUG, verbose=False):381 experiment('simple', host=host, port=port,382 debug=debug, verbose=verbose)383 experiment('good_sensors', host=host, port=port,384 debug=debug, verbose=verbose)385 experiment('no_sensor_diag', host=host, port=port,386 debug=debug, verbose=verbose)387# ======================================================388# unittest389# ======================================================390class Test_MCLServer(unittest.TestCase):391 def test_MCLServer_empty(self):392 s = MCLServer()393 self.assertNotEqual(s, None)394# ======================================================395# parseCommandLine396# ======================================================397def parseCommandLine():398 # 1. Create the command line parser399 parser = OptionParser()400 parser.add_option("-u", "--unittest", default=False,401 action="store_true", dest="unittest",402 help="Execute unittests")403 parser.add_option("-v", "--verbose", default=False,404 action="store_true", dest="verbose",405 help="Print status messages to stdout")406 parser.add_option("-o", "--port", default=MCL_PORT,407 action="store", dest="port",408 help="MCL server port number")409 parser.add_option("-d", "--debug", default=MCL_DEBUG,410 action="count", dest="debug",411 help="Telnet debug level: default=%s" % MCL_DEBUG)412 # 2. Parse the command line413 options, args = parser.parse_args()414 options.explain = []415 # 3. Validate mcl port416 try:417 options.port = int(options.port)418 except ValueError:419 parser.error('Invalid MCL port number')420 # 4. Check the arguments (if any)421 if args:422 parser.error('No arguments allowed')423 # 5. Return the options424 return options425#=============================================426#in-between-parser427#=============================================428MCL_NAME = "test_name"429def send_command(server, command, params):430 params.insert(0, MCL_NAME)431 str = command + '(' + string.join(params, ',') + ')'432 433 if(server.send_ok(str)):434 print str + ' executed successfully'435 else:436 print 'error executing ' + str437 438def execute_commands(saveFile = None, autoConfig = True):439 try:440 mcl = MCLServerTCP(host=MCL_HOST, port=MCL_PORT, debug=MCL_DEBUG)441 except Exception, e:442 print ("%s: unable to connect to MCL at %s port %s"443 % (name, host, port))444 print e445 return False446 447 try:...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run fMBT automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
