diff --git a/CHANGES b/CHANGES index a139ea0..205e37f 100755 --- a/CHANGES +++ b/CHANGES @@ -1,12 +1,26 @@ ------------------------------------------------------------------------ +CHANGE NOTES FOR 0.3.3 (RELEASED: Oct 09, 2025) + +------------------------------------------------------------------------ + +GENERAL CHANGES + +- Added Create Multipart Request Body and Set Multipart Request Header keywords to simplify + building and sending multipart/form-data requests with support for multiple files per + field and optional additional form data. + See: https://github.com/annoviko/robotframework-httpctrl/issues/46 + + +------------------------------------------------------------------------ + CHANGE NOTES FOR 0.3.2 (RELEASED: Sep 11, 2025) ------------------------------------------------------------------------ GENERAL CHANGES -- Add `pyproject.toml` to support modern PEP 517/518 builds. +- Added `pyproject.toml` to support modern PEP 517/518 builds. See: https://github.com/annoviko/robotframework-httpctrl/issues/44 - Removed `setup.py` - old-style installs may not work. diff --git a/LICENSE b/LICENSE index 34a4d3f..e134ceb 100755 --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2018-2022 Andrei Novikov +Copyright (c) 2018-2025 Andrei Novikov Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are diff --git a/src/HttpCtrl/__init__.py b/src/HttpCtrl/__init__.py index 198253c..565bb41 100755 --- a/src/HttpCtrl/__init__.py +++ b/src/HttpCtrl/__init__.py @@ -3,26 +3,31 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ import datetime import http.client +import io import json +import mimetypes +import os import threading +import uuid from robot.api import logger -from HttpCtrl.utils.logger import LoggerAssistant - -from HttpCtrl.internal_messages import IgnoreRequest from HttpCtrl.http_server import HttpServer from HttpCtrl.http_stub import HttpStubContainer, HttpStubCriteria +from HttpCtrl.internal_messages import IgnoreRequest from HttpCtrl.request_storage import RequestStorage -from HttpCtrl.response_storage import ResponseStorage from HttpCtrl.response import Response +from HttpCtrl.response_storage import ResponseStorage +from HttpCtrl.utils.logger import LoggerAssistant + +CRLF = "\r\n" class Client: @@ -116,8 +121,9 @@ def __init__(self): self.__event_queue = threading.Condition() self.__async_queue = {} - - def initialize_client(self, server_host, server_port=None, client_host=None, client_port=0): + def initialize_client( + self, server_host, server_port=None, client_host=None, client_port=0 + ): """ Initialize client using host and port of a server which will be used for communication. @@ -183,14 +189,12 @@ def initialize_client(self, server_host, server_port=None, client_host=None, cli self.__client_host = client_host self.__client_port = client_port - def __get_source_address(self): if self.__client_host is None: return None return self.__client_host, int(self.__client_port) - def __send(self, connection_type, method, url, body): if self.__server_host is None or self.__server_port is None: raise AssertionError("Client is not initialized (host and port are empty).") @@ -198,25 +202,42 @@ def __send(self, connection_type, method, url, body): endpoint = "%s:%s" % (self.__server_host, str(self.__server_port)) source_address = self.__get_source_address() - logger.info("Connect to the server (type: '%s', endpoint: '%s')" % (connection_type, endpoint)) - - if connection_type == 'http': - connection = http.client.HTTPConnection(endpoint, source_address=source_address) - elif connection_type == 'https': - connection = http.client.HTTPSConnection(endpoint, source_address=source_address) + logger.info( + "Connect to the server (type: '%s', endpoint: '%s')" + % (connection_type, endpoint) + ) + + if connection_type == "http": + connection = http.client.HTTPConnection( + endpoint, source_address=source_address + ) + elif connection_type == "https": + connection = http.client.HTTPSConnection( + endpoint, source_address=source_address + ) else: - raise AssertionError("Internal error of the client, please report to " - "'https://github.com/annoviko/robotframework-httpctrl/issues'.") - - logger.info("Send request to the server (method: '%s', url: '%s')." % (method, url)) + raise AssertionError( + "Internal error of the client, please report to " + "'https://github.com/annoviko/robotframework-httpctrl/issues'." + ) + + logger.info( + "Send request to the server (method: '%s', url: '%s')." % (method, url) + ) try: connection.request(method, url, body, self.__request_headers) except Exception as exception: - logger.info("Impossible to send request to the server (reason: '%s')." % str(exception)) + logger.info( + "Impossible to send request to the server (reason: '%s')." + % str(exception) + ) self.__request_headers = {} - logger.info("Request (type: '%s', method '%s') was sent to '%s'." % (connection_type, method, endpoint)) + logger.info( + "Request (type: '%s', method '%s') was sent to '%s'." + % (connection_type, method, endpoint) + ) logger.info("%s %s" % (method, url)) if body is not None: body_to_log = LoggerAssistant.get_body(body) @@ -224,11 +245,10 @@ def __send(self, connection_type, method, url, body): return connection - def __read_body_to_file(self, server_response, filename): logger.info("Write body to file '%s'." % filename) - default_chunk_size = 10000000 # 10 MByte + default_chunk_size = 10000000 # 10 MByte with open(filename, "wb") as file_stream: while True: obtained_chunk = server_response.read(default_chunk_size) @@ -237,7 +257,6 @@ def __read_body_to_file(self, server_response, filename): file_stream.write(obtained_chunk) - def __wait_response(self, connection, read_body_to_file): try: server_response = connection.getresponse() @@ -253,17 +272,18 @@ def __wait_response(self, connection, read_body_to_file): else: self.__response_body_filename = read_body_to_file self.__response_body = None - - self.__read_body_to_file(server_response, read_body_to_file) + self.__read_body_to_file(server_response, read_body_to_file) except Exception as exception: - logger.info("Server has not provided response to the request (reason: %s)." % str(exception)) + logger.info( + "Server has not provided response to the request (reason: %s)." + % str(exception) + ) finally: connection.close() - def __wait_response_async(self, connection, read_body_to_file): try: server_response = connection.getresponse() @@ -285,34 +305,169 @@ def __wait_response_async(self, connection, read_body_to_file): for key, value in headers: headers_dict[key] = value - response_instance = Response(server_response.status, server_response.reason, - response_body, response_body_file, - headers_dict) + response_instance = Response( + server_response.status, + server_response.reason, + response_body, + response_body_file, + headers_dict, + ) self.__async_queue[connection] = response_instance self.__event_queue.notify_all() except Exception as exception: - logger.info("Server has not provided response to the request (reason: %s)." % str(exception)) + logger.info( + "Server has not provided response to the request (reason: %s)." + % str(exception) + ) finally: connection.close() - def __send_request(self, connection_type, method, url, body, read_body_to_file): connection = self.__send(connection_type, method, url, body) self.__wait_response(connection, read_body_to_file) - - def __sent_request_async(self, connection_type, method, url, body, read_body_to_file): + def __sent_request_async( + self, connection_type, method, url, body, read_body_to_file + ): connection = self.__send(connection_type, method, url, body) - wait_thread = threading.Thread(target=self.__wait_response_async, args=(connection, read_body_to_file)) + wait_thread = threading.Thread( + target=self.__wait_response_async, args=(connection, read_body_to_file) + ) wait_thread.daemon = True wait_thread.start() return connection + def __write_multipart_request_data_body(self, body, data, boundary): + for field_name, value in data.items(): + body.write(f"--{boundary}{CRLF}".encode()) + body.write( + f'Content-Disposition: form-data; name="{field_name}"{CRLF}{CRLF}'.encode() + ) + body.write(str(value).encode()) + body.write(CRLF.encode()) + + def create_multipart_request_body(self, files, data=None): + """ + + Create a multipart/form-data body for HTTP requests. + + This helper method builds a multipart request body from provided files + and optional additional form data. It returns both the body as bytes + and the boundary string, which is required to set the Content-Type header. + + + `files` [in] (dict): A dictionary mapping form field names to a list of file paths. + Each file path must exist; otherwise, FileNotFoundError is raised. + Example: {"images": ["cat.jpg", "dog.jpg"], "docs": ["file.pdf"]} + + `data` [in] (dict, optional): Additional form fields to include in the body. + Keys are field names, values are strings or numbers. Defaults to None. + + Returns: + Tuple[bytes, str]: A tuple containing the multipart body as bytes + and the boundary string used in the Content-Type header. + + + Example to create a multipart/form-data body: + + +-------------------------------+---------------+ + | Create Multipart Request Body | ${files dict} | + +-------------------------------+---------------+ + + .. code:: text + + # Prepare file attachments + @{images_list}= Create List cat.jpg dog.jpg + ${files}= Create Dictionary images=@{images_list} + + # Build multipart body and get boundary + ${body} ${boundary}= Create Multipart Request Body ${files} + + # Set multipart Content-Type header + Set Multipart Request Header ${boundary} + + # Send multipart request to your endpoint + Send HTTP Request POST /your-endpoint ${body} + + Example to create a multipart/form-data body with images and PDFs: + + .. code:: text + + # Prepare file attachments + @{images_list}= Create List cat.jpg dog.jpg + @{pdf_list}= Create List report1.pdf report2.pdf + ${files}= Create Dictionary images=@{images_list} documents=@{pdf_list} + + # Build multipart body and get boundary + ${body} ${boundary}= Create Multipart Request Body ${files} + + # Set multipart Content-Type header + Set Multipart Request Header ${boundary} + + # Send multipart request to your endpoint + Send HTTP Request POST /your-endpoint ${body} + + Example to create a multipart/form-data body with additional information: + + +-------------------------------+---------------+--------------+ + | Create Multipart Request Body | ${files dict} | ${data dict} | + +-------------------------------+---------------+--------------+ + + .. code:: text + + # Prepare file attachments + @{images_list}= Create List cat.jpg dog.jpg + ${files}= Create Dictionary images=@{images_list} + + # Prepare additional form data + @{ages_list}= Create List 5 10 + ${data}= Create Dictionary animal_ages=@{ages_list} + + # Build multipart body and get boundary + ${body} ${boundary}= Create Multipart Request Body ${files} ${data} + + # Set multipart Content-Type header + Set Multipart Request Header ${boundary} + + # Send multipart request to your endpoint + Send HTTP Request POST /your-endpoint ${body} + + """ + boundary = "----WebKitFormBoundary" + uuid.uuid4().hex + body = io.BytesIO() + + for field_name, paths in files.items(): + for path in paths: + if os.path.exists(path) is False: + raise FileNotFoundError( + f"File not found: '{path}'. Ensure the file exists and is accessible." + ) + + filename = os.path.basename(path) + mime_type = mimetypes.guess_type(path)[0] or "application/octet-stream" + + with open(path, "rb") as f: + file_data = f.read() + + body.write(f"--{boundary}{CRLF}".encode()) + body.write( + f'Content-Disposition: form-data; name="{field_name}"; filename="{filename}"{CRLF}'.encode() + ) + body.write(f"Content-Type: {mime_type}{CRLF}{CRLF}".encode()) + body.write(file_data) + body.write(CRLF.encode()) + + if data is not None: + self.__write_multipart_request_data_body(body, data, boundary) + + body.write(f"--{boundary}--{CRLF}".encode()) + + return body.getvalue(), boundary def send_http_request(self, method, url, body=None, resp_body_to_file=None): """ @@ -326,7 +481,7 @@ def send_http_request(self, method, url, body=None, resp_body_to_file=None): `body` [in] (string): Body of the request. - `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response + `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response body is writing in RAM. It is useful to write response body into a file when it is expected to be big enough to keep it in the memory. @@ -350,7 +505,7 @@ def send_http_request(self, method, url, body=None, resp_body_to_file=None): ${body}= Set Variable { "message": "Hello World!" } Send HTTP Request POST /post ${body} - + Example where GET request is sent and where response body is written into a file: +-------------------+-----+--------------------+-----------------------------------+ @@ -362,8 +517,7 @@ def send_http_request(self, method, url, body=None, resp_body_to_file=None): Send HTTP Request GET /download_big_file resp_body_to_file=big_archive.tar """ - self.__send_request('http', method, url, body, resp_body_to_file) - + self.__send_request("http", method, url, body, resp_body_to_file) def send_http_request_async(self, method, url, body=None, resp_body_to_file=None): """ @@ -378,7 +532,7 @@ def send_http_request_async(self, method, url, body=None, resp_body_to_file=None `body` [in] (string): Body of the request. - `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response + `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response body is writing in RAM. It is useful to write response body into a file when it is expected to be big enough to keep it in the memory. @@ -403,8 +557,7 @@ def send_http_request_async(self, method, url, body=None, resp_body_to_file=None ${connection}= Send HTTP Request Async GET /download_big_file resp_body_to_file=big_archive.tar """ - return self.__sent_request_async('http', method, url, body, resp_body_to_file) - + return self.__sent_request_async("http", method, url, body, resp_body_to_file) def send_https_request(self, method, url, body=None, resp_body_to_file=None): """ @@ -417,7 +570,7 @@ def send_https_request(self, method, url, body=None, resp_body_to_file=None): `body` [in] (string): Body of the request. - `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response + `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response body is writing in RAM. It is useful to write response body into a file when it is expected to be big enough to keep it in the memory. @@ -433,8 +586,7 @@ def send_https_request(self, method, url, body=None, resp_body_to_file=None): Send HTTPS Request PATCH /patch ${body} """ - self.__send_request('https', method, url, body, resp_body_to_file) - + self.__send_request("https", method, url, body, resp_body_to_file) def send_https_request_async(self, method, url, body=None, resp_body_to_file=None): """ @@ -449,7 +601,7 @@ def send_https_request_async(self, method, url, body=None, resp_body_to_file=Non `body` [in] (string): Body of the request. - `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response + `resp_body_to_file` [in] (string): Path to file where response body should be written. By default is `None` - response body is writing in RAM. It is useful to write response body into a file when it is expected to be big enough to keep it in the memory. @@ -464,14 +616,44 @@ def send_https_request_async(self, method, url, body=None, resp_body_to_file=Non ${connection}= Send HTTPS Request Async DELETE /delete """ - return self.__sent_request_async('https', method, url, body, resp_body_to_file) + return self.__sent_request_async("https", method, url, body, resp_body_to_file) + + def set_multipart_request_header(self, boundary): + """ + + Set the Content-Type header for a multipart/form-data HTTP request. + Should be called before \`Send HTTP Request\` or \`Send HTTPS Request\`. + + This helper method simplifies multipart requests by automatically + constructing and setting the correct Content-Type header value, + including the boundary string used to separate parts in the request body. + + `boundary` [in] (string): The boundary string returned from `create_multipart_request_body`, + used to delimit sections of the multipart form data. + Example where the multipart header is set automatically: - def set_request_header(self, key, value): + +------------------------------+-----------------------------------+ + | Set Multipart Request Header | ${boundary} | + +------------------------------+-----------------------------------+ + + .. code:: text + + ${files}= Create Dictionary images=${["cat.jpg", "dog.jpg"]} + ${data}= Create Dictionary animal_ages=[5, 10] + + ${body} ${boundary}= Create Multipart Request Body ${files} ${data} + Set Multipart Request Header ${boundary} + + """ + header_value = f"multipart/form-data; boundary={boundary}" + self.set_request_header("Content-Type", header_value) + + def set_request_header(self, key: str, value: str): """ - Set HTTP header for request that is going to be sent. Should be called before 'Send HTTP Request' or - 'Send HTTPS Request'. + Set HTTP header for request that is going to be sent. Should be called before \`Send HTTP Request\` or + \`Send HTTPS Request\`. `key` [in] (string): Header name that should be used in the request (be aware of case-sensitive headers). @@ -492,7 +674,10 @@ def set_request_header(self, key, value): """ self.__request_headers[key] = value - + logger.info( + "Header (%s: %s) is added (total headers: %d)." + % (key, value, len(self.__request_headers)) + ) def get_response_status(self): """ @@ -517,14 +702,12 @@ def get_response_status(self): self.__response_status = None return status - def get_response_message(self): with self.__response_guard: message = self.__response_message self.__response_message = None return message - def get_response_headers(self): """ @@ -548,7 +731,6 @@ def get_response_headers(self): self.__response_headers = None return headers - def get_response_body(self): """ @@ -582,7 +764,6 @@ def get_response_body(self): return body - def get_async_response(self, connection, timeout=0): """ @@ -628,8 +809,7 @@ def predicate(): return self.__async_queue.pop(connection, None) - - def get_status_from_response(self, response : Response): + def get_status_from_response(self, response: Response): """ Return response status as an integer value from the specified response object that was obtained by function @@ -657,8 +837,7 @@ def get_status_from_response(self, response : Response): return response.get_status() - - def get_reason_from_response(self, response : Response): + def get_reason_from_response(self, response: Response): """ Return response reason as a string from the specified response object that was obtained by function @@ -687,8 +866,7 @@ def get_reason_from_response(self, response : Response): return response.get_reason() - - def get_headers_from_response(self, response : Response) -> dict: + def get_headers_from_response(self, response: Response) -> dict: """ Return response headers as a dictionary from the specified response object that was obtained by function @@ -716,8 +894,7 @@ def get_headers_from_response(self, response : Response) -> dict: return response.get_headers() - - def get_body_from_response(self, response : Response): + def get_body_from_response(self, response: Response): """ Return response body as a byte array from the specified response object that was obtained by function @@ -746,7 +923,6 @@ def get_body_from_response(self, response : Response): return response.get_body() - class Server: """ @@ -868,11 +1044,9 @@ def __init__(self): self.__server = None self.__thread = None - def __del__(self): self.stop_server() - def start_server(self, host, port): """ @@ -925,16 +1099,17 @@ def start_server(self, host, port): self.stop_server() - logger.info("Prepare HTTP server '%s:%s' and thread to serve it." % (host, port)) + logger.info( + "Prepare HTTP server '%s:%s' and thread to serve it." % (host, port) + ) self.__server = HttpServer(host, int(port)) self.__thread = threading.Thread(target=self.__server.start, args=()) self.__thread.start() - # In case of start-stop, stop may be finished before server start and ir will be impossible to join thread. + # In case of start-stop, stop may be finished before server start and ir will be impossible to join thread. self.__server.wait_run_state() - def stop_server(self): """ @@ -970,7 +1145,6 @@ def stop_server(self): logger.info("HTTP server is stopped.") - def wait_for_request(self, timeout=5): """ @@ -1010,7 +1184,6 @@ def wait_for_request(self, timeout=5): logger.info("Request is received: %s" % self.__request) - def wait_for_no_request(self, timeout=5.0): """ @@ -1049,7 +1222,6 @@ def wait_for_no_request(self, timeout=5.0): logger.info("Request is not received.") - def wait_and_ignore_request(self): """ @@ -1074,11 +1246,10 @@ def wait_and_ignore_request(self): ResponseStorage().push(IgnoreRequest()) logger.info("Request is ignored by closing connection.") - def set_stub_reply(self, method, url, status, body=None): """ - - Sets stub reply for HTTP(S) server. This function sets a server stub to reply automatically by a specific + + Sets stub reply for HTTP(S) server. This function sets a server stub to reply automatically by a specific response to a specific request. When the stub is used to reply, then corresponding statistic is incremented (see \`Get Stub Count\`). @@ -1097,9 +1268,9 @@ def set_stub_reply(self, method, url, status, body=None): +----------------+------+-----------------+-----+ .. code:: text - + Set Stub Reply POST /api/v1/request 200 - + Example how to set stub to reply automatically using a specific body. +----------------+------+-----------------+-----+--------------------------+ @@ -1112,19 +1283,20 @@ def set_stub_reply(self, method, url, status, body=None): """ if self.__server is None: - message_error = "Impossible to set server stub reply (reason: 'server is not created')." + message_error = ( + "Impossible to set server stub reply (reason: 'server is not created')." + ) raise AssertionError(message_error) - + criteria = HttpStubCriteria(method=method, url=url) response = Response(int(status), None, body, None, None) HttpStubContainer().add(criteria, response) - def get_stub_count(self, method, url): """ - + Returns server stub statistic that defines how many time the stub was used by server to reply. - + `method` [in] (string): Request method that is used to handle by server stub (GET, POST, DELETE, etc., see: RFC 7231, RFC 5789). `url` [in] (string): Path to the resource that is used by server stub, for example, in case address www.httpbin.org/ip - '/ip' is an path. @@ -1157,7 +1329,6 @@ def get_stub_count(self, method, url): criteria = HttpStubCriteria(method=method, url=url) return HttpStubContainer().count(criteria) - def get_request_source_address(self): """ @@ -1177,7 +1348,6 @@ def get_request_source_address(self): """ return self.__request.get_source_address() - def get_request_source_port(self): """ @@ -1197,7 +1367,6 @@ def get_request_source_port(self): """ return str(self.__request.get_source_port()) - def get_request_source_port_as_integer(self): """ @@ -1217,7 +1386,6 @@ def get_request_source_port_as_integer(self): """ return self.__request.get_source_port() - def get_request_method(self): """ @@ -1237,7 +1405,6 @@ def get_request_method(self): """ return self.__request.get_method() - def get_request_body(self): """ @@ -1257,7 +1424,6 @@ def get_request_body(self): """ return self.__request.get_body() - def get_request_headers(self): """ @@ -1277,7 +1443,6 @@ def get_request_headers(self): """ return self.__request.get_headers() - def get_request_url(self): """ @@ -1297,7 +1462,6 @@ def get_request_url(self): """ return self.__request.get_url() - def set_reply_header(self, key, value): """ @@ -1334,7 +1498,6 @@ def set_reply_header(self, key, value): """ self.__response_headers[key] = value - def reply_by(self, status, body=None): """ @@ -1484,7 +1647,7 @@ def get_json_value_from_string(json_string, path): """ json_content = json.loads(json_string) - keys = path.split('/') + keys = path.split("/") current_element = json_content for key in keys: @@ -1495,7 +1658,6 @@ def get_json_value_from_string(json_string, path): return current_element - @staticmethod def set_json_value_in_string(json_string, path, value): """ @@ -1543,7 +1705,7 @@ def set_json_value_in_string(json_string, path, value): """ json_content = json.loads(json_string) - keys = path.split('/') + keys = path.split("/") current_element = json_content for key in keys: @@ -1561,7 +1723,6 @@ def set_json_value_in_string(json_string, path, value): return json.dumps(json_content) - class Logging: """ diff --git a/src/HttpCtrl/http_handler.py b/src/HttpCtrl/http_handler.py index 1d1c3b6..aac8f73 100755 --- a/src/HttpCtrl/http_handler.py +++ b/src/HttpCtrl/http_handler.py @@ -3,19 +3,20 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ from http.server import SimpleHTTPRequestHandler + from robot.api import logger -from HttpCtrl.internal_messages import TerminationRequest, IgnoreRequest +from HttpCtrl.http_stub import HttpStubContainer, HttpStubCriteria +from HttpCtrl.internal_messages import IgnoreRequest, TerminationRequest from HttpCtrl.request import Request from HttpCtrl.request_storage import RequestStorage from HttpCtrl.response_storage import ResponseStorage -from HttpCtrl.http_stub import HttpStubContainer, HttpStubCriteria class HttpHandler(SimpleHTTPRequestHandler): @@ -25,55 +26,43 @@ def __init__(self, *args, **kwargs): SimpleHTTPRequestHandler.__init__(self, *args, **kwargs) - def do_GET(self): - self.__default_handler('GET') - + self.__default_handler("GET") def do_POST(self): - self.__default_handler('POST') - + self.__default_handler("POST") def do_PUT(self): - self.__default_handler('PUT') - + self.__default_handler("PUT") def do_OPTIONS(self): - self.__default_handler('OPTIONS') - + self.__default_handler("OPTIONS") def do_HEAD(self): - self.__default_handler('HEAD') - + self.__default_handler("HEAD") def do_PATCH(self): - self.__default_handler('PATCH') - + self.__default_handler("PATCH") def do_DELETE(self): - self.__default_handler('DELETE') - + self.__default_handler("DELETE") def log_message(self, format, *args): return - def log_error(self, format, *args): return - - def log_request(self, code='-', size='-'): + def log_request(self, code="-", size="-"): return - def __extract_body(self): - body_length = int(self.headers.get('Content-Length', 0)) + body_length = int(self.headers.get("Content-Length", 0)) if body_length > 0: return self.rfile.read(body_length) return None - def __default_handler(self, method): host, port = self.client_address[:2] body = self.__extract_body() @@ -89,14 +78,17 @@ def __default_handler(self, method): RequestStorage().push(request) response = ResponseStorage().pop() - if isinstance(response, TerminationRequest) or isinstance(response, IgnoreRequest): + if isinstance(response, TerminationRequest) or isinstance( + response, IgnoreRequest + ): return try: self.__send_response(response) except Exception as exception: - logger.info("Response was not sent to client due to reason: '%s'." % str(exception)) - + logger.info( + "Response was not sent to client due to reason: '%s'." % str(exception) + ) def __send_response(self, response): if response is None: @@ -115,7 +107,7 @@ def __send_response(self, response): if isinstance(response.get_body(), str): body = response.get_body().encode("utf-8") - self.send_header('Content-Length', str(len(body))) + self.send_header("Content-Length", str(len(body))) self.end_headers() diff --git a/src/HttpCtrl/http_server.py b/src/HttpCtrl/http_server.py index d75b154..e141e4f 100755 --- a/src/HttpCtrl/http_server.py +++ b/src/HttpCtrl/http_server.py @@ -3,7 +3,7 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ @@ -11,7 +11,6 @@ import ipaddress import socket import threading - from socketserver import TCPServer from robot.api import logger @@ -36,12 +35,10 @@ def __init__(self, host, port): self.__is_run_state = False self.__cv_run = threading.Condition() - def __del__(self): if self.__server is not None: self.stop() - def start(self): self.__handler = HttpHandler self.__server = self.__create_tcp_server() @@ -58,13 +55,11 @@ def start(self): self.stop() raise exception - def wait_run_state(self): with self.__cv_run: while not self.__is_run_state: self.__cv_run.wait() - def stop(self): if self.__server is not None: ResponseStorage().push(TerminationRequest()) @@ -76,7 +71,6 @@ def stop(self): with self.__cv_run: self.__is_run_state = False - def __create_tcp_server(self): tcp_server = self.__create_ipv6_tcp_server() if tcp_server is not None: @@ -84,25 +78,31 @@ def __create_tcp_server(self): return self.__create_ipv4_tcp_server() - def __create_ipv6_tcp_server(self): try: TCPServerIPv6.allow_reuse_address = True - ipaddress.IPv6Address(self.__host) # if throws exception then address is not IPv6 + ipaddress.IPv6Address( + self.__host + ) # if throws exception then address is not IPv6 tcp_server = TCPServerIPv6((self.__host, self.__port), self.__handler) - logger.info("IPv6 TCP server '%s:%s' is created for HTTP." % (self.__host, str(self.__port))) + logger.info( + "IPv6 TCP server '%s:%s' is created for HTTP." + % (self.__host, str(self.__port)) + ) return tcp_server except: return None - def __create_ipv4_tcp_server(self): TCPServer.allow_reuse_address = True tcp_server = TCPServer((self.__host, self.__port), self.__handler) - logger.info("IPv4 TCP server '%s:%s' is created for HTTP." % (self.__host, str(self.__port))) + logger.info( + "IPv4 TCP server '%s:%s' is created for HTTP." + % (self.__host, str(self.__port)) + ) return tcp_server diff --git a/src/HttpCtrl/http_stub.py b/src/HttpCtrl/http_stub.py index debf132..4de481f 100755 --- a/src/HttpCtrl/http_stub.py +++ b/src/HttpCtrl/http_stub.py @@ -3,7 +3,7 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ @@ -15,15 +15,14 @@ class HttpStubCriteria: def __init__(self, **kwargs): - self.method = kwargs.get('method', None) + self.method = kwargs.get("method", None) if self.method is not None: self.method = self.method.upper() - self.url = kwargs.get('url', None) + self.url = kwargs.get("url", None) if self.url is not None: self.url = self.url.lower() - def __eq__(self, other): return (self.method == other.method) and (self.url == other.url) @@ -40,20 +39,17 @@ def __init__(self): self.__stubs = [] self.__lock = Lock() - def add(self, criteria, response): with self.__lock: self.__stubs.append(HttpStub(criteria, response)) - def count(self, criteria): with self.__lock: for stub in self.__stubs: if self.__is_satisfy(stub, criteria) is True: return stub.count - - return 0 + return 0 def get(self, criteria): with self.__lock: @@ -61,18 +57,15 @@ def get(self, criteria): if self.__is_satisfy(stub, criteria) is True: stub.count += 1 return stub - - return None + return None def clear(self): with self.__lock: self.__stubs.clear() - def __is_satisfy(self, stub, criteria): if stub.criteria == criteria: return True - + return False - \ No newline at end of file diff --git a/src/HttpCtrl/internal_messages.py b/src/HttpCtrl/internal_messages.py index 367aabe..751667a 100755 --- a/src/HttpCtrl/internal_messages.py +++ b/src/HttpCtrl/internal_messages.py @@ -3,7 +3,7 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ diff --git a/src/HttpCtrl/request.py b/src/HttpCtrl/request.py index ec199b0..66d9cff 100755 --- a/src/HttpCtrl/request.py +++ b/src/HttpCtrl/request.py @@ -3,12 +3,11 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ - from HttpCtrl.utils.logger import LoggerAssistant @@ -22,7 +21,14 @@ def __init__(self, host, port, method, url, headers, body=None): self.__headers = headers def __copy__(self): - return Request(self.__source_host, self.__source_port, self.__method, self.__url, self.__headers, self.__body) + return Request( + self.__source_host, + self.__source_port, + self.__method, + self.__url, + self.__headers, + self.__body, + ) def __str__(self): body_to_log = LoggerAssistant.get_body(self.__body) diff --git a/src/HttpCtrl/request_storage.py b/src/HttpCtrl/request_storage.py index dde07b7..9db0c3b 100755 --- a/src/HttpCtrl/request_storage.py +++ b/src/HttpCtrl/request_storage.py @@ -3,7 +3,7 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ @@ -20,18 +20,15 @@ def __init__(self): self.__request = None self.__event_incoming = threading.Condition() - def __ready(self): return self.__request is not None - def push(self, request): with self.__event_incoming: logger.info("Push request to the Request Storage: %s" % request) self.__request = request self.__event_incoming.notify() - def pop(self, timeout=5.0): with self.__event_incoming: if not self.__ready(): @@ -43,7 +40,6 @@ def pop(self, timeout=5.0): self.__request = None return request - def clear(self): with self.__event_incoming: self.__request = None diff --git a/src/HttpCtrl/response.py b/src/HttpCtrl/response.py index 57eb94c..3e0af1e 100755 --- a/src/HttpCtrl/response.py +++ b/src/HttpCtrl/response.py @@ -3,17 +3,16 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ - from HttpCtrl.utils.logger import LoggerAssistant class Response: - def __init__(self, status, reason, body, body_file, headers : dict): + def __init__(self, status, reason, body, body_file, headers: dict): self.__status = status self.__reason = reason self.__body = body @@ -25,13 +24,18 @@ def __str__(self): if self.__body_file is None: return str(self.__status) else: - return "%s\n" % (self.__status, self.__body_file) + return "%s\n" % ( + self.__status, + self.__body_file, + ) body_to_log = LoggerAssistant.get_body(self.__body) return "%s\n%s" % (str(self.__status), body_to_log) def __copy__(self): - return Response(self.__status, self.__reason, self.__body, self.__body_file, self.__headers) + return Response( + self.__status, self.__reason, self.__body, self.__body_file, self.__headers + ) def get_status(self): return self.__status @@ -46,7 +50,7 @@ def get_body(self): if (self.__body is None) and (self.__body_file is not None): with open(self.__body_file) as file_stream: return file_stream.read() - + return self.__body def get_headers(self) -> dict: diff --git a/src/HttpCtrl/response_storage.py b/src/HttpCtrl/response_storage.py index d33c9bd..19731a9 100755 --- a/src/HttpCtrl/response_storage.py +++ b/src/HttpCtrl/response_storage.py @@ -3,7 +3,7 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ @@ -20,18 +20,15 @@ def __init__(self): self.__response = None self.__event_incoming = threading.Condition() - def __ready(self): return self.__response is not None - def push(self, response): with self.__event_incoming: logger.info("Push response to the Response Storage: %s" % response) self.__response = response self.__event_incoming.notify() - def pop(self, timeout=5.0): with self.__event_incoming: if not self.__ready(): @@ -43,7 +40,6 @@ def pop(self, timeout=5.0): self.__response = None return response - def clear(self): with self.__event_incoming: self.__response = None diff --git a/src/HttpCtrl/utils/__init__.py b/src/HttpCtrl/utils/__init__.py index 81e5d96..42d10db 100755 --- a/src/HttpCtrl/utils/__init__.py +++ b/src/HttpCtrl/utils/__init__.py @@ -3,7 +3,7 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ diff --git a/src/HttpCtrl/utils/logger.py b/src/HttpCtrl/utils/logger.py index 7ceb4c1..134141d 100755 --- a/src/HttpCtrl/utils/logger.py +++ b/src/HttpCtrl/utils/logger.py @@ -3,19 +3,17 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ - from robot.api import logger class LoggerAssistant: __MAX_BODY_SIZE_TO_LOG = 512 - @staticmethod def set_body_size(body_size): if body_size is None: @@ -25,13 +23,18 @@ def set_body_size(body_size): LoggerAssistant.__MAX_BODY_SIZE_TO_LOG = body_size - @staticmethod def get_body(body): if body is not None: - if (LoggerAssistant.__MAX_BODY_SIZE_TO_LOG is None) or (len(body) < LoggerAssistant.__MAX_BODY_SIZE_TO_LOG): + if (LoggerAssistant.__MAX_BODY_SIZE_TO_LOG is None) or ( + len(body) < LoggerAssistant.__MAX_BODY_SIZE_TO_LOG + ): return body else: - return "%s...\n...\n" % (body[:LoggerAssistant.__MAX_BODY_SIZE_TO_LOG], LoggerAssistant.__MAX_BODY_SIZE_TO_LOG, len(body)) + return "%s...\n...\n" % ( + body[: LoggerAssistant.__MAX_BODY_SIZE_TO_LOG], + LoggerAssistant.__MAX_BODY_SIZE_TO_LOG, + len(body), + ) return "" diff --git a/src/HttpCtrl/utils/singleton.py b/src/HttpCtrl/utils/singleton.py index 9f89e28..0adea1c 100755 --- a/src/HttpCtrl/utils/singleton.py +++ b/src/HttpCtrl/utils/singleton.py @@ -3,15 +3,17 @@ HttpCtrl library provides HTTP/HTTPS client and server API to Robot Framework to make REST API testing easy. Authors: Andrei Novikov -Date: 2018-2022 +Date: 2018-2025 Copyright: The 3-Clause BSD License """ + class Singleton(type): _instances = {} + def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) - return cls._instances[cls] \ No newline at end of file + return cls._instances[cls]