How to use test_report_host method in locust

Best Python code snippet using locust

test_web.py

Source:test_web.py

...258        r = requests.get("http://127.0.0.1:%i/stats/report?download=1" % self.web_port)259        self.assertEqual(200, r.status_code)260        self.assertIn("attachment", r.headers.get("Content-Disposition", ""))261        self.assertNotIn("Download the Report", r.text, "Download report link found in HTML content")262    def test_report_host(self):263        self.environment.host = "http://test.com"264        self.stats.log_request("GET", "/test", 120, 5612)265        r = requests.get("http://127.0.0.1:%i/stats/report" % self.web_port)266        self.assertEqual(200, r.status_code)267        self.assertIn("http://test.com", r.text)268    def test_report_host2(self):269        class MyUser(User):270            host = "http://test2.com"271            @task272            def my_task(self):273                pass274        self.environment.host = None275        self.environment.user_classes = [MyUser]276        self.stats.log_request("GET", "/test", 120, 5612)...

adgather.py

Source:adgather.py

1#!/usr/bin/python262import os3#####################################4# SO advertisement information gathering and reporting to payer5#6#####################################7import sys8ROOT = os.environ.get("CALIDUS_ROOT_DIR", "/SO")9LIBS = os.path.join(ROOT, "lib/pylibs")10sys.path.insert(0, LIBS)11files = os.listdir(LIBS)12for name in files:13    if name.endswith("zip"):14        file_path = os.path.join(LIBS, name)15        sys.path.insert(0, file_path)16import getopt17import logging18import zmq19import struct20import hashlib21import time22from logging.handlers import RotatingFileHandler23from logging.handlers import SysLogHandler24import event_pack25import dbhelper26import xmlhelper27import urllib228import urlparse29import socket30import json31from threading import Thread32from Queue import Queue33# typedef struct ses_vc_swch_info_ {34#    uint64_t            sid;35#    int32_t             sock_fd;36#    int32_t             ip;37#    int32_t             port;38#    int32_t             reserved;39# define VC_GUID_LEN 12840#    char                guid[VC_GUID_LEN];41# define MAX_UA_LEN 20042#    char                ua[MAX_UA_LEN];43#    char                url[0];44# }ses_vc_swch_info_t;45VC_SWCH_FORMAT = "QIIII128s200s1024s"46VC_SWCH_KEYS = ("sid", "sock_fd", "ip", "port", "pad", "guid", "ua", "url")47pendingReportList = Queue()48def runPendingReportList():49    global pendingReportList50    logger = logging.getLogger('adgather')51    while True:52        newUrl = pendingReportList.get()53        try:54            curtime = time.time()55            ret = urllib2.urlopen(newUrl).read()56            tDelay = int((time.time() - curtime) * 1000)57            logger.info("URI : [%s] ret : [%s] delay %s ms pending %s", newUrl, ret, tDelay, pendingReportList.qsize())58        except Exception, e:59            logger = logging.getLogger('adgather')60            logger.warning("runPendingReportList URI : [%s] %s %s pending %s", newUrl, e, pendingReportList.qsize())61            pass62def UrlProxyThread(name):63    logger = logging.getLogger('adgather')64    logger.info("runPendingReportList %s", name)65    while True:66        try:67            runPendingReportList()68        except Exception, e:69            logger = logging.getLogger('adgather')70            logger.warning("runPendingReportList %s ", e)71            pass72        time.sleep(1)73def startUrlProxy(idx):74    try:75        thread = Thread(target=UrlProxyThread, args=("UrlProxyThread-%s" % idx,))76        thread.daemon = True77        thread.start()78        # thread.join()79    except Exception, e:80        logger = logging.getLogger('adgather')81        logger.warning("startUrlProxy%s %s ", idx, e)82        pass83def int2ip(addr):84    return socket.inet_ntoa(struct.pack("I", addr))85class adStatus:86    """This class will subscribe the session swith information, find out the 87    advertisement been played and report to payers"""88    def __init__(self, cfg, debug):89        self.initLog(debug)90        self.adList = dict()91        self.adExtList = dict()92        self.reportCnt = 093        self.dispatchList = list()94        conf = cfg['configurations']95        try:96            infoDbCfg = conf['ad_info_db']97            self.logger.info(infoDbCfg)98            # adInfoDBArg = {"host":"epg.jstv.streamocean.net", "user":"root", "passwd":"db", "db":"so_ads"}99            self.adInfoDB = dbhelper.DB(**infoDbCfg)100        except Exception, e:101            print102            e103            self.logger.warning('Remote AD DB Error%s  ', e)104            raise105        # adStatisticDBArg = {"host":"127.0.0.1", "user":"root", "passwd":"db", "db":"so_svr_st"}106        try:107            adStatisticDBArg = conf['ad_statistics_db']108            self.logger.info(adStatisticDBArg)109            self.adStatisticDB = dbhelper.DB(**adStatisticDBArg)110        except Exception, e:111            self.logger.exception("AD Statistic DB Error %s", e)112            raise113        self.testReportHost = conf.get('test_report_host', '')114        self.addbUpInterval = conf.get('ad_info_up_sec', 3600)115        self.msg_svr = conf['msg_svr']116        formatter = logging.Formatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s', \117                                      datefmt='%a, %d %b %Y %H:%M:%S')118        syslog_server = conf.get('syslog_server', '127.0.0.1:6603')119        sysloghost = syslog_server.split(':')[0]120        syslogport = int(syslog_server.split(':')[1])121        # syslog = SysLogHandler(address=(('%s'%sysloghost, syslogport)))122        syslog = SysLogHandler(address='/dev/log', facility=SysLogHandler.LOG_LOCAL4)123        syslog.setFormatter(formatter)124        self.logger.addHandler(syslog)125    def initLog(self, debug=False):126        self.logger = logging.getLogger('adgather')127        formatter = logging.Formatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s', \128                                      datefmt='%a, %d %b %Y %H:%M:%S')129        if ROOT == '/SO':130            log_path = '/SO_logs/btrace/adgather.log'131        else:132            log_path = 'adgather.log'133        filelog = RotatingFileHandler(log_path, maxBytes=50 * 1024 * 1024, backupCount=20)134        filelog.setFormatter(formatter)135        # add ch to logger136        self.logger.addHandler(filelog)137        logging.getLogger('db').addHandler(filelog)138        if debug:139            console = logging.StreamHandler()140            self.logger.addHandler(console)141            self.logger.setLevel(logging.DEBUG)142            self.logger.debug("Set log level to debug mode")143        else:144            self.logger.setLevel(logging.INFO)145    def dispatchReport(self, url):146        global pendingReportList147        if pendingReportList.qsize() > 32 * 4 * 1024:148            self.logger.warning("pendingReportList %s drop", pendingReportList.qsize())149            return150        pendingReportList.put(url)151    def adListDiff(self, new, old):152        added = dict()153        deleted = dict()154        kept = dict()155        for key in new:156            if key in old:157                kept[key] = new[key]158            else:159                added[key] = new[key]160        for key in old:161            if key not in new: deleted[key] = old[key]162        return added, deleted, kept163    def updateCachedAd(self):164        sql = """select ad2vod_ingestion.ad2vod_guid , ad_info.ad_imptraurl, ad_info.ad_extension from ad2vod_ingestion, ad_info where ad_info.ad_id = ad2vod_ingestion.ad_id;"""165        try:166            ret = self.adInfoDB.executeAndFetch(sql)167        except Exception, e:168            self.logger.warning("Can not update ad data %s", e)169            return None170        if ret is None or len(ret) == 0:171            self.logger.warning("Ad data empty")172            return None173        newAdList = dict()174        newAdExtList = dict()175        for item in ret:176            key = item[0]177            value = item[1]178            newAdList[key] = value179            if len(item) > 2:180                newAdExtList[key] = item[2]181        added, deleted, kept = self.adListDiff(newAdList, self.adList)182        if len(added) != 0 or len(deleted) != 0:183            self.logger.info("Updated ad list added %s, deleted %s, kept len %s", added, deleted, len(kept))184        else:185            self.logger.info("Updated ad list, no change, ad list len %s", len(kept))186        self.adList = newAdList187        self.adExtList = newAdExtList188        self.logger.debug("Updated ad list %s", self.adList)189    def saveInfo(self, ip, mac, guid, ua):190        ipint = int(ip)191        sql = """INSERT INTO ad_statistic_t SET ip=%s, mac='%s', guid='%s', ua='%s';""" % (ipint, mac, guid, ua)192        try:193            # ret = self.adStatisticDB.executeOnly(sql)194            pass195        except Exception, e:196            self.logger.warning("Save to ad statistics error %s ", e)197    def parseVcSwitchMsg(self, message):198        """Parse the received message to a dict"""199        try:200            values = map(lambda v: v.strip('\0') if isinstance(v, str) else v,201                         struct.unpack(VC_SWCH_FORMAT, message))202            swchIfo = dict(zip(VC_SWCH_KEYS, values))203        except Exception, e:204            self.logger.warning("Can not recognize message %s %s", message, e)205            return None206        return swchIfo207    def getdm(self, swchIfo):208        ua = swchIfo['ua']209        url = swchIfo['url']210        if 'Alilive' in ua or 'sp=7po' in url:211            return '7po'212        else:213            return 'VLC'214    def report(self, swchIfo):215        """Fill the ad info to the URL which the payer want to get"""216        guid = swchIfo['guid']217        if guid not in self.adList:218            return  # Not a ad guid219        url = swchIfo['url']220        # halimin221        ipintT = swchIfo['ip']222        ipstrT = int2ip(ipintT)223        self.logger.info("guid : [%s] ip : [%s] Session URL: %s UA: %s" % (guid, ipstrT, url, swchIfo['ua']))224        # self.logger.info("guid : [%s] Session URL: %s UA: %s"%(guid, url, swchIfo['ua']))225        # ip226        ipint = swchIfo['ip']227        ipstr = int2ip(ipint)228        blockIpList = ['127.0.0.1']229        if ipstr in blockIpList:230            self.logger.info("blocked ip %s in list %s", ipstr, blockIpList)231            return232        res = urlparse.urlparse(url)233        query = res.query234        orgSesQueryDict = dict(urlparse.parse_qsl(query))235        # mac236        mac = ''237        if 'mac' not in orgSesQueryDict:238            mac = '000000000000'239        else:240            mac = orgSesQueryDict['mac'].upper()241            if len(mac) != 12:242                mac = '000000000000'243                self.logger.warning("Invalid mac %s" % mac)244        macTemp = ''245        for i in range(0, 5):246            macTemp += mac[2 * i:2 * i + 2]247            macTemp += ':'248        macTemp += mac[10:12]249        self.logger.debug("MAC before md5 %s", macTemp)250        m = hashlib.md5(macTemp)251        macMd5 = m.hexdigest()252        self.saveInfo(ipint, mac, guid, swchIfo['ua'])253        orgUrl = self.adList[guid]254        # print orgUrl255        dm = self.getdm(swchIfo)256        newUrl = orgUrl.replace('%dm%', dm)257        newUrl = newUrl.replace('%ip%', ipstr)258        newUrl = newUrl.replace('%mac%', macMd5)259        if self.testReportHost != '':260            newUrl = newUrl.replace('advapi.joyplus.tv/advapi', self.testReportHost)261        # print newUrl262        self.reportCnt += 1263        if self.reportCnt % 1000 == 0:264            # print "Reported %s"%self.reportCnt265            self.logger.info("Reported : [%s]" % self.reportCnt)266        try:267            self.dispatchReport(newUrl)268        except Exception, e:269            self.logger.warning("URI : [%s] Failed : [%s]", newUrl, e)270            pass271        curExtList = list()272        if guid in self.adExtList:273            curExtList = self.adExtList[guid]274        curExtList = json.loads(curExtList)275        for item in curExtList:276            if item['type'] == "report_url":277                for urlitem in item['content'].values():278                    if urlitem != orgUrl:279                        newUrl = urlitem.encode('utf-8')280                        newUrl = newUrl.replace('%dm%', dm)281                        newUrl = newUrl.replace('%ip%', ipstr)282                        newUrl = newUrl.replace('%mac%', macMd5)283                        try:284                            self.dispatchReport(newUrl)285                        except Exception, e:286                            self.logger.warning("Failed : [%s] %s", newUrl, e)287                            pass288    def vc_switchEventHandler(self, message):289        """Handle the vc switch event"""290        swchIfo = self.parseVcSwitchMsg(message)291        if swchIfo is None:292            return293        self.logger.debug("Parsered msg %s", swchIfo)294        guid = swchIfo['guid']295        if guid not in self.adList:296            self.logger.info("Received guid %s not in list message %s", guid, swchIfo)297            return298        try:299            self.report(swchIfo)300        except Exception, e:301            self.logger.warning("Report error for message %s error %s ", message, e)302            raise303    def run(self):304        context = zmq.Context()305        socket = context.socket(zmq.SUB)306        try:307            socket.connect(self.msg_svr)308            socket.setsockopt(zmq.SUBSCRIBE, "ses_vc_switch")309        except Exception, e:310            self.logger.warning("Can not connect to msg_svr %s %s", self.msg_svr, e)311        self.logger.debug("Starting to gather ad information...")312        updateTimerStart = time.time()313        self.updateCachedAd()314        while True:315            now = time.time()316            if now - updateTimerStart > self.addbUpInterval:317                self.logger.debug("Update the cached ad guid...")318                self.updateCachedAd()319                updateTimerStart = now320            self.logger.debug("Receive....")321            evtType, data = socket.recv_multipart()322            if evtType == "ses_vc_switch":323                events = event_pack.unpack(data)324                self.logger.debug("Received len %s", len(events))325                self.logger.debug("data: %s", events)326                self.vc_switchEventHandler(events[0])327            else:328                self.logger.warning("Get wrong msg %s", evtType)329def loadcfg(path):330    try:331        f = open(path)332        content = f.read()333        name, val = xmlhelper.parseSimpleXmlConfig(content)334    except Exception as e:335        print336        e337        self.logger.warning("loadcfg error %s", e)338        sys.exit(1)339    assert (name == 'so_ad_srv')340    return val341def main():342    try:343        opts, _ = getopt.getopt(sys.argv[1:], "dc:m")344    except getopt.GetoptError, err:345        print346        str(err)347        sys.exit(1)348    debug = False349    cfg = dict()350    for opt, val in opts:351        if opt == "-d":352            debug = True353            print354            "Debug ..."355        elif opt == "-c":356            try:357                cfg = loadcfg(val)358            except:359                print360                "Please check you cfg file"361                sys.exit(1)362    if cfg['configurations']['enable'] != 1:363        time.sleep(5)364        sys.exit(1)365    try:366        ap = adStatus(cfg, debug)367    except Exception, e:368        print369        e370        print371        "Init error"372        sys.exit(1)373    for i in range(0, 16):374        startUrlProxy(i)375    ap.run()376if __name__ == "__main__":...

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.