KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use proc_stop method in yandex-tank

Best Python code snippet using yandex-tank

utils.py

Source:utils.py

1import http.client, urllib.request, urllib.parse, urllib.error, base642import json3from flask_restful import abort4from typing import List5from config import DE_LIJN_API_KEY, WEATHER_API_KEY, TOMTOM_API_KEY6headers = {7    'Ocp-Apim-Subscription-Key': DE_LIJN_API_KEY8}9def make_lijn_request(request_type, url, params=None):10    """11    Makes a request to the De Lijn API.12    :param request_type: The type of the request, usually 'GET'13    :param url: The url of the request14    :param params: The additional parameters of the request15    :return: a tuple of a json object and a HTTP status code16    """17    try:18        # Protection against stupid self19        if url[0] != '/':20            url = '/' + url21        # Create connection22        conn = http.client.HTTPSConnection('api.delijn.be')23        # Make request24        if params is None:25            conn.request(request_type, url, "{body}", headers)26        else:27            full_url = url + '?{}'.format(params)  # Format parameters28            conn.request(request_type, full_url, "{body}", headers)29        resp = conn.getresponse()30        status = int(resp.getcode())31        data = resp.read()32        conn.close()33        return json.loads(data), status34    except Exception as e:35        print("Error :'(", e.__str__())36def make_weather_request(lat, lon):37    """38    Makes a request for the OpenWeather API.39    :param lat: latitude of the location to get the weather of. This is a positive float40    :param lon: longitude of the location to get the weather of. This is a positive float41    :return: a json object and an HTTP status code42    """43    try:44        conn = http.client.HTTPSConnection("api.openweathermap.org")45        url = "/data/2.5/weather?lat={}&lon={}&appid={}&units=metric".format(lat, lon, WEATHER_API_KEY)46        conn.request("GET", url)47        resp = conn.getresponse()48        data = resp.read()49        status = int(resp.getcode())50        conn.close()51        return json.loads(data), status52    except Exception as e:53        print("Error in fetching weather data", e.__str__())54def make_tomtom_request(lat1, lon1, lat2, lon2):55    """56    Make a request for the TomTom Routing API57    :param lat1: latitude of first point of route58    :param lon1: longitude of first point of route59    :param lat2: latitude of second point of route60    :param lon2: longitude of second point of route61    :return: a tuple of a json object and a HTTP status code62    """63    try:64        conn = http.client.HTTPSConnection("api.tomtom.com")65        latlon = "{},{}:{},{}".format(lat1, lon1, lat2, lon2)66        url = "/routing/1/calculateRoute/{}/json?key={}&travelMode=bus".format(latlon, TOMTOM_API_KEY)67        conn.request("GET", url)68        resp = conn.getresponse()69        data = resp.read()70        status = int(resp.getcode())71        conn.close()72        return json.loads(data), status73    except Exception as e:74        print("Error in fetching route", e.__str__())75def format_latlng(raw_latlng: dict) -> List[float]:76    """77    Convert the lat lng data to a useful structure78    :param raw_latlng: raw lat lng object79    :return: list of the latitude and longitude80    """81    return [float(raw_latlng.get("latitude")), float(raw_latlng.get('longitude'))]82def check_status(status: int, error: str):83    """84    Check the status code of an HTTP request and abort the request85    :param status: status code of the request86    :param error: additional error message87    :return: Nothing88    """89    if status == 404:90        abort(status, message="Not Found", error=error, status=status)91    elif status == 400:92        abort(status, message="Bad Request", error=error, status=status)93    elif status == 403:94        abort(status, message="Forbidden", error=error, status=status)95    elif status == 408:96        abort(status, message="Request Timeout", error=error, status=status)97    elif status == 418:98        # just a funny one, found on https://developer.mozilla.org/nl/docs/Web/HTTP/Status/41899        abort(status, message="I'm a teapot", error="The server refuses the attempt to brew coffee with a teapot",100              status=status)101    elif status == 500:102        abort(status, message="Internal Server Error", error=error, status=status)103def get_stop_details(entity_number: int, stop_number: int):104    """105    Get the details of a stop of De Lijn106    :param entity_number: entity number of the stop107    :param stop_number: number of the stop108    :return: a json object with the data of the stop109    """110    data, status = make_lijn_request("GET", "DLKernOpenData/api/v1/haltes/{}/{}".format(entity_number,111                                                                                        stop_number))112    check_status(status,113                 "Cannot get stop with number {} for entity {}".format(stop_number, entity_number))114    proc_stop = dict()115    proc_stop["desc"] = data.get("omschrijving")116    proc_stop["entityID"] = data.get("entiteitnummer")117    proc_stop["village"] = data.get("omschrijvingGemeente")118    proc_stop["latlng"] = data.get("geoCoordinaat")119    return proc_stop120def get_entity_from_url(object: dict) -> int:121    """122    Get the entity number from an object received from De Lijn API.123    :param object: object that contains reference url's124    :return: entity number of the object125    """126    # Split URL on '/'127    url = object.get("links")[0].get("url").split("/")...

liunx.py

Source:liunx.py

1#!/usr/bin/python2# coding=utf-83import os4import time5import psutil6import os7from email.mime.multipart import MIMEMultipart8from email.mime.text import MIMEText9import smtplib10import json11# os.system('/usr/local/nbpt/mobilets/ts-control-center/start.sh')12smtpserver='smtp.qq.com'13username='1213867918'14password='pskepuqwvgqebaga'15def jiank():16    monitor_name = set(['dataprovider','gsmts','tsreport','pasm','ts-control-center','ts-result-receiving','processing'])17    proc_dict = {}18    proc_name = set()19    monitor_map = {20        'dataprovider': '/usr/local/nbpt/mobilets/DataProvider/bin/startup.sh',21        'gsmts':'/usr/local/nbpt/mobilets/gsmts/bin/startup.sh',22        'tsreport':'/usr/local/nbpt/mobilets/tsreport/bin/startup.sh',23        'pasm': '/usr/local/nbpt/mobilets/pasm/bin/startup.sh',24        'ts-control-center': '/usr/local/nbpt/mobilets/ts-control-center/start.sh',25        'processing': '/usr/local/nbpt/mobilets/ts-result-processing/bin/start.sh',26        'ts-result-receiving': '/usr/local/nbpt/mobilets/ts-result-receiving/start.sh',27    }28    while True:29        for proc in psutil.process_iter(attrs=['pid', 'name']):30            proc_dict[proc.info['pid']] = proc.info['name']31            proc_name.add(proc.info['name'])32        proc_stop = monitor_name - proc_name33        print(proc_stop)34        if proc_stop:35            for p in proc_stop:36                p_name = p37                data = {38                    "ç¨åºéåºåè¦éç¥": {39                        "ç¨åºéåºæ¶é´": "### %s" % time.strftime("%Y-%m-%d %X") +40                                "#### %sç¨åºéåº,æ£å¨å°è¯èªå¨éå¯!!!" % p_name41                    },42                }43                send_data = json.dumps(data,ensure_ascii=False)44                Send_Email(send_data)45                os.system(monitor_map[p_name])46                proc_set = set()47                for proc_again in psutil.process_iter(attrs=['pid', 'name']):48                    proc_set.add(proc_again.info['name'])49        time.sleep(5)50        proc_name = set()51def Send_Email(content):52    message = MIMEMultipart()53    message['subject'] ='ç¨åºçæ§åè¦éç¥'54    message['From'] ='1213867918@qq.com'55    message['To'] ='xiongt@nbpt.cn'56    content_plain = MIMEText(content, 'html', 'utf-8')57    message.attach(content_plain)58    stmp = smtplib.SMTP_SSL(smtpserver, 465)59    stmp.login(username, password)60    stmp.sendmail('1213867918@qq.com', 'xiongt@nbpt.cn', message.as_string())...

ChatBot.py

Source:ChatBot.py

...12    TelegramModel.TelegramBot().sendMessage('test')13    bot.receivedMessage(update)14    # bot.sendMessage('got text')15    # bot.sendMessage(update.message.text)16def proc_stop(bot, update):17    bot.sendMessage('ë')18    bot.stop()19def firecracker():20    return 'êµ´ë¬ê°ë¤~~~~ ìì¤ë¦¬'21bot = TelegramModel.BotWhale()22# ëªë ¹ì´ ë±ë¡23bot.add_handler('rolling', proc_rolling)24bot.add_handler('stop', proc_stop)25bot.add_msg_handler(proc_received)26# bot.sendMessage('test')27# ìì...

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.