esp-idf/tools/ci/python_packages/idf_http_server_test/client.py

256 lines
9.4 KiB
Python
Raw Normal View History

#!/usr/bin/env python
#
# SPDX-FileCopyrightText: 2018-2022 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from __future__ import print_function, unicode_literals
import argparse
import http.client
import logging
from builtins import str
def verbose_print(verbosity, *args):
if (verbosity):
logging.info(''.join(str(elems) for elems in args))
2018-12-04 02:32:48 -05:00
def test_val(text, expected, received):
if expected != received:
logging.info(' Fail!')
logging.info(' [reason] {} :'.format(text))
logging.info(' expected: {}'.format(expected))
logging.info(' received: {}'.format(received))
return False
return True
2018-12-04 02:32:48 -05:00
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 = 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 : ')
2018-09-10 09:13:47 -04:00
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()
2018-09-10 09:13:47 -04:00
resp_data = resp.read().decode()
# Close HTTP connection
sess.close()
if not (
test_val('Status code mismatch', 200, resp.status) and
test_val('Response mismatch', 'Custom-Value-1', resp.getheader('Custom-Header-1')) and
test_val('Response mismatch', 'Custom-Value-2', resp.getheader('Custom-Header-2')) and
test_val('Response mismatch', 'Hello World!', resp_data)
):
return False
verbose_print(verbosity, 'vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv')
verbose_print(verbosity, 'Server response to GET /hello')
verbose_print(verbosity, 'Response Headers : ')
for k, v in resp_hdrs:
verbose_print(verbosity, '\t', k, ': ', v)
verbose_print(verbosity, 'Response Data : ' + resp_data)
verbose_print(verbosity, '========================================\n')
return True
2018-12-04 02:32:48 -05:00
def test_post_handler(ip, port, msg, verbosity=False):
verbose_print(verbosity, '======== POST HANDLER TEST ============')
# Establish HTTP connection
verbose_print(verbosity, 'Connecting to => ' + ip + ':' + port)
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()
2018-09-10 09:13:47 -04:00
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, '========================================\n')
# Close HTTP connection
sess.close()
return test_val('Response mismatch', msg, resp_data)
2018-12-04 02:32:48 -05:00
def test_put_handler(ip, port, verbosity=False):
verbose_print(verbosity, '======== PUT HANDLER TEST =============')
# Establish HTTP connection
verbose_print(verbosity, 'Connecting to => ' + ip + ':' + port)
sess = http.client.HTTPConnection(ip + ':' + port, timeout=15)
# PUT message to /ctrl to disable /hello and /echo URI handlers
# and set 404 error handler to custom http_404_error_handler()
verbose_print(verbosity, 'Disabling /hello and /echo handlers')
sess.request('PUT', url='/ctrl', body='0')
resp = sess.getresponse()
resp.read()
try:
# Send HTTP request to /hello URI
sess.request('GET', url='/hello')
resp = sess.getresponse()
resp_data = resp.read().decode()
# 404 Error must be returned from server as URI /hello is no longer available.
# But the custom error handler http_404_error_handler() will not close the
# session if the requested URI is /hello
if not test_val('Status code mismatch', 404, resp.status):
raise AssertionError
# Compare error response string with expectation
verbose_print(verbosity, 'Response on GET /hello : ' + resp_data)
if not test_val('Response mismatch', '/hello URI is not available', resp_data):
raise AssertionError
# Using same session for sending an HTTP request to /echo, as it is expected
# that the custom error handler http_404_error_handler() would not have closed
# the session
sess.request('POST', url='/echo', body='Some content')
resp = sess.getresponse()
resp_data = resp.read().decode()
# 404 Error must be returned from server as URI /hello is no longer available.
# The custom error handler http_404_error_handler() will close the session
# this time as the requested URI is /echo
if not test_val('Status code mismatch', 404, resp.status):
raise AssertionError
# Compare error response string with expectation
verbose_print(verbosity, 'Response on POST /echo : ' + resp_data)
if not test_val('Response mismatch', '/echo URI is not available', resp_data):
raise AssertionError
try:
# Using same session should fail as by now the session would have closed
sess.request('POST', url='/hello', body='Some content')
resp = sess.getresponse()
resp.read().decode()
# If control reaches this point then the socket was not closed.
# This is not expected
verbose_print(verbosity, 'Socket not closed by server')
raise AssertionError
except http.client.HTTPException:
# Catch socket error as we tried to communicate with an already closed socket
pass
except http.client.HTTPException:
verbose_print(verbosity, 'Socket closed by server')
return False
except AssertionError:
return False
finally:
# Close HTTP connection
sess.close()
verbose_print(verbosity, 'Enabling /hello handler')
# Create new connection
sess = http.client.HTTPConnection(ip + ':' + port, timeout=15)
# PUT message to /ctrl to enable /hello URI handler
# and restore 404 error handler to default
sess.request('PUT', url='/ctrl', body='1')
resp = sess.getresponse()
resp.read()
# Close HTTP connection
sess.close()
# Create new connection
sess = http.client.HTTPConnection(ip + ':' + port, timeout=15)
try:
# Sending HTTP request to /hello should work now
sess.request('GET', url='/hello')
resp = sess.getresponse()
resp_data = resp.read().decode()
if not test_val('Status code mismatch', 200, resp.status):
raise AssertionError
verbose_print(verbosity, 'Response on GET /hello : ' + resp_data)
if not test_val('Response mismatch', 'Hello World!', resp_data):
raise AssertionError
# 404 Error handler should have been restored to default
sess.request('GET', url='/invalid')
resp = sess.getresponse()
resp_data = resp.read().decode()
if not test_val('Status code mismatch', 404, resp.status):
raise AssertionError
verbose_print(verbosity, 'Response on GET /invalid : ' + resp_data)
if not test_val('Response mismatch', 'Nothing matches the given URI', resp_data):
raise AssertionError
except http.client.HTTPException:
verbose_print(verbosity, 'Socket closed by server')
return False
except AssertionError:
return False
finally:
# Close HTTP connection
sess.close()
return True
2018-12-04 02:32:48 -05:00
def test_custom_uri_query(ip, port, query, verbosity=False):
verbose_print(verbosity, '======== GET HANDLER TEST =============')
# Establish HTTP connection
verbose_print(verbosity, 'Connecting to => ' + ip + ':' + port)
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()
2018-09-10 09:13:47 -04:00
resp_data = resp.read().decode()
verbose_print(verbosity, 'vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv')
verbose_print(verbosity, 'Server response to GET /hello')
verbose_print(verbosity, 'Response Data : ' + resp_data)
verbose_print(verbosity, '========================================\n')
# Close HTTP connection
sess.close()
return 'Hello World!' == resp_data
2018-12-04 02:32:48 -05:00
if __name__ == '__main__':
# Configure argument parser
parser = argparse.ArgumentParser(description='Run HTTPd Test')
2018-12-04 02:32:48 -05:00
parser.add_argument('IP', metavar='IP', type=str, help='Server IP')
parser.add_argument('port', metavar='port', type=str, help='Server port')
parser.add_argument('msg', metavar='message', type=str, help='Message to be sent to server')
args = vars(parser.parse_args())
# Get arguments
ip = args['IP']
port = args['port']
msg = args['msg']
if not (
test_get_handler(ip, port, True) and
test_put_handler(ip, port, True) and
test_post_handler(ip, port, msg, True)
):
logging.info('Failed!')