KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
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 locust 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?
