--- a/Pipfile
+++ b/Pipfile
@@ -10,11 +10,12 @@ pipenv = "==2018.5.18"
virtualenv = "==15.2.0"
six = "==1.10.0"
attrs = "==18.1.0"
pytest = "==3.2.5"
jsmin = "==2.1.0"
python-hglib = "==2.4"
blessings = "==1.6.1"
requests = "==2.9.1"
+browsermob-proxy = "==0.8.0"
[requires]
python_version = "2.7"
--- a/Pipfile.lock
+++ b/Pipfile.lock
@@ -1,12 +1,12 @@
{
"_meta": {
"hash": {
- "sha256": "1882eda49800a768871b7aab4de017ceb3c4397dcef3ec810e1416a965090b59"
+ "sha256": "2bd4e754eae750c3dacf6151a5d6c13dee6e7857772901871501aae6f78ed02b"
},
"pipfile-spec": 6,
"requires": {
"python_version": "2.7"
},
"sources": [
{
"name": "pypi",
@@ -28,16 +28,24 @@
"hashes": [
"sha256:26dbaf2f89c3e6dee11c10f7c0b85756ed75cf602b1bb7935b4efd8ed67a000f",
"sha256:466e43ff45723b272311de0437649df464b33b4daba7a54c69493212958e19c7",
"sha256:74919575885552e14bc24a68f8b539690bd1b5629180faa830b1a25b8c7fb6ea"
],
"index": "pypi",
"version": "==1.6.1"
},
+ "browsermob-proxy": {
+ "hashes": [
+ "sha256:5f0e72767938d268999f1b56b0e8ff01cecd051bb868637ff550e25495cc840b",
+ "sha256:fb345bc2207fccdb8a584694c8d02d01c2cfc539c9d43bbed38f0c54e1abbbaf"
+ ],
+ "index": "pypi",
+ "version": "==0.8.0"
+ },
"certifi": {
"hashes": [
"sha256:13e698f54293db9f89122b0581843a782ad0934a4fe0172d2a980ba77fc61bb7",
"sha256:9fa520c1bacfb634fa7af20a76bcbd3d5fb390481724c597da32c719a7dca4b0"
],
"version": "==2018.4.16"
},
"jsmin": {
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/PKG-INFO
@@ -0,0 +1,19 @@
+Metadata-Version: 1.1
+Name: browsermob-proxy
+Version: 0.8.0
+Summary: A library for interacting with the Browsermob Proxy
+Home-page: http://oss.theautomatedtester.co.uk/browsermob-proxy-py
+Author: David Burns
+Author-email: david.burns@theautomatedtester.co.uk
+License: UNKNOWN
+Description: UNKNOWN
+Platform: UNKNOWN
+Classifier: Development Status :: 3 - Alpha
+Classifier: Intended Audience :: Developers
+Classifier: License :: OSI Approved :: Apache Software License
+Classifier: Operating System :: POSIX
+Classifier: Operating System :: Microsoft :: Windows
+Classifier: Operating System :: MacOS :: MacOS X
+Classifier: Topic :: Software Development :: Testing
+Classifier: Topic :: Software Development :: Libraries
+Classifier: Programming Language :: Python
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/browsermobproxy/__init__.py
@@ -0,0 +1,6 @@
+__version__ = '0.5.0'
+
+from .server import RemoteServer, Server
+from .client import Client
+
+__all__ = ['RemoteServer', 'Server', 'Client', 'browsermobproxy']
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/browsermobproxy/client.py
@@ -0,0 +1,359 @@
+import requests
+
+try:
+ from urllib.parse import urlencode, unquote
+except ImportError:
+ from urllib import urlencode, unquote
+import json
+
+
+class Client(object):
+ def __init__(self, url, params=None, options=None):
+ """
+ Initialises a new Client object
+
+
+ :param url: This is where the BrowserMob Proxy lives
+ :param params: URL query (for example httpProxy and httpsProxy vars)
+ :param options: Dictionary that can contain the port of an existing
+ proxy to use (for example 'existing_proxy_port_to_use')
+ """
+ params = params if params is not None else {}
+ options = options if options is not None else {}
+ self.host = "http://" + url
+ if params:
+ urlparams = "?" + unquote(urlencode(params))
+ else:
+ urlparams = ""
+ if 'existing_proxy_port_to_use' in options:
+ self.port = options['existing_proxy_port_to_use']
+ else:
+ resp = requests.post('%s/proxy' % self.host + urlparams)
+ content = resp.content.decode('utf-8')
+ try:
+ jcontent = json.loads(content)
+ except Exception as e:
+ raise Exception("Could not read Browsermob-Proxy json\n"
+ "Another server running on this port?\n%s..." % content[:512])
+ self.port = jcontent['port']
+ url_parts = self.host.split(":")
+ self.proxy = url_parts[1][2:] + ":" + str(self.port)
+
+ def close(self):
+ """
+ shuts down the proxy and closes the port
+ """
+ r = requests.delete('%s/proxy/%s' % (self.host, self.port))
+ return r.status_code
+
+ # webdriver integration
+ # ...as a proxy object
+ def selenium_proxy(self):
+ """
+ Returns a Selenium WebDriver Proxy class with details of the HTTP Proxy
+ """
+ from selenium import webdriver
+ return webdriver.Proxy({
+ "httpProxy": self.proxy,
+ "sslProxy": self.proxy,
+ })
+
+ def webdriver_proxy(self):
+ """
+ Returns a Selenium WebDriver Proxy class with details of the HTTP Proxy
+ """
+ return self.selenium_proxy()
+
+ # ...as a capability
+ def add_to_capabilities(self, capabilities):
+ """
+ Adds an 'proxy' entry to a desired capabilities dictionary with the
+ BrowserMob proxy information
+
+
+ :param capabilities: The Desired capabilities object from Selenium WebDriver
+ """
+ capabilities['proxy'] = {
+ 'proxyType': "MANUAL",
+ 'httpProxy': self.proxy,
+ 'sslProxy': self.proxy
+ }
+
+ def add_to_webdriver_capabilities(self, capabilities):
+ self.add_to_capabilities(capabilities)
+
+ # browsermob proxy api
+ @property
+ def proxy_ports(self):
+ """
+ Return a list of proxy ports available
+ """
+ # r should look like {u'proxyList': [{u'port': 8081}]}
+ r = requests.get('%s/proxy' % self.host).json()
+ ports = [port['port'] for port in r['proxyList']]
+
+ return ports
+
+ @property
+ def har(self):
+ """
+ Gets the HAR that has been recorded
+ """
+ r = requests.get('%s/proxy/%s/har' % (self.host, self.port))
+
+ return r.json()
+
+ def new_har(self, ref=None, options=None, title=None):
+ """
+ This sets a new HAR to be recorded
+
+ :param str ref: A reference for the HAR. Defaults to None
+ :param dict options: A dictionary that will be passed to BrowserMob
+ Proxy with specific keywords. Keywords are:
+
+ - captureHeaders: Boolean, capture headers
+ - captureContent: Boolean, capture content bodies
+ - captureBinaryContent: Boolean, capture binary content
+
+ :param str title: the title of first har page. Defaults to ref.
+ """
+ options = options if options is not None else {}
+ payload = {"initialPageRef": ref} if ref is not None else {}
+ if title is not None:
+ payload.update({'initialPageTitle': title})
+
+ if options:
+ payload.update(options)
+
+ r = requests.put('%s/proxy/%s/har' % (self.host, self.port), payload)
+ if r.status_code == 200:
+ return (r.status_code, r.json())
+ else:
+ return (r.status_code, None)
+
+ def new_page(self, ref=None, title=None):
+ """
+ This sets a new page to be recorded
+
+ :param str ref: A reference for the new page. Defaults to None
+ :param str title: the title of new har page. Defaults to ref.
+ """
+ payload = {"pageRef": ref} if ref is not None else {}
+ if title is not None:
+ payload.update({'pageTitle': title})
+ r = requests.put('%s/proxy/%s/har/pageRef' % (self.host, self.port),
+ payload)
+ return r.status_code
+
+ def blacklist(self, regexp, status_code):
+ """
+ Sets a list of URL patterns to blacklist
+
+ :param str regex: a comma separated list of regular expressions
+ :param int status_code: the HTTP status code to return for URLs
+ that do not match the blacklist
+ """
+ r = requests.put('%s/proxy/%s/blacklist' % (self.host, self.port),
+ {'regex': regexp, 'status': status_code})
+ return r.status_code
+
+ def whitelist(self, regexp, status_code):
+ """
+ Sets a list of URL patterns to whitelist
+
+ :param str regex: a comma separated list of regular expressions
+ :param int status_code: the HTTP status code to return for URLs
+ that do not match the whitelist
+ """
+ r = requests.put('%s/proxy/%s/whitelist' % (self.host, self.port),
+ {'regex': regexp, 'status': status_code})
+ return r.status_code
+
+ def basic_authentication(self, domain, username, password):
+ """
+ This add automatic basic authentication
+
+ :param str domain: domain to set authentication credentials for
+ :param str username: valid username to use when authenticating
+ :param str password: valid password to use when authenticating
+ """
+ r = requests.post(url='%s/proxy/%s/auth/basic/%s' % (self.host, self.port, domain),
+ data=json.dumps({'username': username, 'password': password}),
+ headers={'content-type': 'application/json'})
+ return r.status_code
+
+ def headers(self, headers):
+ """
+ This sets the headers that will set by the proxy on all requests
+
+ :param dict headers: this is a dictionary of the headers to be set
+ """
+ if not isinstance(headers, dict):
+ raise TypeError("headers needs to be dictionary")
+
+ r = requests.post(url='%s/proxy/%s/headers' % (self.host, self.port),
+ data=json.dumps(headers),
+ headers={'content-type': 'application/json'})
+ return r.status_code
+
+ def response_interceptor(self, js):
+ """
+ Executes the java/js code against each response
+ `HttpRequest request <https://netty.io/4.1/api/io/netty/handler/codec/http/HttpRequest.html>`_,
+ `HttpMessageContents contents <https://raw.githubusercontent.com/lightbody/browsermob-proxy/master/browsermob-core/src/main/java/net/lightbody/bmp/util/HttpMessageContents.java>`_,
+ `HttpMessageInfo messageInfo <https://raw.githubusercontent.com/lightbody/browsermob-proxy/master/browsermob-core/src/main/java/net/lightbody/bmp/util/HttpMessageInfo.java>`_
+ are available objects to interact with.
+ :param str js: the js/java code to execute
+ """
+ r = requests.post(url='%s/proxy/%s/filter/response' % (self.host, self.port),
+ data=js,
+ headers={'content-type': 'text/plain'})
+ return r.status_code
+
+ def request_interceptor(self, js):
+ """
+ Executes the java/js code against each response
+ `HttpRequest request <https://netty.io/4.1/api/io/netty/handler/codec/http/HttpRequest.html>`_,
+ `HttpMessageContents contents <https://raw.githubusercontent.com/lightbody/browsermob-proxy/master/browsermob-core/src/main/java/net/lightbody/bmp/util/HttpMessageContents.java>`_,
+ `HttpMessageInfo messageInfo <https://raw.githubusercontent.com/lightbody/browsermob-proxy/master/browsermob-core/src/main/java/net/lightbody/bmp/util/HttpMessageInfo.java>`_
+ are available objects to interact with.
+ :param str js: the js/java code to execute
+ """
+ r = requests.post(url='%s/proxy/%s/filter/request' % (self.host, self.port),
+ data=js,
+ headers={'content-type': 'text/plain'})
+ return r.status_code
+
+ LIMITS = {
+ 'upstream_kbps': 'upstreamKbps',
+ 'downstream_kbps': 'downstreamKbps',
+ 'latency': 'latency'
+ }
+
+ def limits(self, options):
+ """
+ Limit the bandwidth through the proxy.
+
+ :param dict options: A dictionary with all the details you want to set.
+ downstream_kbps - Sets the downstream kbps
+ upstream_kbps - Sets the upstream kbps
+ latency - Add the given latency to each HTTP request
+ """
+ params = {}
+
+ for (k, v) in list(options.items()):
+ if k not in self.LIMITS:
+ raise KeyError('invalid key: %s' % k)
+
+ params[self.LIMITS[k]] = int(v)
+
+ if len(list(params.items())) == 0:
+ raise KeyError("You need to specify one of the valid Keys")
+
+ r = requests.put('%s/proxy/%s/limit' % (self.host, self.port),
+ params)
+ return r.status_code
+
+ TIMEOUTS = {
+ 'request': 'requestTimeout',
+ 'read': 'readTimeout',
+ 'connection': 'connectionTimeout',
+ 'dns': 'dnsCacheTimeout'
+ }
+
+ def timeouts(self, options):
+ """
+ Configure various timeouts in the proxy
+
+ :param dict options: A dictionary with all the details you want to set.
+ request - request timeout (in seconds)
+ read - read timeout (in seconds)
+ connection - connection timeout (in seconds)
+ dns - dns lookup timeout (in seconds)
+ """
+ params = {}
+
+ for (k, v) in list(options.items()):
+ if k not in self.TIMEOUTS:
+ raise KeyError('invalid key: %s' % k)
+
+ params[self.TIMEOUTS[k]] = int(v)
+
+ if len(list(params.items())) == 0:
+ raise KeyError("You need to specify one of the valid Keys")
+
+ r = requests.put('%s/proxy/%s/timeout' % (self.host, self.port),
+ params)
+ return r.status_code
+
+ def remap_hosts(self, address=None, ip_address=None, hostmap=None):
+ """
+ Remap the hosts for a specific URL
+
+ :param str address: url that you wish to remap
+ :param str ip_address: IP Address that will handle all traffic for
+ the address passed in
+ :param **hostmap: Other hosts to be added as keyword arguments
+ """
+ hostmap = hostmap if hostmap is not None else {}
+ if (address is not None and ip_address is not None):
+ hostmap[address] = ip_address
+
+ r = requests.post('%s/proxy/%s/hosts' % (self.host, self.port),
+ json.dumps(hostmap),
+ headers={'content-type': 'application/json'})
+ return r.status_code
+
+ def wait_for_traffic_to_stop(self, quiet_period, timeout):
+ """
+ Waits for the network to be quiet
+
+ :param int quiet_period: number of milliseconds the network needs
+ to be quiet for
+ :param int timeout: max number of milliseconds to wait
+ """
+ r = requests.put('%s/proxy/%s/wait' % (self.host, self.port),
+ {'quietPeriodInMs': quiet_period, 'timeoutInMs': timeout})
+ return r.status_code
+
+ def clear_dns_cache(self):
+ """
+ Clears the DNS cache associated with the proxy instance
+ """
+ r = requests.delete('%s/proxy/%s/dns/cache' % (self.host, self.port))
+ return r.status_code
+
+ def rewrite_url(self, match, replace):
+ """
+ Rewrites the requested url.
+
+ :param match: a regex to match requests with
+ :param replace: unicode \
+ a string to replace the matches with
+ """
+ params = {
+ "matchRegex": match,
+ "replace": replace
+ }
+ r = requests.put('%s/proxy/%s/rewrite' % (self.host, self.port),
+ params)
+ return r.status_code
+
+ def clear_all_rewrite_url_rules(self):
+ """
+ Clears all URL rewrite rules
+ :return: status code
+ """
+
+ r = requests.delete('%s/proxy/%s/rewrite' % (self.host, self.port))
+ return r.status_code
+
+ def retry(self, retry_count):
+ """
+ Retries. No idea what its used for, but its in the API...
+
+ :param int retry_count: the number of retries
+ """
+ r = requests.put('%s/proxy/%s/retry' % (self.host, self.port),
+ {'retrycount': retry_count})
+ return r.status_code
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/browsermobproxy/exceptions.py
@@ -0,0 +1,2 @@
+class ProxyServerError(Exception):
+ pass
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/browsermobproxy/server.py
@@ -0,0 +1,143 @@
+import os
+import platform
+import socket
+import subprocess
+import time
+
+from .client import Client
+from .exceptions import ProxyServerError
+
+
+class RemoteServer(object):
+
+ def __init__(self, host, port):
+ """
+ Initialises a RemoteServer object
+
+ :param host: The host of the proxy server.
+ :param port: The port of the proxy server.
+ """
+ self.host = host
+ self.port = port
+
+ @property
+ def url(self):
+ """
+ Gets the url that the proxy is running on. This is not the URL clients
+ should connect to.
+ """
+ return "http://%s:%d" % (self.host, self.port)
+
+ def create_proxy(self, params=None):
+ """
+ Gets a client class that allow to set all the proxy details that you
+ may need to.
+
+ :param dict params: Dictionary where you can specify params
+ like httpProxy and httpsProxy
+ """
+ params = params if params is not None else {}
+ client = Client(self.url[7:], params)
+ return client
+
+ def _is_listening(self):
+ try:
+ socket_ = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ socket_.settimeout(1)
+ socket_.connect((self.host, self.port))
+ socket_.close()
+ return True
+ except socket.error:
+ return False
+
+
+class Server(RemoteServer):
+
+ def __init__(self, path='browsermob-proxy', options=None):
+ """
+ Initialises a Server object
+
+ :param str path: Path to the browsermob proxy batch file
+ :param dict options: Dictionary that can hold the port.
+ More items will be added in the future.
+ This defaults to an empty dictionary
+ """
+ options = options if options is not None else {}
+
+ path_var_sep = ':'
+ if platform.system() == 'Windows':
+ path_var_sep = ';'
+ if not path.endswith('.bat'):
+ path += '.bat'
+
+ exec_not_on_path = True
+ for directory in os.environ['PATH'].split(path_var_sep):
+ if(os.path.isfile(os.path.join(directory, path))):
+ exec_not_on_path = False
+ break
+
+ if not os.path.isfile(path) and exec_not_on_path:
+ raise ProxyServerError("Browsermob-Proxy binary couldn't be found "
+ "in path provided: %s" % path)
+
+ self.path = path
+ self.host = 'localhost'
+ self.port = options.get('port', 8080)
+ self.process = None
+
+ if platform.system() == 'Darwin':
+ self.command = ['sh']
+ else:
+ self.command = []
+ self.command += [path, '--port=%s' % self.port]
+
+ def start(self, options=None):
+ """
+ This will start the browsermob proxy and then wait until it can
+ interact with it
+
+ :param dict options: Dictionary that can hold the path and filename
+ of the log file with resp. keys of `log_path` and `log_file`
+ """
+ if options is None:
+ options = {}
+ log_path = options.get('log_path', os.getcwd())
+ log_file = options.get('log_file', 'server.log')
+ retry_sleep = options.get('retry_sleep', 0.5)
+ retry_count = options.get('retry_count', 60)
+ log_path_name = os.path.join(log_path, log_file)
+ self.log_file = open(log_path_name, 'w')
+
+ self.process = subprocess.Popen(self.command,
+ stdout=self.log_file,
+ stderr=subprocess.STDOUT)
+ count = 0
+ while not self._is_listening():
+ if self.process.poll():
+ message = (
+ "The Browsermob-Proxy server process failed to start. "
+ "Check {0}"
+ "for a helpful error message.".format(self.log_file))
+
+ raise ProxyServerError(message)
+ time.sleep(retry_sleep)
+ count += 1
+ if count == retry_count:
+ self.stop()
+ raise ProxyServerError("Can't connect to Browsermob-Proxy")
+
+ def stop(self):
+ """
+ This will stop the process running the proxy
+ """
+ if self.process.poll() is not None:
+ return
+
+ try:
+ self.process.kill()
+ self.process.wait()
+ except AttributeError:
+ # kill may not be available under windows environment
+ pass
+
+ self.log_file.close()
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/browsermobproxy/webdriver_event_listener.py
@@ -0,0 +1,35 @@
+from selenium.webdriver.support.abstract_event_listener import AbstractEventListener
+
+class WebDriverEventListener(AbstractEventListener):
+
+ def __init__(self, client, refs=None):
+ refs = refs if refs is not None else {}
+ self.client = client
+ self.hars = []
+ self.refs = refs
+
+ def before_navigate_to(self, url, driver):
+ if len(self.hars) != 0:
+ self.hars.append(self.client.har)
+ self.client.new_har("navigate-to-%s" % url, self.refs)
+
+ def before_navigate_back(self, driver=None):
+ if driver:
+ name = "-from-%s" % driver.current_url
+ else:
+ name = "navigate-back"
+ self.client.new_page(name)
+
+ def before_navigate_forward(self, driver=None):
+ if driver:
+ name = "-from-%s" % driver.current_url
+ else:
+ name = "navigate-forward"
+ self.client.new_page(name)
+
+ def before_click(self, element, driver):
+ name = "click-element-%s" % element.id
+ self.client.new_page(name)
+
+ def before_quit(self, driver):
+ self.hars.append(self.client.har)
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/setup.cfg
@@ -0,0 +1,5 @@
+[egg_info]
+tag_build =
+tag_date = 0
+tag_svn_revision = 0
+
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/setup.py
@@ -0,0 +1,20 @@
+from setuptools import setup, find_packages
+
+setup(name='browsermob-proxy',
+ version='0.8.0',
+ description='A library for interacting with the Browsermob Proxy',
+ author='David Burns',
+ author_email='david.burns@theautomatedtester.co.uk',
+ url='http://oss.theautomatedtester.co.uk/browsermob-proxy-py',
+ classifiers=['Development Status :: 3 - Alpha',
+ 'Intended Audience :: Developers',
+ 'License :: OSI Approved :: Apache Software License',
+ 'Operating System :: POSIX',
+ 'Operating System :: Microsoft :: Windows',
+ 'Operating System :: MacOS :: MacOS X',
+ 'Topic :: Software Development :: Testing',
+ 'Topic :: Software Development :: Libraries',
+ 'Programming Language :: Python'],
+ packages = find_packages(),
+ install_requires=['requests>=2.9.1'],
+ )
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/test/test_client.py
@@ -0,0 +1,272 @@
+import os.path
+import pytest
+import sys
+
+
+def setup_module(module):
+ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+class TestClient(object):
+ def setup_method(self, method):
+ from browsermobproxy.client import Client
+ self.client = Client("localhost:9090")
+
+ def teardown_method(self, method):
+ self.client.close()
+
+ def test_we_can_get_list_of_ports(self):
+ """
+ GET /proxy - get a list of ports attached to ProxyServer instances
+ managed by ProxyManager
+ """
+ ports = self.client.proxy_ports
+
+ assert(len(ports) == 1)
+ assert(9090 not in ports)
+
+ def test_headers_type(self):
+ """
+ /proxy/:port/headers needs to take a dictionary
+ """
+ with pytest.raises(TypeError):
+ self.client.headers(['foo'])
+
+ def test_headers_content(self):
+ """
+ /proxy/:port/headers needs to take a dictionary
+ and returns 200 when its successful
+ """
+ s = self.client.headers({'User-Agent': 'rubber ducks floating in a row'})
+ assert(s == 200)
+
+ def test_new_har(self):
+ """
+ /proxy/:port/har
+ and returns 204 when creating a har with a particular name the first time
+ and returns 200 and the previous har when creating one with the same name
+ """
+ status_code, har = self.client.new_har()
+ assert(status_code == 204)
+ assert(har is None)
+ status_code, har = self.client.new_har()
+ assert(status_code == 200)
+ assert('log' in har)
+
+ def _test_new_har(self):
+ """
+ /proxy/:port/har
+ and returns 204 when creating a har with a particular name the first time
+ and returns 200 and the previous har when creating one with the same name
+ """
+ status_code, har = self.client.new_har("elephants")
+ assert(status_code == 204)
+ assert(har is None)
+ status_code, har = self.client.new_har("elephants")
+ assert(status_code == 200)
+ assert('elephants' == har["log"]["pages"][0]['id'])
+
+ def test_new_page_defaults(self):
+ """
+ /proxy/:port/pageRef
+ adds a new page of 'Page N' when no page name is given
+ """
+ self.client.new_har()
+ self.client.new_page()
+ har = self.client.har
+ assert(len(har["log"]["pages"]) == 2)
+ assert(har["log"]["pages"][1]["id"] == "Page 1")
+
+ def test_new_named_page(self):
+ """
+ /proxy/:port/pageRef
+ adds a new page of 'buttress'
+ """
+ self.client.new_har()
+ self.client.new_page('buttress')
+ har = self.client.har
+ assert(len(har["log"]["pages"]) == 2)
+ assert(har["log"]["pages"][1]["id"] == "buttress")
+
+ def test_single_whitelist(self):
+ """
+ /proxy/:port/whitelist
+ adds a whitelist
+ """
+ status_code = self.client.whitelist("http://www\\.facebook\\.com/.*", 200)
+ assert(status_code == 200)
+
+ def test_multiple_whitelists(self):
+ """
+ /proxy/:port/whitelist
+ adds a whitelist
+ """
+ status_code = self.client.whitelist("http://www\\.facebook\\.com/.*,http://cdn\\.twitter\\.com", 200)
+ assert(status_code == 200)
+
+ def test_blacklist(self):
+ """
+ /proxy/:port/blacklist
+ adds a blacklist
+ """
+ status_code = self.client.blacklist("http://www\\.facebook\\.com/.*", 200)
+ assert(status_code == 200)
+
+ def test_basic_authentication(self):
+ """
+ /proxy/:port/auth/basic
+ adds automatic basic authentication
+ """
+ status_code = self.client.basic_authentication("www.example.com", "myUsername", "myPassword")
+ assert(status_code == 200)
+
+ def test_limits_invalid_key(self):
+ """
+ /proxy/:port/limits
+ pre-sending checking that the parameter is correct
+ """
+ with pytest.raises(KeyError):
+ self.client.limits({"hurray": "explosions"})
+
+ def test_limits_key_no_value(self):
+ """
+ /proxy/:port/limits
+ pre-sending checking that a parameter exists
+ """
+ with pytest.raises(KeyError):
+ self.client.limits({})
+
+ def test_limits_all_key_values(self):
+ """
+ /proxy/:port/limits
+ can send all 3 at once based on the proxy implementation
+ """
+ limits = {"upstream_kbps": 320, "downstream_kbps": 560, "latency": 30}
+ status_code = self.client.limits(limits)
+ assert(status_code == 200)
+
+ def test_rewrite(self):
+ """
+ /proxy/:port/rewrite
+
+ """
+ match = "/foo"
+ replace = "/bar"
+ status_code = self.client.rewrite_url(match, replace)
+ assert(status_code == 200)
+
+ def test_close(self):
+ """
+ /proxy/:port
+ close the proxy port
+ """
+ status_code = self.client.close()
+ assert(status_code == 200)
+ status_code = self.client.close()
+ assert(status_code == 404)
+
+ def test_response_interceptor_with_parsing_js(self):
+ """
+ /proxy/:port/interceptor/response
+ This test is only checking very basic syntax rules. The snippet needs to be JAVA/JS
+ Code only gets validated if the filter/interceptor is used.
+ """
+ js = 'alert("foo")'
+ status_code = self.client.response_interceptor(js)
+ assert(status_code == 200)
+
+ def test_response_interceptor_with_invalid_js(self):
+ """
+ /proxy/:port/interceptor/response
+ This test is only checking very basic syntax rules. The snippet needs to be JAVA/JS
+ Code only gets validated if the filter/interceptor is used.
+ """
+ js = 'alert("foo"'
+ status_code = self.client.response_interceptor(js)
+ assert(status_code == 500)
+
+ def test_request_interceptor_with_parsing_js(self):
+ """
+ /proxy/:port/interceptor/request
+ This test is only checking very basic syntax rules. The snippet needs to be JAVA/JS
+ Code only gets validated if the filter/interceptor is used.
+ """
+ js = 'alert("foo")'
+ status_code = self.client.request_interceptor(js)
+ assert(status_code == 200)
+
+ def test_request_interceptor_with_invalid_js(self):
+ """
+ /proxy/:port/interceptor/request
+ This test is only checking very basic syntax rules. The snippet needs to be JAVA/JS
+ Code only gets validated if the filter/interceptor is used.
+ """
+ js = 'alert("foo"'
+ status_code = self.client.request_interceptor(js)
+ assert(status_code == 500)
+
+ def test_timeouts_invalid_timeouts(self):
+ """
+ /proxy/:port/timeout
+ pre-sending checking that the parameter is correct
+ """
+ with pytest.raises(KeyError):
+ self.client.timeouts({"hurray": "explosions"})
+
+ def test_timeouts_key_no_value(self):
+ """
+ /proxy/:port/timeout
+ pre-sending checking that a parameter exists
+ """
+ with pytest.raises(KeyError):
+ self.client.timeouts({})
+
+ def test_timeouts_all_key_values(self):
+ """
+ /proxy/:port/timeout
+ can send all 3 at once based on the proxy implementation
+ """
+ timeouts = {"request": 2, "read": 2, "connection": 2, "dns": 3}
+ status_code = self.client.timeouts(timeouts)
+ assert(status_code == 200)
+
+ def test_remap_hosts(self):
+ """
+ /proxy/:port/hosts
+ """
+ status_code = self.client.remap_hosts("example.com", "1.2.3.4")
+ assert(status_code == 200)
+
+ def test_remap_hosts_with_hostmap(self):
+ """
+ /proxy/:port/hosts
+ """
+ status_code = self.client.remap_hosts(hostmap={"example.com": "1.2.3.4"})
+ assert(status_code == 200)
+
+ def test_wait_for_traffic_to_stop(self):
+ """
+ /proxy/:port/wait
+ """
+ status_code = self.client.wait_for_traffic_to_stop(2000, 10000)
+ assert(status_code == 200)
+
+ def test_clear_dns_cache(self):
+ """
+ /proxy/:port/dns/cache
+ """
+ status_code = self.client.clear_dns_cache()
+ assert(status_code == 200)
+
+ def test_rewrite_url(self):
+ """
+ /proxy/:port/rewrite
+ """
+ status_code = self.client.rewrite_url('http://www.facebook\.com', 'http://microsoft.com')
+ assert(status_code == 200)
+
+ def test_retry(self):
+ """
+ /proxy/:port/retry
+ """
+ status_code = self.client.retry(4)
+ assert(status_code == 200)
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/test/test_remote.py
@@ -0,0 +1,55 @@
+from os import environ
+
+from selenium import webdriver
+import selenium.webdriver.common.desired_capabilities
+import os
+import sys
+import pytest
+
+
+def setup_module(module):
+ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+
+class TestRemote(object):
+ def setup_method(self, method):
+ from browsermobproxy.client import Client
+ self.client = Client("localhost:9090")
+ chrome_binary = environ.get("CHROME_BIN", None)
+ self.desired_caps = selenium.webdriver.common.desired_capabilities.DesiredCapabilities.CHROME
+ if chrome_binary is not None:
+ self.desired_caps.update({
+ "chromeOptions": {
+ "binary": chrome_binary,
+ "args": ['no-sandbox']
+ }
+ })
+ self.driver = webdriver.Remote(
+ desired_capabilities=self.desired_caps,
+ proxy=self.client)
+
+ def teardown_method(self, method):
+ self.client.close()
+ self.driver.quit()
+
+ @pytest.mark.human
+ def test_set_clear_url_rewrite_rule(self):
+ targetURL = "https://www.saucelabs.com/versions.js"
+ assert 200 == self.client.rewrite_url(
+ "https://www.saucelabs.com/versions.+", "https://www.saucelabs.com/versions.json"
+ )
+ self.driver.get(targetURL)
+ assert "Sauce Connect" in self.driver.page_source
+ assert self.client.clear_all_rewrite_url_rules() == 200
+ self.driver.get(targetURL)
+ assert "Sauce Connect" not in self.driver.page_source
+
+ @pytest.mark.human
+ def test_response_interceptor(self):
+ content = "Response successfully intercepted"
+ targetURL = "https://saucelabs.com/versions.json?hello"
+ self.client.response_interceptor(
+ """if(messageInfo.getOriginalUrl().contains('?hello')){contents.setTextContents("%s");}""" % content
+ )
+ self.driver.get(targetURL)
+ assert content in self.driver.page_source
new file mode 100644
--- /dev/null
+++ b/third_party/python/browsermob-proxy/test/test_webdriver.py
@@ -0,0 +1,75 @@
+from os import environ
+
+from selenium import webdriver
+import selenium.webdriver.common.desired_capabilities
+from selenium.webdriver.common.proxy import Proxy
+import os
+import sys
+import copy
+import pytest
+
+def setup_module(module):
+ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+
+class TestWebDriver(object):
+ def setup_method(self, method):
+ from browsermobproxy.client import Client
+ self.client = Client("localhost:9090")
+ self.driver = None
+
+ def teardown_method(self, method):
+ self.client.close()
+ if self.driver is not None:
+ self.driver.quit()
+
+ @pytest.mark.human
+ def test_i_want_my_by_capability(self):
+ capabilities = selenium.webdriver.common.desired_capabilities.DesiredCapabilities.CHROME
+ self.client.add_to_capabilities(capabilities)
+ # sets self.driver for proper clean up
+ self._create_webdriver(capabilites=capabilities)
+
+ self._run_url_rewrite_test()
+
+ @pytest.mark.human
+ def test_i_want_my_by_proxy_object(self):
+ self._create_webdriver(capabilites=selenium.webdriver.common.desired_capabilities.DesiredCapabilities.CHROME,
+ proxy=self.client)
+
+ self._run_url_rewrite_test()
+
+ def test_what_things_look_like(self):
+ bmp_capabilities = copy.deepcopy(selenium.webdriver.common.desired_capabilities.DesiredCapabilities.FIREFOX)
+ self.client.add_to_capabilities(bmp_capabilities)
+
+ proxy_capabilities = copy.deepcopy(selenium.webdriver.common.desired_capabilities.DesiredCapabilities.FIREFOX)
+ proxy_addr = 'localhost:%d' % self.client.port
+ proxy = Proxy({'httpProxy': proxy_addr,'sslProxy': proxy_addr})
+ proxy.add_to_capabilities(proxy_capabilities)
+
+ assert bmp_capabilities == proxy_capabilities
+
+ def _create_webdriver(self, capabilites, proxy=None):
+ chrome_binary = environ.get("CHROME_BIN", None)
+ if chrome_binary is not None:
+ capabilites.update({
+ "chromeOptions": {
+ "binary": chrome_binary,
+ "args": ['no-sandbox']
+ }
+ })
+ if proxy is None:
+ self.driver = webdriver.Remote(desired_capabilities=capabilites)
+ else:
+ self.driver = webdriver.Remote(desired_capabilities=capabilites, proxy=proxy)
+
+ def _run_url_rewrite_test(self):
+ targetURL = "https://www.saucelabs.com/versions.js"
+ assert 200 == self.client.rewrite_url(
+ "https://www.saucelabs.com/versions.+", "https://www.saucelabs.com/versions.json"
+ )
+ self.driver.get(targetURL)
+ assert "Sauce Connect" in self.driver.page_source
+ assert self.client.clear_all_rewrite_url_rules() == 200
+ self.driver.get(targetURL)
+ assert "Sauce Connect" not in self.driver.page_source