def __init__(self, immediate_ms = 5000):
     parse_args()
     self.dn = None
     
     # check if security is enabled, if so read certificate files
     self.security = not args.DISABLE_SSL
     if self.security:
         mplane.utils.check_file(args.CERTFILE)
         self.cert = mplane.utils.normalize_path(mplane.utils.read_setting(args.CERTFILE, "cert"))
         self.key = mplane.utils.normalize_path(mplane.utils.read_setting(args.CERTFILE, "key"))
         self.ca = mplane.utils.normalize_path(mplane.utils.read_setting(args.CERTFILE, "ca-chain"))
         mplane.utils.check_file(self.cert)
         mplane.utils.check_file(self.key)
         mplane.utils.check_file(self.ca)
         self.pool = HTTPSConnectionPool(args.SUPERVISOR_IP4, args.SUPERVISOR_PORT, key_file=self.key, cert_file=self.cert, ca_certs=self.ca)
     else: 
         self.pool = HTTPConnectionPool(args.SUPERVISOR_IP4, args.SUPERVISOR_PORT)
         self.cert = None
     
     # get server DN, for Access Control purposes
     self.dn = self.get_dn()
     
     # generate a Service for each capability
     self.immediate_ms = immediate_ms
     self.scheduler = mplane.scheduler.Scheduler(self.security, self.cert)
     self.scheduler.add_service(FirelogService(firelog_capability(args.URL)))

 def test_none_total_applies_connect(self):
     url = '/sleep?seconds=0.005'
     timeout = Timeout(total=None, connect=0.001)
     pool = HTTPConnectionPool(TARPIT_HOST, self.port, timeout=timeout)
     conn = pool._get_conn()
     self.assertRaises(ConnectTimeoutError, pool._make_request, conn, 'GET',
                       url)

    def test_delayed_body_read_timeout(self):
        timed_out = Event()

        def socket_handler(listener):
            sock = listener.accept()[0]
            buf = b''
            body = 'Hi'
            while not buf.endswith(b'\r\n\r\n'):
                buf = sock.recv(65536)
            sock.send(('HTTP/1.1 200 OK\r\n'
                       'Content-Type: text/plain\r\n'
                       'Content-Length: %d\r\n'
                       '\r\n' % len(body)).encode('utf-8'))

            timed_out.wait()
            sock.send(body.encode('utf-8'))
            sock.close()

        self._start_server(socket_handler)
        pool = HTTPConnectionPool(self.host, self.port)

        response = pool.urlopen('GET', '/', retries=0, preload_content=False,
                                timeout=Timeout(connect=1, read=0.001))
        try:
            self.assertRaises(ReadTimeoutError, response.read)
        finally:
            timed_out.set()

    def test_headers_are_sent_with_the_original_case(self):
        headers = {"foo": "bar", "bAz": "quux"}
        parsed_headers = {}

        def socket_handler(listener):
            sock = listener.accept()[0]

            buf = b""
            while not buf.endswith(b"\r\n\r\n"):
                buf += sock.recv(65536)

            headers_list = [header for header in buf.split(b"\r\n")[1:] if header]

            for header in headers_list:
                (key, value) = header.split(b": ")
                parsed_headers[key.decode()] = value.decode()

            # Send incomplete message (note Content-Length)
            sock.send(("HTTP/1.1 204 No Content\r\n" "Content-Length: 0\r\n" "\r\n").encode("utf-8"))

            sock.close()

        self._start_server(socket_handler)
        expected_headers = {"Accept-Encoding": "identity", "Host": "{0}:{1}".format(self.host, self.port)}
        expected_headers.update(headers)

        pool = HTTPConnectionPool(self.host, self.port, retries=False)
        pool.request("GET", "/", headers=HTTPHeaderDict(headers))
        self.assertEqual(expected_headers, parsed_headers)

 def test_connection_refused(self):
     # Does the pool retry if there is no listener on the port?
     # Note: Socket server is not started until after the test.
     pool = HTTPConnectionPool(self.host, self.port)
     with self.assertRaises(MaxRetryError):
         pool.request('GET', '/')
     self._start_server(lambda x: None)

 def test_timeout(self):
     pool = HTTPConnectionPool(HOST, PORT, timeout=0.1)
     try:
         r = pool.get_url('/sleep', fields={'seconds': 0.2})
         self.fail("Failed to raise TimeoutError exception")
     except TimeoutError, e:
         pass

    def test_headers_are_sent_with_the_original_case(self):
        headers = {'foo': 'bar', 'bAz': 'quux'}
        parsed_headers = {}

        def socket_handler(listener):
            sock = listener.accept()[0]

            buf = b''
            while not buf.endswith(b'\r\n\r\n'):
                buf += sock.recv(65536)

            headers_list = [header for header in buf.split(b'\r\n')[1:] if header]

            for header in headers_list:
                (key, value) = header.split(b': ')
                parsed_headers[key.decode('ascii')] = value.decode('ascii')

            sock.send((
                'HTTP/1.1 204 No Content\r\n'
                'Content-Length: 0\r\n'
                '\r\n').encode('utf-8'))

            sock.close()

        self._start_server(socket_handler)
        expected_headers = {'Accept-Encoding': 'identity',
                            'Host': '{0}:{1}'.format(self.host, self.port)}
        expected_headers.update(headers)

        pool = HTTPConnectionPool(self.host, self.port, retries=False)
        pool.request('GET', '/', headers=HTTPHeaderDict(headers))
        self.assertEqual(expected_headers, parsed_headers)

    def test_incomplete_response(self):
        body = 'Response'
        partial_body = body[:2]

        def socket_handler(listener):
            sock = listener.accept()[0]

            # Consume request
            buf = b''
            while not buf.endswith(b'\r\n\r\n'):
                buf = sock.recv(65536)

            # Send partial response and close socket.
            sock.send((
                'HTTP/1.1 200 OK\r\n'
                'Content-Type: text/plain\r\n'
                'Content-Length: %d\r\n'
                '\r\n'
                '%s' % (len(body), partial_body)).encode('utf-8')
            )
            sock.close()

        self._start_server(socket_handler)
        pool = HTTPConnectionPool(self.host, self.port)

        response = pool.request('GET', '/', retries=0, preload_content=False)
        self.assertRaises(ProtocolError, response.read)

 def __init__(self, immediate_ms = 5000):
     parse_args()
     self.dn = None
     
     # check if security is enabled, if so read certificate files
     self.security = not args.DISABLE_SSL
     if self.security:
         mplane.utils.check_file(args.CERTFILE)
         self.cert = mplane.utils.normalize_path(mplane.utils.read_setting(args.CERTFILE, "cert"))
         self.key = mplane.utils.normalize_path(mplane.utils.read_setting(args.CERTFILE, "key"))
         self.ca = mplane.utils.normalize_path(mplane.utils.read_setting(args.CERTFILE, "ca-chain"))
         mplane.utils.check_file(self.cert)
         mplane.utils.check_file(self.key)
         mplane.utils.check_file(self.ca)
         self.pool = HTTPSConnectionPool(args.SUPERVISOR_IP4, args.SUPERVISOR_PORT, key_file=self.key, cert_file=self.cert, ca_certs=self.ca)
     else: 
         self.pool = HTTPConnectionPool(args.SUPERVISOR_IP4, args.SUPERVISOR_PORT)
     
     # get server DN, for Access Control purposes
     self.dn = self.get_dn()
     
     # generate a Service for each capability
     self.immediate_ms = immediate_ms
     self.scheduler = mplane.scheduler.Scheduler(self.security, self.cert)
     self.scheduler.add_service(tStatService(mplane.tstat_caps.tcp_flows_capability(args.IP4_NET), args.TSTAT_RUNTIMECONF))
     self.scheduler.add_service(tStatService(mplane.tstat_caps.e2e_tcp_flows_capability(args.IP4_NET), args.TSTAT_RUNTIMECONF))
     self.scheduler.add_service(tStatService(mplane.tstat_caps.tcp_options_capability(args.IP4_NET), args.TSTAT_RUNTIMECONF))
     self.scheduler.add_service(tStatService(mplane.tstat_caps.tcp_p2p_stats_capability(args.IP4_NET), args.TSTAT_RUNTIMECONF))
     self.scheduler.add_service(tStatService(mplane.tstat_caps.tcp_layer7_capability(args.IP4_NET), args.TSTAT_RUNTIMECONF))

 def test_bad_connect(self):
     pool = HTTPConnectionPool('badhost.invalid', self.port)
     try:
         pool.request('GET', '/', retries=5)
         self.fail("should raise timeout exception here")
     except MaxRetryError as e:
         self.assertEqual(type(e.reason), NewConnectionError)

 def test_bad_connect(self):
     pool = HTTPConnectionPool('badhost.invalid', self.port)
     try:
         pool.request('GET', '/', retries=5)
         self.fail("should raise timeout exception here")
     except MaxRetryError as e:
         self.assertTrue(isinstance(e.reason, ProtocolError), e.reason)

    def test_response_headers_are_returned_in_the_original_order(self):
        # NOTE: Probability this test gives a false negative is 1/(K!)
        K = 16
        # NOTE: Provide headers in non-sorted order (i.e. reversed)
        #       so that if the internal implementation tries to sort them,
        #       a change will be detected.
        expected_response_headers = [('X-Header-%d' % i, str(i)) for i in reversed(range(K))]
        
        def socket_handler(listener):
            sock = listener.accept()[0]

            buf = b''
            while not buf.endswith(b'\r\n\r\n'):
                buf += sock.recv(65536)

            sock.send(b'HTTP/1.1 200 OK\r\n' +
                      b'\r\n'.join([
                          (k.encode('utf8') + b': ' + v.encode('utf8'))
                          for (k, v) in expected_response_headers
                      ]) +
                      b'\r\n')
            sock.close()

        self._start_server(socket_handler)
        pool = HTTPConnectionPool(self.host, self.port)
        r = pool.request('GET', '/', retries=0)
        actual_response_headers = [
            (k, v) for (k, v) in r.headers.items()
            if k.startswith('X-Header-')
        ]
        self.assertEqual(expected_response_headers, actual_response_headers)

    def test_enforce_content_length_no_body(self):
        done_event = Event()

        def socket_handler(listener):
            sock = listener.accept()[0]

            buf = b''
            while not buf.endswith(b'\r\n\r\n'):
                buf += sock.recv(65536)

            sock.send(
                b'HTTP/1.1 200 OK\r\n'
                b'Content-Length: 22\r\n'
                b'Content-type: text/plain\r\n'
                b'\r\n'
            )
            done_event.wait(1)
            sock.close()

        self._start_server(socket_handler)
        conn = HTTPConnectionPool(self.host, self.port, maxsize=1)
        self.addCleanup(conn.close)

        # Test stream on 0 length body
        head_response = conn.request('HEAD', url='/', preload_content=False,
                                     enforce_content_length=True)
        data = [chunk for chunk in head_response.stream(1)]
        self.assertEqual(len(data), 0)

        done_event.set()

    def test_stream_none_unchunked_response_does_not_hang(self):
        done_event = Event()

        def socket_handler(listener):
            sock = listener.accept()[0]

            buf = b''
            while not buf.endswith(b'\r\n\r\n'):
                buf += sock.recv(65536)

            sock.send(
                b'HTTP/1.1 200 OK\r\n'
                b'Content-Length: 12\r\n'
                b'Content-type: text/plain\r\n'
                b'\r\n'
                b'hello, world'
            )
            done_event.wait(5)
            sock.close()

        self._start_server(socket_handler)
        pool = HTTPConnectionPool(self.host, self.port, retries=False)
        self.addCleanup(pool.close)
        r = pool.request('GET', '/', timeout=1, preload_content=False)

        # Stream should read to the end.
        self.assertEqual([b'hello, world'], list(r.stream(None)))

        done_event.set()

 def test_source_address_ignored(self):
     # source_address is ignored in Python 2.6 and older.
     for addr in INVALID_SOURCE_ADDRESSES:
         pool = HTTPConnectionPool(
             self.host, self.port, source_address=addr)
         r = pool.request('GET', '/source_address')
         assert r.status == 200

 def test_httplib_headers_case_insensitive(self):
     self.start_response_handler(
         b"HTTP/1.1 200 OK\r\n" b"Content-Length: 0\r\n" b"Content-type: text/plain\r\n" b"\r\n"
     )
     pool = HTTPConnectionPool(self.host, self.port, retries=False)
     HEADERS = {"Content-Length": "0", "Content-type": "text/plain"}
     r = pool.request("GET", "/")
     self.assertEqual(HEADERS, dict(r.headers.items()))  # to preserve case sensitivity

 def test_source_address(self):
     for addr in VALID_SOURCE_ADDRESSES:
         pool = HTTPConnectionPool(self.host, self.port,
                 source_address=addr, retries=False)
         r = pool.request('GET', '/source_address')
         assert r.data == b(addr[0]), (
             "expected the response to contain the source address {addr}, "
             "but was {data}".format(data=r.data, addr=b(addr[0])))

    def test_release_conn_parameter(self):
        MAXSIZE=5
        pool = HTTPConnectionPool(self.host, self.port, maxsize=MAXSIZE)
        self.assertEqual(pool.pool.qsize(), MAXSIZE)

        # Make request without releasing connection
        pool.request('GET', '/', release_conn=False, preload_content=False)
        self.assertEqual(pool.pool.qsize(), MAXSIZE-1)

 def test_timeout(self):
     pool = HTTPConnectionPool(self.host, self.port, timeout=0.01)
     try:
         pool.request('GET', '/sleep',
                      fields={'seconds': '0.02'})
         self.fail("Failed to raise TimeoutError exception")
     except TimeoutError:
         pass

 def test_socket_options(self):
     """Test that connections accept socket options."""
     # This test needs to be here in order to be run. socket.create_connection actually tries to
     # connect to the host provided so we need a dummyserver to be running.
     pool = HTTPConnectionPool(self.host, self.port, socket_options=[(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)])
     s = pool._new_conn()._new_conn()  # Get the socket
     using_keepalive = s.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE) > 0
     self.assertTrue(using_keepalive)
     s.close()