# - Simple GET on /hello/restart_results (returns the leak results)
+from __future__ import division
+from __future__ import print_function
+from future import standard_library
+standard_library.install_aliases()
+from builtins import str
+from builtins import range
+from builtins import object
import threading
import socket
import time
import argparse
-import httplib
+import http.client
import sys
import string
import random
_verbose_ = False
-class Session:
+class Session(object):
def __init__(self, addr, port, timeout = 15):
self.client = socket.create_connection((addr, int(port)), timeout = timeout)
self.target = addr
def send_err_check(self, request, data=None):
rval = True
try:
- self.client.sendall(request);
+ self.client.sendall(request.encode());
if data:
- self.client.sendall(data)
+ self.client.sendall(data.encode())
except socket.error as err:
self.client.close()
- print "Socket Error in send :", err
+ print("Socket Error in send :", err)
rval = False
return rval
def send_get(self, path, headers=None):
request = "GET " + path + " HTTP/1.1\r\nHost: " + self.target
if headers:
- for field, value in headers.iteritems():
+ for field, value in headers.items():
request += "\r\n"+field+": "+value
request += "\r\n\r\n"
return self.send_err_check(request)
def send_put(self, path, data, headers=None):
request = "PUT " + path + " HTTP/1.1\r\nHost: " + self.target
if headers:
- for field, value in headers.iteritems():
+ for field, value in headers.items():
request += "\r\n"+field+": "+value
request += "\r\nContent-Length: " + str(len(data)) +"\r\n\r\n"
return self.send_err_check(request, data)
def send_post(self, path, data, headers=None):
request = "POST " + path + " HTTP/1.1\r\nHost: " + self.target
if headers:
- for field, value in headers.iteritems():
+ for field, value in headers.items():
request += "\r\n"+field+": "+value
request += "\r\nContent-Length: " + str(len(data)) +"\r\n\r\n"
return self.send_err_check(request, data)
state = 'nothing'
resp_read = ''
while True:
- char = self.client.recv(1)
+ char = self.client.recv(1).decode()
if char == '\r' and state == 'nothing':
state = 'first_cr'
elif char == '\n' and state == 'first_cr':
return headers
except socket.error as err:
self.client.close()
- print "Socket Error in recv :", err
+ print("Socket Error in recv :", err)
return None
def read_resp_data(self):
read_data = ''
if self.encoding != 'chunked':
while len(read_data) != self.content_len:
- read_data += self.client.recv(self.content_len)
+ read_data += self.client.recv(self.content_len).decode()
else:
chunk_data_buf = ''
while (True):
read_ch = self.client.recv(1)
# Check CRLF
if (read_ch == '\r'):
- read_ch = self.client.recv(1)
+ read_ch = self.client.recv(1).decode()
if (read_ch == '\n'):
# If CRLF decode length of chunk
chunk_len = int(chunk_data_buf, 16)
# Fetch remaining CRLF
if self.client.recv(2) != "\r\n":
# Error in packet
- print "Error in chunked data"
+ print("Error in chunked data")
return None
if not chunk_len:
# If last chunk
return read_data
except socket.error as err:
self.client.close()
- print "Socket Error in recv :", err
+ print("Socket Error in recv :", err)
return None
def close(self):
def test_val(text, expected, received):
if expected != received:
- print " Fail!"
- print " [reason] " + text + ":"
- print " expected: " + str(expected)
- print " received: " + str(received)
+ print(" Fail!")
+ print(" [reason] " + text + ":")
+ print(" expected: " + str(expected))
+ print(" received: " + str(received))
return False
return True
# Pipeline 3 requests
if (_verbose_):
- print " Thread: Using adder start " + str(self.id)
+ print(" Thread: Using adder start " + str(self.id))
for _ in range(self.depth):
self.session.send_post('/adder', str(self.id))
def adder_result(self):
if len(self.response) != self.depth:
- print "Error : missing response packets"
+ print("Error : missing response packets")
return False
for i in range(len(self.response)):
if not test_val("Thread" + str(self.id) + " response[" + str(i) + "]",
def get_hello(dut, port):
# GET /hello should return 'Hello World!'
- print "[test] GET /hello returns 'Hello World!' =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] GET /hello returns 'Hello World!' =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("GET", "/hello")
resp = conn.getresponse()
if not test_val("status_code", 200, resp.status):
conn.close()
return False
- if not test_val("data", "Hello World!", resp.read()):
+ if not test_val("data", "Hello World!", resp.read().decode()):
conn.close()
return False
if not test_val("data", "application/json", resp.getheader('Content-Type')):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def put_hello(dut, port):
# PUT /hello returns 405'
- print "[test] PUT /hello returns 405' =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] PUT /hello returns 405' =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("PUT", "/hello", "Hello")
resp = conn.getresponse()
if not test_val("status_code", 405, resp.status):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def post_hello(dut, port):
# POST /hello returns 405'
- print "[test] POST /hello returns 404' =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] POST /hello returns 404' =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("POST", "/hello", "Hello")
resp = conn.getresponse()
if not test_val("status_code", 405, resp.status):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def post_echo(dut, port):
# POST /echo echoes data'
- print "[test] POST /echo echoes data' =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] POST /echo echoes data' =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("POST", "/echo", "Hello")
resp = conn.getresponse()
if not test_val("status_code", 200, resp.status):
conn.close()
return False
- if not test_val("data", "Hello", resp.read()):
+ if not test_val("data", "Hello", resp.read().decode()):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def put_echo(dut, port):
# PUT /echo echoes data'
- print "[test] PUT /echo echoes data' =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] PUT /echo echoes data' =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("PUT", "/echo", "Hello")
resp = conn.getresponse()
if not test_val("status_code", 200, resp.status):
conn.close()
return False
- if not test_val("data", "Hello", resp.read()):
+ if not test_val("data", "Hello", resp.read().decode()):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def get_echo(dut, port):
# GET /echo returns 404'
- print "[test] GET /echo returns 405' =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] GET /echo returns 405' =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("GET", "/echo")
resp = conn.getresponse()
if not test_val("status_code", 405, resp.status):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def get_hello_type(dut, port):
# GET /hello/type_html returns text/html as Content-Type'
- print "[test] GET /hello/type_html has Content-Type of text/html =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] GET /hello/type_html has Content-Type of text/html =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("GET", "/hello/type_html")
resp = conn.getresponse()
if not test_val("status_code", 200, resp.status):
conn.close()
return False
- if not test_val("data", "Hello World!", resp.read()):
+ if not test_val("data", "Hello World!", resp.read().decode()):
conn.close()
return False
if not test_val("data", "text/html", resp.getheader('Content-Type')):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def get_hello_status(dut, port):
# GET /hello/status_500 returns status 500'
- print "[test] GET /hello/status_500 returns status 500 =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] GET /hello/status_500 returns status 500 =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("GET", "/hello/status_500")
resp = conn.getresponse()
if not test_val("status_code", 500, resp.status):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def get_false_uri(dut, port):
# GET /false_uri returns status 404'
- print "[test] GET /false_uri returns status 404 =>",
- conn = httplib.HTTPConnection(dut, int(port), timeout=15)
+ print("[test] GET /false_uri returns status 404 =>", end=' ')
+ conn = http.client.HTTPConnection(dut, int(port), timeout=15)
conn.request("GET", "/false_uri")
resp = conn.getresponse()
if not test_val("status_code", 404, resp.status):
conn.close()
return False
- print "Success"
+ print("Success")
conn.close()
return True
def parallel_sessions_adder(dut, port, max_sessions):
# POSTs on /adder in parallel sessions
- print "[test] POST {pipelined} on /adder in " + str(max_sessions) + " sessions =>",
+ print("[test] POST {pipelined} on /adder in " + str(max_sessions) + " sessions =>", end=' ')
t = []
# Create all sessions
- for i in xrange(max_sessions):
+ for i in range(max_sessions):
t.append(adder_thread(i, dut, port))
- for i in xrange(len(t)):
+ for i in range(len(t)):
t[i].start()
- for i in xrange(len(t)):
+ for i in range(len(t)):
t[i].join()
res = True
- for i in xrange(len(t)):
+ for i in range(len(t)):
if not test_val("Thread" + str(i) + " Failed", t[i].adder_result(), True):
res = False
t[i].close()
if (res):
- print "Success"
+ print("Success")
return res
def async_response_test(dut, port):
# Test that an asynchronous work is executed in the HTTPD's context
# This is tested by reading two responses over the same session
- print "[test] Test HTTPD Work Queue (Async response) =>",
+ print("[test] Test HTTPD Work Queue (Async response) =>", end=' ')
s = Session(dut, port)
s.send_get('/async_data')
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def leftover_data_test(dut, port):
# Leftover data in POST is purged (valid and invalid URIs)
- print "[test] Leftover data in POST is purged (valid and invalid URIs) =>",
- s = httplib.HTTPConnection(dut + ":" + port, timeout=15)
+ print("[test] Leftover data in POST is purged (valid and invalid URIs) =>", end=' ')
+ s = http.client.HTTPConnection(dut + ":" + port, timeout=15)
s.request("POST", url='/leftover_data', body="abcdefghijklmnopqrstuvwxyz\r\nabcdefghijklmnopqrstuvwxyz")
resp = s.getresponse()
- if not test_val("Partial data", "abcdefghij", resp.read()):
+ if not test_val("Partial data", "abcdefghij", resp.read().decode()):
s.close()
return False
s.request("GET", url='/hello')
resp = s.getresponse()
- if not test_val("Hello World Data", "Hello World!", resp.read()):
+ if not test_val("Hello World Data", "Hello World!", resp.read().decode()):
s.close()
return False
s.request("GET", url='/hello')
resp = s.getresponse()
- if not test_val("Hello World Data", "Hello World!", resp.read()):
+ if not test_val("Hello World Data", "Hello World!", resp.read().decode()):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def spillover_session(dut, port, max_sess):
# Session max_sess_sessions + 1 is rejected
- print "[test] Session max_sess_sessions (" + str(max_sess) + ") + 1 is rejected =>",
+ print("[test] Session max_sess_sessions (" + str(max_sess) + ") + 1 is rejected =>", end=' ')
s = []
_verbose_ = True
- for i in xrange(max_sess + 1):
+ for i in range(max_sess + 1):
if (_verbose_):
- print "Executing " + str(i)
+ print("Executing " + str(i))
try:
- a = httplib.HTTPConnection(dut + ":" + port, timeout=15)
+ a = http.client.HTTPConnection(dut + ":" + port, timeout=15)
a.request("GET", url='/hello')
resp = a.getresponse()
- if not test_val("Connection " + str(i), "Hello World!", resp.read()):
+ if not test_val("Connection " + str(i), "Hello World!", resp.read().decode()):
a.close()
break
s.append(a)
except:
if (_verbose_):
- print "Connection " + str(i) + " rejected"
+ print("Connection " + str(i) + " rejected")
a.close()
break
a.close()
# Check if number of connections is equal to max_sess
- print ["Fail","Success"][len(s) == max_sess]
+ print(["Fail","Success"][len(s) == max_sess])
return (len(s) == max_sess)
def recv_timeout_test(dut, port):
- print "[test] Timeout occurs if partial packet sent =>",
+ print("[test] Timeout occurs if partial packet sent =>", end=' ')
s = Session(dut, port)
- s.client.sendall("GE")
+ s.client.sendall(b"GE")
s.read_resp_hdrs()
resp = s.read_resp_data()
if not test_val("Request Timeout", "Server closed this connection", resp):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def packet_size_limit_test(dut, port, test_size):
- print "[test] send size limit test =>",
+ print("[test] send size limit test =>", end=' ')
retry = 5
while (retry):
retry -= 1
- print "data size = ", test_size
- s = httplib.HTTPConnection(dut + ":" + port, timeout=15)
- random_data = ''.join(string.printable[random.randint(0,len(string.printable))-1] for _ in range(test_size))
+ print("data size = ", test_size)
+ s = http.client.HTTPConnection(dut + ":" + port, timeout=15)
+ random_data = ''.join(string.printable[random.randint(0,len(string.printable))-1] for _ in list(range(test_size)))
path = "/echo"
s.request("POST", url=path, body=random_data)
resp = s.getresponse()
if not test_val("Error", "200", str(resp.status)):
if test_val("Error", "408", str(resp.status)):
- print "Data too large to be allocated"
- test_size = test_size/10
+ print("Data too large to be allocated")
+ test_size = test_size//10
else:
- print "Unexpected error"
+ print("Unexpected error")
s.close()
- print "Retry..."
+ print("Retry...")
continue
- resp = resp.read()
+ resp = resp.read().decode()
result = (resp == random_data)
if not result:
test_val("Data size", str(len(random_data)), str(len(resp)))
s.close()
- print "Retry..."
+ print("Retry...")
continue
s.close()
- print "Success"
+ print("Success")
return True
- print "Failed"
+ print("Failed")
return False
def code_500_server_error_test(dut, port):
- print "[test] 500 Server Error test =>",
+ print("[test] 500 Server Error test =>", end=' ')
s = Session(dut, port)
- s.client.sendall("abcdefgh\0")
+ s.client.sendall(b"abcdefgh\0")
s.read_resp_hdrs()
resp = s.read_resp_data()
# Presently server sends back 400 Bad Request
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def code_501_method_not_impl(dut, port):
- print "[test] 501 Method Not Implemented =>",
+ print("[test] 501 Method Not Implemented =>", end=' ')
s = Session(dut, port)
path = "/hello"
- s.client.sendall("ABC " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n")
+ s.client.sendall(("ABC " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
# Presently server sends back 400 Bad Request
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def code_505_version_not_supported(dut, port):
- print "[test] 505 Version Not Supported =>",
+ print("[test] 505 Version Not Supported =>", end=' ')
s = Session(dut, port)
path = "/hello"
- s.client.sendall("GET " + path + " HTTP/2.0\r\nHost: " + dut + "\r\n\r\n")
+ s.client.sendall(("GET " + path + " HTTP/2.0\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
if not test_val("Server Error", "505", s.status):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def code_400_bad_request(dut, port):
- print "[test] 400 Bad Request =>",
+ print("[test] 400 Bad Request =>", end=' ')
s = Session(dut, port)
path = "/hello"
- s.client.sendall("XYZ " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n")
+ s.client.sendall(("XYZ " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
if not test_val("Client Error", "400", s.status):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def code_404_not_found(dut, port):
- print "[test] 404 Not Found =>",
+ print("[test] 404 Not Found =>", end=' ')
s = Session(dut, port)
path = "/dummy"
- s.client.sendall("GET " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n")
+ s.client.sendall(("GET " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
if not test_val("Client Error", "404", s.status):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def code_405_method_not_allowed(dut, port):
- print "[test] 405 Method Not Allowed =>",
+ print("[test] 405 Method Not Allowed =>", end=' ')
s = Session(dut, port)
path = "/hello"
- s.client.sendall("POST " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n")
+ s.client.sendall(("POST " + path + " HTTP/1.1\r\nHost: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
if not test_val("Client Error", "405", s.status):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def code_408_req_timeout(dut, port):
- print "[test] 408 Request Timeout =>",
+ print("[test] 408 Request Timeout =>", end=' ')
s = Session(dut, port)
- s.client.sendall("POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: 10\r\n\r\nABCD")
+ s.client.sendall(("POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: 10\r\n\r\nABCD").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
if not test_val("Client Error", "408", s.status):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def code_411_length_required(dut, port):
- print "[test] 411 Length Required =>",
+ print("[test] 411 Length Required =>", end=' ')
s = Session(dut, port)
path = "/echo"
- s.client.sendall("POST " + path + " HTTP/1.1\r\nHost: " + dut + "\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n")
+ s.client.sendall(("POST " + path + " HTTP/1.1\r\nHost: " + dut + "\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
# Presently server sends back 400 Bad Request
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
def send_getx_uri_len(dut, port, length):
method = "GET "
version = " HTTP/1.1\r\n"
path = "/"+"x"*(length - len(method) - len(version) - len("/"))
- s.client.sendall(method)
+ s.client.sendall(method.encode())
time.sleep(1)
- s.client.sendall(path)
+ s.client.sendall(path.encode())
time.sleep(1)
- s.client.sendall(version + "Host: " + dut + "\r\n\r\n")
+ s.client.sendall((version + "Host: " + dut + "\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
s.close()
return s.status
def code_414_uri_too_long(dut, port, max_uri_len):
- print "[test] 414 URI Too Long =>",
+ print("[test] 414 URI Too Long =>", end=' ')
status = send_getx_uri_len(dut, port, max_uri_len)
if not test_val("Client Error", "404", status):
return False
status = send_getx_uri_len(dut, port, max_uri_len + 1)
if not test_val("Client Error", "414", status):
return False
- print "Success"
+ print("Success")
return True
def send_postx_hdr_len(dut, port, length):
host = "Host: " + dut
custom_hdr_field = "\r\nCustom: "
custom_hdr_val = "x"*(length - len(host) - len(custom_hdr_field) - len("\r\n\r\n") + len("0"))
- request = "POST " + path + " HTTP/1.1\r\n" + host + custom_hdr_field + custom_hdr_val + "\r\n\r\n"
- s.client.sendall(request[:length/2])
+ request = ("POST " + path + " HTTP/1.1\r\n" + host + custom_hdr_field + custom_hdr_val + "\r\n\r\n").encode()
+ s.client.sendall(request[:length//2])
time.sleep(1)
- s.client.sendall(request[length/2:])
+ s.client.sendall(request[length//2:])
hdr = s.read_resp_hdrs()
resp = s.read_resp_data()
s.close()
return False, s.status
def code_431_hdr_too_long(dut, port, max_hdr_len):
- print "[test] 431 Header Too Long =>",
+ print("[test] 431 Header Too Long =>", end=' ')
res, status = send_postx_hdr_len(dut, port, max_hdr_len)
if not res:
return False
res, status = send_postx_hdr_len(dut, port, max_hdr_len + 1)
if not test_val("Client Error", "431", status):
return False
- print "Success"
+ print("Success")
return True
def test_upgrade_not_supported(dut, port):
- print "[test] Upgrade Not Supported =>",
+ print("[test] Upgrade Not Supported =>", end=' ')
s = Session(dut, port)
path = "/hello"
- s.client.sendall("OPTIONS * HTTP/1.1\r\nHost:" + dut + "\r\nUpgrade: TLS/1.0\r\nConnection: Upgrade\r\n\r\n");
+ s.client.sendall(("OPTIONS * HTTP/1.1\r\nHost:" + dut + "\r\nUpgrade: TLS/1.0\r\nConnection: Upgrade\r\n\r\n").encode())
s.read_resp_hdrs()
resp = s.read_resp_data()
if not test_val("Client Error", "200", s.status):
s.close()
return False
s.close()
- print "Success"
+ print("Success")
return True
if __name__ == '__main__':
_verbose_ = True
- print "### Basic HTTP Client Tests"
+ print("### Basic HTTP Client Tests")
get_hello(dut, port)
post_hello(dut, port)
put_hello(dut, port)
get_hello_status(dut, port)
get_false_uri(dut, port)
- print "### Error code tests"
+ print("### Error code tests")
code_500_server_error_test(dut, port)
code_501_method_not_impl(dut, port)
code_505_version_not_supported(dut, port)
# Not supported yet (Error on chunked request)
###code_411_length_required(dut, port)
- print "### Sessions and Context Tests"
+ print("### Sessions and Context Tests")
parallel_sessions_adder(dut, port, max_sessions)
leftover_data_test(dut, port)
async_response_test(dut, port)
# See the License for the specific language governing permissions and
# limitations under the License.
-import httplib
+from __future__ import print_function
+from __future__ import unicode_literals
+from future import standard_library
+standard_library.install_aliases()
+from builtins import str
+import http.client
import argparse
def verbose_print(verbosity, *args):
if (verbosity):
- print ''.join(str(elems) for elems in args)
+ print(''.join(str(elems) for elems in args))
def test_get_handler(ip, port, verbosity = False):
verbose_print(verbosity, "======== GET HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = httplib.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
uri = "/hello?query1=value1&query2=value2&query3=value3"
# GET hello response
test_headers = {"Test-Header-1":"Test-Value-1", "Test-Header-2":"Test-Value-2"}
verbose_print(verbosity, "Sending GET to URI : ", uri)
verbose_print(verbosity, "Sending additional headers : ")
- for k, v in test_headers.iteritems():
+ for k, v in test_headers.items():
verbose_print(verbosity, "\t", k, ": ", v)
sess.request("GET", url=uri, headers=test_headers)
resp = sess.getresponse()
resp_hdrs = resp.getheaders()
- resp_data = resp.read()
+ resp_data = resp.read().decode()
try:
if resp.getheader("Custom-Header-1") != "Custom-Value-1":
return False
verbose_print(verbosity, "======== POST HANDLER TEST ============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = httplib.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
# POST message to /echo and get back response
sess.request("POST", url="/echo", body=msg)
resp = sess.getresponse()
- resp_data = resp.read()
+ resp_data = resp.read().decode()
verbose_print(verbosity, "Server response to POST /echo (" + msg + ")")
verbose_print(verbosity, "vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv")
verbose_print(verbosity, resp_data)
verbose_print(verbosity, "======== PUT HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = httplib.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
# PUT message to /ctrl to disable /hello URI handler
verbose_print(verbosity, "Disabling /hello handler")
sess.request("GET", url="/hello")
resp = sess.getresponse()
- resp_data1 = resp.read()
+ resp_data1 = resp.read().decode()
verbose_print(verbosity, "Response on GET /hello : " + resp_data1)
# PUT message to /ctrl to enable /hello URI handler
sess.request("GET", url="/hello")
resp = sess.getresponse()
- resp_data2 = resp.read()
+ resp_data2 = resp.read().decode()
verbose_print(verbosity, "Response on GET /hello : " + resp_data2)
# Close HTTP connection
verbose_print(verbosity, "======== GET HANDLER TEST =============")
# Establish HTTP connection
verbose_print(verbosity, "Connecting to => " + ip + ":" + port)
- sess = httplib.HTTPConnection(ip + ":" + port, timeout = 15)
+ sess = http.client.HTTPConnection(ip + ":" + port, timeout = 15)
uri = "/hello?" + query
# GET hello response
verbose_print(verbosity, "Sending GET to URI : ", uri)
sess.request("GET", url=uri, headers={})
resp = sess.getresponse()
- resp_data = resp.read()
+ resp_data = resp.read().decode()
verbose_print(verbosity, "vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv")
verbose_print(verbosity, "Server response to GET /hello")
msg = args['msg']
if not test_get_handler (ip, port, True):
- print "Failed!"
+ print("Failed!")
if not test_post_handler(ip, port, msg, True):
- print "Failed!"
+ print("Failed!")
if not test_put_handler (ip, port, True):
- print "Failed!"
+ print("Failed!")
projects / esp-idf / commitdiff