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

How to use do_proxy method in Airtest

Best Python code snippet using Airtest

dhttpd.py

Source:dhttpd.py

1#!/usr/bin/python2# (c) Vincent Simonet, 2015.3import argparse4import httplib5import mimetypes6import os7import posixpath8import shutil9import threading10import urllib11import BaseHTTPServer12import SimpleHTTPServer13from SocketServer import ThreadingMixIn14class DelegatingHTTPServer(BaseHTTPServer.HTTPServer):15  def __init__(self, server_address,16               RequestHandlerClass,17               dserver_address,18               localdir):19    BaseHTTPServer.HTTPServer.__init__(self,20                                       server_address, RequestHandlerClass)21    self.dserver_host = dserver_address[0]22    self.dserver_port = dserver_address[1]23    self.localdir = os.path.abspath(localdir)24class ThreadedDelegatingHTTPServer(ThreadingMixIn, DelegatingHTTPServer):25    """Handle requests in a separate thread."""26class RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):27  _FILTERED_HEADERS = set(['connection', 'host'])28  def send_local(self):29    path = self.translate_path(self.path)30    f = None31    if os.path.isdir(path):32      for index in "index.html", "index.htm":33        index = os.path.join(path, index)34        if os.path.exists(index):35          path = index36          break37        else:38          return False39    ctype = self.guess_type(path)40    if ctype.startswith('text/'):41      mode = 'r'42    else:43      mode = 'rb'44    try:45      f = open(path, mode)46    except IOError:47      return False48    self.send_response(200)49    self.send_header("Content-type", ctype)50    self.end_headers()51    self.copyfile(f, self.wfile)52    f.close()53    return True54  def translate_path(self, path):55    path = posixpath.normpath(urllib.unquote(path))56    words = path.split('/')57    words = filter(None, words)58    path = self.server.localdir59    for word in words:60      drive, word = os.path.splitdrive(word)61      head, word = os.path.split(word)62      if word in (os.curdir, os.pardir): continue63      path = os.path.join(path, word)64    return path65  def copyfile(self, source, outputfile):66    shutil.copyfileobj(source, outputfile)67  def guess_type(self, path):68    base, ext = posixpath.splitext(path)69    if self.extensions_map.has_key(ext):70      return self.extensions_map[ext]71    ext = ext.lower()72    if self.extensions_map.has_key(ext):73      return self.extensions_map[ext]74    else:75      return self.extensions_map['']76  extensions_map = mimetypes.types_map.copy()77  extensions_map.update({78    '': 'application/octet-stream', # Default79    '.py': 'text/plain',80    '.c': 'text/plain',81    '.h': 'text/plain',82  })83  def do_LOCAL_or_PROXY(self):84    if not self.send_local():85      self.do_PROXY()86  def do_PROXY(self):87    conn = httplib.HTTPConnection(self.server.dserver_host,88                                  self.server.dserver_port)89    content_len = int(self.headers.get('content-length', 0))90    body = self.rfile.read(content_len)91    headers = dict((key, self.headers.get(key)) for key in self.headers92                   if not key in self._FILTERED_HEADERS)93    conn.request(self.command, self.path, body, headers)94    resp = conn.getresponse()95    self.send_response(resp.status, resp.reason)96    for name, value in resp.getheaders():97      self.send_header(name, value)98    self.end_headers()99    self.wfile.write(resp.read())100    conn.close()101  do_GET     = do_LOCAL_or_PROXY102  do_POST    = do_PROXY103  do_HEAD    = do_LOCAL_or_PROXY104  do_PUT     = do_PROXY105  do_DELETE  = do_PROXY106  do_OPTIONS = do_PROXY107_DESCRIPTION = """108An HTTP server that serves files from a local directory and a delegate HTTP109server.  For every request, the server first look for a local file to search.110If a file is found, it is returned directly.  If no local file is found, the111request is forwarded to the delegate HTTP server.112"""113def _parse_host_port(s):114  parts = s.split(':')115  if len(parts) == 1:116    return parts[0], 80117  else:118    return parts[0], int(parts[1])119def main():120  parser = argparse.ArgumentParser(description=_DESCRIPTION)121  parser.add_argument('--dserver',122                      action='store',123                      type=str,124                      dest='dserver',125                      help='Delegate HTTP server (default: localhost:8080)',126                      metavar='HOST:PORT',127                      default='localhost:8080')128  parser.add_argument('-p', '--port',129                      action='store',130                      type=int,131                      dest='port',132                      help='Port to listen (default: 8081)',133                      metavar='PORT',134                      default='8081')135  parser.add_argument('-l', '--localdir',136                      action='store',137                      type=str,138                      dest='localdir',139                      help='Local directory to serve',140                      metavar='DIR',141                      default='.')142  args = parser.parse_args()143  server_address = ('', args.port)144  dserver_address = _parse_host_port(args.dserver)145  httpd = ThreadedDelegatingHTTPServer(server_address,146                                       RequestHandler,147                                       dserver_address,148                                       args.localdir)149  httpd.daemon_threads = True150  print 'Listening port ' + str(args.port)151  try:152    httpd.serve_forever()153  except KeyboardInterrupt:154    httpd.shutdown()155    print '\nThat\'s all folks!'156if __name__ == '__main__':...

middleware.py

Source:middleware.py

1import re2import logging3from django import http4from django.conf import settings5import requests6PRERENDER_TOKEN = getattr(settings, 'PRERENDER_TOKEN', None)7PRERENDER_UA_REGEX = getattr(settings, 'PRERENDER_UA_REGEX', "Ask Jeeves|baiduspider|twitterbot|facebookexternalhit|rogerbot|linkedinbot|embedly|quora link preview|showyoubot|outbrain|pinterest|slackbot|vkShare|W3C_Validator")8PRERENDER_GOOGLEBOT = getattr(settings, 'PRERENDER_GOOGLEBOT', False)9PRERENDER_BINGBOT = getattr(settings, 'PRERENDER_BINGBOT', False)10PRERENDER_PROXY = getattr(settings, 'PRERENDER_PROXY', 'https://service.prerender.io/')11PRERENDER_PATHS_EXACT = getattr(settings, 'PRERENDER_PATHS_EXACT', ())12PRERENDER_PATHS_STARTSWITH = getattr(settings, 'PRERENDER_PATHS_STARTSWITH', ())13if PRERENDER_GOOGLEBOT:14  PRERENDER_UA_REGEX = 'googlebot|' + PRERENDER_UA_REGEX15  16if PRERENDER_BINGBOT:17  PRERENDER_UA_REGEX = 'BingPreview|bingbot|' + PRERENDER_UA_REGEX18  19logger = logging.getLogger(__name__)20class PreRender:21  def __init__ (self, get_response):22    self.get_response = get_response23    24  def __call__ (self, request):25    if PRERENDER_TOKEN:26      ua = request.META.get('HTTP_USER_AGENT', '')27      do_proxy = False28      29      if 'Prerender' in ua:30        do_proxy = False31        32      elif request.method == 'GET':33        if "_escaped_fragment_" in request.GET or re.search(PRERENDER_UA_REGEX, ua, re.I):34          if PRERENDER_PATHS_EXACT and request.path in PRERENDER_PATHS_EXACT:35            do_proxy = True36            37          elif PRERENDER_PATHS_STARTSWITH and request.path.startswith(PRERENDER_PATHS_STARTSWITH):38            do_proxy = True39            40      if do_proxy:41        return self.proxy(request)42        43    response = self.get_response(request)44    return response45    46  @staticmethod47  def proxy (request):48    url = PRERENDER_PROXY + request.build_absolute_uri()49    logger.info('Fetching Proxy: {}'.format(url))50    r = requests.get(51      url,52      headers={'X-Prerender-Token': PRERENDER_TOKEN},53      allow_redirects=False54    )55    return http.HttpResponse(r.text)...

proxy.py

Source:proxy.py

...4from . import register_command5from .pretty import print_info, print_fail6from ..request import Request7class APIProxyHandler(http.server.BaseHTTPRequestHandler):8    def do_proxy(self):9        content_len = int(self.headers.get('Content-Length', 0))10        body = self.rfile.read(content_len)11        path = re.sub(r'^/?v(\d+)/', '/', self.path)12        rqst = Request(self.command, path, body)13        resp = rqst.send()14        self.send_response(resp.status, resp.reason)15        self.send_header('Content-Type', resp.content_type)16        self.send_header('Content-Length', resp.content_length)17        self.end_headers()18        self.wfile.write(resp.content)19    def do_HEAD(self):20        self.do_proxy()21    def do_GET(self):22        self.do_proxy()23    def do_POST(self):24        self.do_proxy()25    def do_PUT(self):26        self.do_proxy()27    def do_DELETE(self):28        self.do_proxy()29    def do_PATCH(self):30        self.do_proxy()31    def do_OPTIONS(self):32        self.do_proxy()33@register_command34def proxy(args):35    '''36    Run a non-encyprted non-authorized API proxy server.37    Use this only for development and testing!38    '''39    print_info('Starting an insecure API proxy at http://{0}:{1}'40               .format(args.bind, args.port))41    try:42        addr = (args.bind, args.port)43        httpd = http.server.HTTPServer(addr, APIProxyHandler)44        httpd.serve_forever()45    except (KeyboardInterrupt, SystemExit):46        print()...

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.