How to use test_novnc method in tempest

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:...

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.