Best Python code snippet using tempest_python
cloudsim_rest.py
Source:cloudsim_rest.py
1#!/usr/bin/env python32import argparse3import os4import sys5import time6from itertools import cycle7import subprocess8import json9def status_iterator(token, groupid, terminate_str):10 # Loop while starting up.11 t0 = time.time()12 t1 = t013 status_cmd = "curl -X GET -H \"Private-Token: %s\" https://staging-cloudsim-nps.ignitionrobotics.org/1.0/simulations/%s"%(token, groupid)14 status_str = ""15 chcycle = cycle(["-","/","-","\\"])16 while terminate_str not in status_str:17 p = subprocess.Popen(status_cmd, shell=True, executable='/bin/bash' ,stdout=subprocess.PIPE, stderr=subprocess.PIPE)18 status, err = p.communicate()19 #print status20 statusd = json.loads(status)21 newstatus_str = statusd['status']22 if newstatus_str == status_str:23 sys.stdout.write("\r")24 else:25 sys.stdout.write("\n")26 t1 = time.time()27 if "running" in newstatus_str:28 msg = "[%7.1f, %7.1f]\t Status <%s> uri <%s> %s"%(time.time()-t0,29 time.time()-t1,30 newstatus_str,31 statusd['uri'],32 next(chcycle))33 else:34 msg = "[%7.1f, %7.1f]\t Status <%s> %s"%(time.time()-t0,35 time.time()-t1,36 newstatus_str,37 next(chcycle))38 sys.stdout.write(msg)39 sys.stdout.flush()40 41 status_str = newstatus_str42 time.sleep(2)43 44 sys.stdout.write("\n")45 sys.stdout.flush()46 47valid_commands = ['start', 'status', 'stop', 'stopall']48parser = argparse.ArgumentParser(description='Helper script for CloudSim REST calls.')49parser.add_argument('command', type=str, choices=valid_commands)50args = parser.parse_args()51# Image name52#image = "tfoote/test_novnc:main"53#name = "npstest"54image = "learninglab/me4823:matlab_small"55name = "4823matlabsmall"56#image = "learninglab/me4823:main"57#name = "4823main"58# Get token from file59home = os.environ.get("HOME")60token_fname = os.path.join(home,'.cloudsim_token')61if not os.path.isfile(token_fname):62 print("You need to save your cloudsim token as the file <%s>"%token_fname)63 sys.exit(1)64with open(token_fname) as f:65 token = f.readline().strip()66print("Using access token <%s>"%(token))67# File name for storing group ids68gid_fname = os.path.join(home,'.cloudsim_groupid')69if ( (args.command == 'stop') or args.command == 'status'):70 with open(gid_fname) as f:71 for line in f:72 pass73 groupid = line74 if args.command == 'stop':75 print("Stopping group id <%s>"%groupid)76 stop_cmd = "curl -X POST -H \"Private-Token: %s\" https://staging-cloudsim-nps.ignitionrobotics.org/1.0/stop/%s"%(token, groupid)77 p = subprocess.Popen(stop_cmd, shell=True, executable='/bin/bash' ,stdout=subprocess.PIPE, stderr=subprocess.PIPE)78 out, err = p.communicate()79 print(out)80 81 status_iterator(token, groupid, "running")82elif args.command == 'stopall':83 print ("Stopping all images..")84 stopall_cmd = "curl -X POST -H \"Private-Token: %s\" https://staging-cloudsim-nps.ignitionrobotics.org/1.0/stop/all"%token85 print(stopall_cmd)86 p = subprocess.Popen(stopall_cmd, shell=True, executable='/bin/bash' ,stdout=subprocess.PIPE, stderr=subprocess.PIPE)87 out, err = p.communicate()88 print(out)89 if os.path.isfile(gid_fname):90 print("")91 print("Removing file <%s>"%gid_fname)92 os.remove(gid_fname)93 94elif args.command == 'start':95 start_cmd = "curl -X POST -H \"Private-Token: %s\" https://staging-cloudsim-nps.ignitionrobotics.org/1.0/start -F \"image=%s\" -F \"name=%s\""%(token, image, name)96 print("Starting new image")97 print("Starting new instance using image <%s>"%image)98 #print start_cmd99 p = subprocess.Popen(start_cmd, shell=True, executable='/bin/bash' ,stdout=subprocess.PIPE, stderr=subprocess.PIPE)100 out, err = p.communicate()101 print(out)102 outd = json.loads(out)103 success = False104 try:105 groupid = outd['Simulation']['groupid']106 success = True107 except KeyError:108 print("WARNING: Looks ike the simulation didn't start!")109 if success:110 # Save as a one line text file111 with open(gid_fname, 'a') as f:112 f.write(groupid + "\n")113 print("Appended groupid <%s> to <%s>"%(groupid,gid_fname))114 ...
test_run_options.py
Source:test_run_options.py
...89 await wait_for_data(run_ws_client_query, "stdout", "output", "BYE\r\n", 0, "1")90 await wait_for_data(run_ws_client_query, "stopped", "exitCode", 0, 0, "1")91@pytest.mark.asyncio92@pytest.mark.skipif("DISPLAY" not in os.environ, reason="requires X11 display")93async def test_novnc(run_ws_client):94 code = """\95import turtle96turtle.color('red', 'yellow')97from time import sleep98sleep(1) # activity monitor takes 1s to detect the window99"""100 start_cmd = create_message(101 "start",102 {"runner": "python3", "code": code, "novncOptions": {"enabled": True}},103 "1",104 )105 await run_ws_client.send_str(start_cmd)106 await receive_data(run_ws_client, "started", process="1")107 await wait_for_data(run_ws_client, "novnc", timeout=0, process="1")...
test_process_handler.py
Source:test_process_handler.py
...61 await asyncio.sleep(0.2)62 p.on_output.assert_called_with("stdout", "hello\r\nhello\r\n")63 p.on_stop.assert_called_with(0)64@pytest.mark.asyncio65async def test_novnc():66 p = ProcessHandler(user)67 p.on_start = AsyncMock()68 p.on_stop = AsyncMock()69 p.on_output = AsyncMock()70 p.on_display_activity = AsyncMock()71 code = """\72from signal import pause73print('doing fake graphics!')74pause()75"""76 novncOptions = {"enabled": True, "height": 620, "width": 780}77 with patch(78 "further_link.runner.process_handler.async_start", AsyncMock()79 ) as vnc_start:...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!