#!/usr/bin/env python # # Copyright 2018 Espressif Systems (Shanghai) PTE LTD # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # Utility for testing the web server. Test cases: # Assume the device supports 'n' simultaneous open sockets # # HTTP Server Tests # # 0. Firmware Settings: # - Create a dormant thread whose sole job is to call httpd_stop() when instructed # - Measure the following before httpd_start() is called: # - current free memory # - current free sockets # - Measure the same whenever httpd_stop is called # - Register maximum possible URI handlers: should be successful # - Register one more URI handler: should fail # - Deregister on URI handler: should be successful # - Register on more URI handler: should succeed # - Register separate handlers for /hello, /hello/type_html. Also # ensure that /hello/type_html is registered BEFORE /hello. (tests # that largest matching URI is picked properly) # - Create URI handler /adder. Make sure it uses a custom free_ctx # structure to free it up # 1. Using Standard Python HTTP Client # - simple GET on /hello (returns Hello World. Ensures that basic # firmware tests are complete, or returns error) # - POST on /hello (should fail) # - PUT on /hello (should fail) # - simple POST on /echo (returns whatever the POST data) # - simple PUT on /echo (returns whatever the PUT data) # - GET on /echo (should fail) # - simple GET on /hello/type_html (returns Content type as text/html) # - simple GET on /hello/status_500 (returns HTTP status 500) # - simple GET on /false_uri (returns HTTP status 404) # - largest matching URI handler is picked is already verified because # of /hello and /hello/type_html tests # # # 2. Session Tests # - Sessions + Pipelining basics: # - Create max supported sessions # - On session i, # - send 3 back-to-back POST requests with data i on /adder # - read back 3 responses. They should be i, 2i and 3i # - Tests that # - pipelining works # - per-session context is maintained for all supported # sessions # - Close all sessions # # - Cleanup leftover data: Tests that the web server properly cleans # up leftover data # - Create a session # - POST on /leftover_data with 52 bytes of data (data includes # \r\n)(the handler only # reads first 10 bytes and returns them, leaving the rest of the # bytes unread) # - GET on /hello (should return 'Hello World') # - POST on /false_uri with 52 bytes of data (data includes \r\n) # (should return HTTP 404) # - GET on /hello (should return 'Hello World') # # - Test HTTPd Asynchronous response # - Create a session # - GET on /async_data # - returns 'Hello World!' as a response # - the handler schedules an async response, which generates a second # response 'Hello Double World!' # # - Spillover test # - Create max supported sessions with the web server # - GET /hello on all the sessions (should return Hello World) # - Create one more session, this should fail # - GET /hello on all the sessions (should return Hello World) # # - Timeout test # - Create a session and only Send 'GE' on the same (simulates a # client that left the network halfway through a request) # - Wait for recv-wait-timeout # - Server should automatically close the socket # ############ TODO TESTS ############# # 3. Stress Tests # # - httperf # - Run the following httperf command: # httperf --server=10.31.130.126 --wsess=8,50,0.5 --rate 8 --burst-length 2 # # - The above implies that the test suite will open # - 8 simultaneous connections with the server # - the rate of opening the sessions will be 8 per sec. So in our # case, a new connection will be opened every 0.2 seconds for 1 second # - The burst length 2 indicates that 2 requests will be sent # simultaneously on the same connection in a single go # - 0.5 seconds is the time between sending out 2 bursts # - 50 is the total number of requests that will be sent out # # - So in the above example, the test suite will open 8 # connections, each separated by 0.2 seconds. On each connection # it will send 2 requests in a single burst. The bursts on a # single connection will be separated by 0.5 seconds. A total of # 25 bursts (25 x 2 = 50) will be sent out # 4. Leak Tests # - Simple Leak test # - Simple GET on /hello/restart (returns success, stop web server, measures leaks, restarts webserver) # - Simple GET on /hello/restart_results (returns the leak results) # - Leak test with open sockets # - Open 8 sessions # - Simple GET on /hello/restart (returns success, stop web server, # measures leaks, restarts webserver) # - All sockets should get closed # - Simple GET on /hello/restart_results (returns the leak results) from __future__ import division from __future__ import print_function from builtins import str from builtins import range from builtins import object import threading import socket import time import argparse import http.client import sys import string import random import Utility _verbose_ = False class Session(object): def __init__(self, addr, port, timeout=15): self.client = socket.create_connection((addr, int(port)), timeout=timeout) self.target = addr self.status = 0 self.encoding = '' self.content_type = '' self.content_len = 0 def send_err_check(self, request, data=None): rval = True try: self.client.sendall(request.encode()) if data: self.client.sendall(data.encode()) except socket.error as err: self.client.close() Utility.console_log("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.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.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.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 read_resp_hdrs(self): try: state = 'nothing' resp_read = '' while True: char = self.client.recv(1).decode() if char == '\r' and state == 'nothing': state = 'first_cr' elif char == '\n' and state == 'first_cr': state = 'first_lf' elif char == '\r' and state == 'first_lf': state = 'second_cr' elif char == '\n' and state == 'second_cr': state = 'second_lf' else: state = 'nothing' resp_read += char if state == 'second_lf': break # Handle first line line_hdrs = resp_read.splitlines() line_comp = line_hdrs[0].split() self.status = line_comp[1] del line_hdrs[0] self.encoding = '' self.content_type = '' headers = dict() # Process other headers for h in range(len(line_hdrs)): line_comp = line_hdrs[h].split(':') if line_comp[0] == 'Content-Length': self.content_len = int(line_comp[1]) if line_comp[0] == 'Content-Type': self.content_type = line_comp[1].lstrip() if line_comp[0] == 'Transfer-Encoding': self.encoding = line_comp[1].lstrip() if len(line_comp) == 2: headers[line_comp[0]] = line_comp[1].lstrip() return headers except socket.error as err: self.client.close() Utility.console_log("Socket Error in recv :", err) return None def read_resp_data(self): try: read_data = '' if self.encoding != 'chunked': while len(read_data) != self.content_len: read_data += self.client.recv(self.content_len).decode() else: chunk_data_buf = '' while (True): # Read one character into temp buffer read_ch = self.client.recv(1) # Check CRLF if (read_ch == '\r'): read_ch = self.client.recv(1).decode() if (read_ch == '\n'): # If CRLF decode length of chunk chunk_len = int(chunk_data_buf, 16) # Keep adding to contents self.content_len += chunk_len rem_len = chunk_len while (rem_len): new_data = self.client.recv(rem_len) read_data += new_data rem_len -= len(new_data) chunk_data_buf = '' # Fetch remaining CRLF if self.client.recv(2) != "\r\n": # Error in packet Utility.console_log("Error in chunked data") return None if not chunk_len: # If last chunk break continue chunk_data_buf += '\r' # If not CRLF continue appending # character to chunked data buffer chunk_data_buf += read_ch return read_data except socket.error as err: self.client.close() Utility.console_log("Socket Error in recv :", err) return None def close(self): self.client.close() def test_val(text, expected, received): if expected != received: Utility.console_log(" Fail!") Utility.console_log(" [reason] " + text + ":") Utility.console_log(" expected: " + str(expected)) Utility.console_log(" received: " + str(received)) return False return True class adder_thread (threading.Thread): def __init__(self, id, dut, port): threading.Thread.__init__(self) self.id = id self.dut = dut self.depth = 3 self.session = Session(dut, port) def run(self): self.response = [] # Pipeline 3 requests if (_verbose_): Utility.console_log(" Thread: Using adder start " + str(self.id)) for _ in range(self.depth): self.session.send_post('/adder', str(self.id)) time.sleep(2) for _ in range(self.depth): self.session.read_resp_hdrs() self.response.append(self.session.read_resp_data()) def adder_result(self): if len(self.response) != self.depth: Utility.console_log("Error : missing response packets") return False for i in range(len(self.response)): if not test_val("Thread" + str(self.id) + " response[" + str(i) + "]", str(self.id * (i + 1)), str(self.response[i])): return False return True def close(self): self.session.close() def get_hello(dut, port): # GET /hello should return 'Hello World!' Utility.console_log("[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().decode()): conn.close() return False if not test_val("data", "text/html", resp.getheader('Content-Type')): conn.close() return False Utility.console_log("Success") conn.close() return True def put_hello(dut, port): # PUT /hello returns 405' Utility.console_log("[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 Utility.console_log("Success") conn.close() return True def post_hello(dut, port): # POST /hello returns 405' Utility.console_log("[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 Utility.console_log("Success") conn.close() return True def post_echo(dut, port): # POST /echo echoes data' Utility.console_log("[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().decode()): conn.close() return False Utility.console_log("Success") conn.close() return True def put_echo(dut, port): # PUT /echo echoes data' Utility.console_log("[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().decode()): conn.close() return False Utility.console_log("Success") conn.close() return True def get_echo(dut, port): # GET /echo returns 404' Utility.console_log("[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 Utility.console_log("Success") conn.close() return True def get_hello_type(dut, port): # GET /hello/type_html returns text/html as Content-Type' Utility.console_log("[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().decode()): conn.close() return False if not test_val("data", "text/html", resp.getheader('Content-Type')): conn.close() return False Utility.console_log("Success") conn.close() return True def get_hello_status(dut, port): # GET /hello/status_500 returns status 500' Utility.console_log("[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 Utility.console_log("Success") conn.close() return True def get_false_uri(dut, port): # GET /false_uri returns status 404' Utility.console_log("[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 Utility.console_log("Success") conn.close() return True def parallel_sessions_adder(dut, port, max_sessions): # POSTs on /adder in parallel sessions Utility.console_log("[test] POST {pipelined} on /adder in " + str(max_sessions) + " sessions =>", end=' ') t = [] # Create all sessions for i in range(max_sessions): t.append(adder_thread(i, dut, port)) for i in range(len(t)): t[i].start() for i in range(len(t)): t[i].join() res = True 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): Utility.console_log("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 Utility.console_log("[test] Test HTTPD Work Queue (Async response) =>", end=' ') s = Session(dut, port) s.send_get('/async_data') s.read_resp_hdrs() if not test_val("First Response", "Hello World!", s.read_resp_data()): s.close() return False s.read_resp_hdrs() if not test_val("Second Response", "Hello Double World!", s.read_resp_data()): s.close() return False s.close() Utility.console_log("Success") return True def leftover_data_test(dut, port): # Leftover data in POST is purged (valid and invalid URIs) Utility.console_log("[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().decode()): s.close() return False s.request("GET", url='/hello') resp = s.getresponse() if not test_val("Hello World Data", "Hello World!", resp.read().decode()): s.close() return False s.request("POST", url='/false_uri', body="abcdefghijklmnopqrstuvwxyz\r\nabcdefghijklmnopqrstuvwxyz") resp = s.getresponse() if not test_val("False URI Status", str(404), str(resp.status)): s.close() return False resp.read() s.request("GET", url='/hello') resp = s.getresponse() if not test_val("Hello World Data", "Hello World!", resp.read().decode()): s.close() return False s.close() Utility.console_log("Success") return True def spillover_session(dut, port, max_sess): # Session max_sess_sessions + 1 is rejected Utility.console_log("[test] Session max_sess_sessions (" + str(max_sess) + ") + 1 is rejected =>", end=' ') s = [] _verbose_ = True for i in range(max_sess + 1): if (_verbose_): Utility.console_log("Executing " + str(i)) try: 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().decode()): a.close() break s.append(a) except Exception: if (_verbose_): Utility.console_log("Connection " + str(i) + " rejected") a.close() break # Close open connections for a in s: a.close() # Check if number of connections is equal to max_sess Utility.console_log(["Fail","Success"][len(s) == max_sess]) return (len(s) == max_sess) def recv_timeout_test(dut, port): Utility.console_log("[test] Timeout occurs if partial packet sent =>", end=' ') s = Session(dut, port) 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() Utility.console_log("Success") return True def packet_size_limit_test(dut, port, test_size): Utility.console_log("[test] send size limit test =>", end=' ') retry = 5 while (retry): retry -= 1 Utility.console_log("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", "500", str(resp.status)): Utility.console_log("Data too large to be allocated") test_size = test_size // 10 else: Utility.console_log("Unexpected error") s.close() Utility.console_log("Retry...") continue resp = resp.read().decode() result = (resp == random_data) if not result: test_val("Data size", str(len(random_data)), str(len(resp))) s.close() Utility.console_log("Retry...") continue s.close() Utility.console_log("Success") return True Utility.console_log("Failed") return False def code_500_server_error_test(dut, port): Utility.console_log("[test] 500 Server Error test =>", end=' ') s = Session(dut, port) # Sending a very large content length will cause malloc to fail content_len = 2**31 s.client.sendall(("POST /echo HTTP/1.1\r\nHost: " + dut + "\r\nContent-Length: " + str(content_len) + "\r\n\r\nABCD").encode()) s.read_resp_hdrs() s.read_resp_data() if not test_val("Server Error", "500", s.status): s.close() return False s.close() Utility.console_log("Success") return True def code_501_method_not_impl(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() # Presently server sends back 400 Bad Request # if not test_val("Server Error", "501", s.status): # s.close() # return False if not test_val("Server Error", "400", s.status): s.close() return False s.close() Utility.console_log("Success") return True def code_505_version_not_supported(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() if not test_val("Server Error", "505", s.status): s.close() return False s.close() Utility.console_log("Success") return True def code_400_bad_request(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() if not test_val("Client Error", "400", s.status): s.close() return False s.close() Utility.console_log("Success") return True def code_404_not_found(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() if not test_val("Client Error", "404", s.status): s.close() return False s.close() Utility.console_log("Success") return True def code_405_method_not_allowed(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() if not test_val("Client Error", "405", s.status): s.close() return False s.close() Utility.console_log("Success") return True def code_408_req_timeout(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() if not test_val("Client Error", "408", s.status): s.close() return False s.close() Utility.console_log("Success") return True def code_411_length_required(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() # Presently server sends back 400 Bad Request # if not test_val("Client Error", "411", s.status): # s.close() # return False if not test_val("Client Error", "400", s.status): s.close() return False s.close() Utility.console_log("Success") return True def send_getx_uri_len(dut, port, length): s = Session(dut, port) method = "GET " version = " HTTP/1.1\r\n" path = "/" + "x" * (length - len(method) - len(version) - len("/")) s.client.sendall(method.encode()) time.sleep(1) s.client.sendall(path.encode()) time.sleep(1) s.client.sendall((version + "Host: " + dut + "\r\n\r\n").encode()) s.read_resp_hdrs() s.read_resp_data() s.close() return s.status def code_414_uri_too_long(dut, port, max_uri_len): Utility.console_log("[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 Utility.console_log("Success") return True def send_postx_hdr_len(dut, port, length): s = Session(dut, port) path = "/echo" 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").encode() s.client.sendall(request[:length // 2]) time.sleep(1) s.client.sendall(request[length // 2:]) hdr = s.read_resp_hdrs() resp = s.read_resp_data() s.close() if "Custom" in hdr: return (hdr["Custom"] == custom_hdr_val), resp return False, s.status def code_431_hdr_too_long(dut, port, max_hdr_len): Utility.console_log("[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 Utility.console_log("Success") return True def test_upgrade_not_supported(dut, port): Utility.console_log("[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").encode()) s.read_resp_hdrs() s.read_resp_data() if not test_val("Client Error", "200", s.status): s.close() return False s.close() Utility.console_log("Success") return True if __name__ == '__main__': # Execution begins here... # Configuration # Max number of threads/sessions max_sessions = 7 max_uri_len = 512 max_hdr_len = 512 parser = argparse.ArgumentParser(description='Run HTTPD Test') parser.add_argument('-4','--ipv4', help='IPv4 address') parser.add_argument('-6','--ipv6', help='IPv6 address') parser.add_argument('-p','--port', help='Port') args = vars(parser.parse_args()) dut4 = args['ipv4'] dut6 = args['ipv6'] port = args['port'] dut = dut4 _verbose_ = True Utility.console_log("### Basic HTTP Client Tests") get_hello(dut, port) post_hello(dut, port) put_hello(dut, port) post_echo(dut, port) get_echo(dut, port) put_echo(dut, port) get_hello_type(dut, port) get_hello_status(dut, port) get_false_uri(dut, port) Utility.console_log("### Error code tests") code_500_server_error_test(dut, port) code_501_method_not_impl(dut, port) code_505_version_not_supported(dut, port) code_400_bad_request(dut, port) code_404_not_found(dut, port) code_405_method_not_allowed(dut, port) code_408_req_timeout(dut, port) code_414_uri_too_long(dut, port, max_uri_len) code_431_hdr_too_long(dut, port, max_hdr_len) test_upgrade_not_supported(dut, port) # Not supported yet (Error on chunked request) # code_411_length_required(dut, port) Utility.console_log("### Sessions and Context Tests") parallel_sessions_adder(dut, port, max_sessions) leftover_data_test(dut, port) async_response_test(dut, port) spillover_session(dut, port, max_sessions) recv_timeout_test(dut, port) packet_size_limit_test(dut, port, 50 * 1024) get_hello(dut, port) sys.exit()