diff --git a/api_auto_generator/Endpoint_Generator.md b/api_auto_generator/Endpoint_Generator.md new file mode 100644 index 00000000..1c7ff11a --- /dev/null +++ b/api_auto_generator/Endpoint_Generator.md @@ -0,0 +1,36 @@ +# Auto Generate Endpoint modules for API Automation Framework # +The Endpoint generator project helps automate creating API automation tests using Qxf2's API Automation framework. It generates Endpoint modules - an abstraction for endpoints in the application under test from an OpenAPI specification. + +## Requirements ## +- An V3.x.x OpenAPI specification for your API app. +- The spec file can be a `JSON` or `YAML` file. + +## How to run the script? ## +- Validate the OpenAPI specification +``` +python api_auto_generator/endpoint_module_generator --spec +``` +This command will help check if the OpenAPI spec can be used to generate Endpoints file. It will raise an exception for invalid or incomplete specs. +- Generate the `Endpoint` module +``` +python api_auto_generator/endpoint_module_generator --spec --generate-endpoints +``` +This command will generate `_Endpoint.py` module in the `endpoints` dir. + +## How does the script work? ## +- The script uses `openapi3_parser` module to parse and read the contents from an OpenAPI spec. +- The endpoints and its details are read from the spec +- A module-name, class-name, instance-method-names are all generated for the endpoint +- The Path & Query parameters to be passed to the Endpoint class is generated +- The json/data params to be passed to the requests method is generated from the request body +- A Python dictionary collecting all these values is generated +- The generated Python dictionary is redered on a Jinja2 template + +## Limitations/Constraints on using the Generate Endpoint script ## + +### Invalid OpenAPI spec ### +- The Generate Endpoint script validates the OpenAPI spec at the start of the execution, using an invalid spec triggers an exception +- The JSON Schema validation is also done in step #1, but the exception raised regarding a JSON Schema error can sometimes be a little confusing, in such cases replace the failing schema with {} to proceed to generate Endpoint files + +### Minimal spec ### +- When using a minimal spec, to check if Endpoint files can be generated from it, run the script using --spec CLI param alone, you can proceed to use --generate-endpoint param if no issue was seen with the previous step \ No newline at end of file diff --git a/api_auto_generator/endpoint_module_generator.py b/api_auto_generator/endpoint_module_generator.py new file mode 100644 index 00000000..aa8affb4 --- /dev/null +++ b/api_auto_generator/endpoint_module_generator.py @@ -0,0 +1,100 @@ +""" +What does this module do? +- It creates an Endpoint file with a class from the OpenAPI spec +- The path key in the spec is translated to an Endpoint +- The operations(http methods) for a path is translated to instance methods for the Endpoint +- The parameters for operations are translated to function parameters for the instance methods +- HTTP Basic, HTTP Bearer and API Keys Auth are currently supported by this module & should passed +through headers +""" + + +from argparse import ArgumentParser +from pathlib import Path +from jinja2 import FileSystemLoader, Environment +from jinja2.exceptions import TemplateNotFound +from loguru import logger +from openapi_spec_parser import OpenAPISpecParser + + +# pylint: disable=line-too-long +# Get the template file location & endpoint destination location relative to this script +ENDPOINT_TEMPLATE_NAME = Path(__file__).parent.joinpath("templates").joinpath("endpoint_template.jinja2") # <- Jinja2 template needs to be on the same directory as this script +ENDPOINT_DESTINATION_DIR = Path(__file__).parent.parent.joinpath("endpoints") # <- The Endpoint files are created in the endpoints dir in the project root + + +class EndpointGenerator(): + """ + A class to Generate Endpoint module using Jinja2 template + """ + + + def __init__(self, logger_obj: logger): + """ + Initialize Endpoint Generator class + """ + self.endpoint_template_filename = ENDPOINT_TEMPLATE_NAME.name + self.jinja_template_dir = ENDPOINT_TEMPLATE_NAME.parent.absolute() + self.logger = logger_obj + self.jinja_environment = Environment(loader=FileSystemLoader(self.jinja_template_dir), + autoescape=True) + + + def endpoint_class_content_generator(self, + endpoint_class_name: str, + endpoint_class_content: dict) -> str: + """ + Create Jinja2 template content + """ + content = None + template = self.jinja_environment.get_template(self.endpoint_template_filename) + content = template.render(class_name=endpoint_class_name, class_content=endpoint_class_content) + self.logger.info(f"Rendered content for {endpoint_class_name} class using Jinja2 template") + return content + + + def generate_endpoint_file(self, + endpoint_filename: str, + endpoint_class_name: str, + endpoint_class_content: dict): + """ + Create an Endpoint file + """ + try: + endpoint_filename = ENDPOINT_DESTINATION_DIR.joinpath(endpoint_filename+'.py') + endpoint_content = self.endpoint_class_content_generator(endpoint_class_name, + endpoint_class_content) + with open(endpoint_filename, 'w', encoding='utf-8') as endpoint_f: + endpoint_f.write(endpoint_content) + except TemplateNotFound: + self.logger.error(f"Unable to find {ENDPOINT_TEMPLATE_NAME.absolute()}") + except Exception as endpoint_creation_err: + self.logger.error(f"Unable to generate Endpoint file - {endpoint_filename} due to {endpoint_creation_err}") + else: + self.logger.success(f"Successfully generated Endpoint file - {endpoint_filename.name}") + + +if __name__ == "__main__": + arg_parser = ArgumentParser(prog="GenerateEndpointFile", + description="Generate Endpoint.py file from OpenAPI spec") + arg_parser.add_argument("--spec", + dest="spec_file", + required=True, + help="Pass the location to the OpenAPI spec file, Passing this param alone will run a dry run of endpoint content generation with actually creating the endpoint") + arg_parser.add_argument("--generate-endpoints", + dest='if_generate_endpoints', + action='store_true', + help="This param will create _endpoint.py file for Path objects from the OpenAPI spec") + + args = arg_parser.parse_args() + try: + parser = OpenAPISpecParser(args.spec_file, logger) + if args.if_generate_endpoints: + endpoint_generator = EndpointGenerator(logger) + for module_name, file_content in parser.parsed_dict.items(): + for class_name, class_content in file_content.items(): + endpoint_generator.generate_endpoint_file(module_name, + class_name, + class_content) + except Exception as ep_generation_err: + raise ep_generation_err diff --git a/api_auto_generator/endpoint_name_generator.py b/api_auto_generator/endpoint_name_generator.py new file mode 100644 index 00000000..05124a7d --- /dev/null +++ b/api_auto_generator/endpoint_name_generator.py @@ -0,0 +1,107 @@ +""" +Module to generate: + 1. Module name + 2. Class name + 3. Method name +""" + + +import re +from typing import Union +from packaging.version import Version, InvalidVersion + + +class NameGenerator(): + "Base class for generating names" + + + def __init__(self, + endpoint_url: str, + if_query_param: bool, + path_params: list, + requestbody_type: str): + "Init NameGen object" + self.endpoint_split, self.api_version_num = self.split_endpoint_string(endpoint_url) + self.common_base = self.endpoint_split[0] + self.endpoints_in_a_file = [ ep for ep in re.split("-|_", self.common_base)] + self.if_query_param = if_query_param + self.path_params = path_params + self.requestbody_type = requestbody_type + + + @property + def module_name(self) -> str : + "Module name for an Endpoint" + return "_" + "_".join(self.endpoints_in_a_file) + "_" + "endpoint" + + + @property + def class_name(self) -> str : + "Class name for Endpoint" + capitalized_endpoints_in_a_file = [ ep.capitalize() for ep in self.endpoints_in_a_file] + return "".join(capitalized_endpoints_in_a_file) + "Endpoint" + + + @property + def url_method_name(self) -> str : + "URL method name for endpoint" + return self.common_base.lower().replace('-', '_') + "_" + "url" + + + @property + def base_api_param_string(self) -> str : + "Base API method parameter string" + param_string = "" + if self.if_query_param: + param_string += ", params=params" + if self.requestbody_type == "json": + param_string += ", json=json" + if self.requestbody_type == "data": + param_string += ", data=data" + param_string += ", headers=headers" + return param_string + + + @property + def instance_method_param_string(self) -> str : + "Instance method parameter string" + param_string = "self" + if self.if_query_param: + param_string += ", params" + for param in self.path_params: + param_string += f", {param[0]}" + if self.requestbody_type == "json": + param_string += ", json" + if self.requestbody_type == "data": + param_string += ", data" + param_string += ', headers' + return param_string + + + def get_instance_method_name(self, http_method: str) -> str : + "Generate Instance method name" + endpoint_split = [ ep.lower().replace('-','_') for ep in self.endpoint_split ] + return http_method + "_" + "_".join(endpoint_split) + + + def split_endpoint_string(self, endpoint_url: str) -> tuple[list[str], Union[str,None]]: + """ + Split the text in the endpoint, clean it up & return a list of text + """ + version_num = None + if endpoint_url == "/": # <- if the endpoint is only / + endpoint_split = ["home_base"] # <- make it /home_base (it needs to be unique) + else: + endpoint_split = endpoint_url.split("/") + # remove {} from path paramters in endpoints + endpoint_split = [ re.sub("{|}","",text) for text in endpoint_split if text ] + for split_values in endpoint_split: + try: + if_api_version = Version(split_values) # <- check if version number present + version_num = [ str(num) for num in if_api_version.release ] + version_num = '_'.join(version_num) + endpoint_split.remove(split_values) + except InvalidVersion: + if split_values == "api": + endpoint_split.remove(split_values) + return (endpoint_split, version_num,) diff --git a/api_auto_generator/openapi_spec_parser.py b/api_auto_generator/openapi_spec_parser.py new file mode 100644 index 00000000..00d801c0 --- /dev/null +++ b/api_auto_generator/openapi_spec_parser.py @@ -0,0 +1,281 @@ +""" +OpenAPI specification Parser +""" +# pylint: disable=locally-disabled, multiple-statements, fixme, line-too-long +# pylint: disable=too-many-nested-blocks + + +from typing import Union, TextIO +from openapi_parser import parse, specification +from openapi_spec_validator.readers import read_from_filename +from openapi_spec_validator import validate_spec +import openapi_spec_validator as osv +from endpoint_name_generator import NameGenerator + + +# pylint: disable=too-many-instance-attributes +# pylint: disable=broad-except +class OpenAPIPathParser(): + "OpenAPI Path object parser" + + + def __init__(self, path: specification.Path, logger_obj): + "Init the instance" + self.module_name = None + self.class_name = None + self.url_method_name = None + self.instance_methods = [] + self.path_dict = {} + self.path = path + self.operations = self.path.operations + self.logger = logger_obj + + # parameters can be in two places: + # 1. path.parameter + # 2. path.operation.parameter + # move all parameters to #2 + for operation in self.operations: + try: + if self.path.parameters: + operation.parameters.append(*path.parameters) + + # Parse operations(HTTP Methods) to get Endpoint instance method details + instance_method_details = {} # Dict to collect all instance methods for a HTTP Method + parsed_parameters = {} # Dict to collect: 1.Query, 2.Path & 3.RequestBody params + + # Parse Query & Path parameters + q_params, p_params = self.parse_parameters(operation.parameters) + parsed_parameters['query_params'] = q_params + parsed_parameters['path_params'] = p_params + + # Parse RequestBody parameters + rb_type = None + rb_param = None + con_schma_type = None + if operation.request_body: + rb_type, rb_param, con_schma_type = self.parse_request_body(operation.request_body) + if rb_type == "json": + parsed_parameters['json_params'] = rb_param + elif rb_type == "data": + parsed_parameters['data_params'] = rb_param + parsed_parameters['content_schema_type'] = con_schma_type + + # Generate: 1.Module, 2.Class, 3.url_method_name, 4.base_api_param_string, + # 5.instance_method_param_string, 6.instance_method_name name using NameGenerator Obj + name_gen_obj = NameGenerator(path.url, + bool(q_params), + p_params, + rb_type) + self.module_name = name_gen_obj.module_name + self.class_name = name_gen_obj.class_name + self.url_method_name = name_gen_obj.url_method_name + base_api_param_string = name_gen_obj.base_api_param_string + instance_method_param_string = name_gen_obj.instance_method_param_string + instance_method_name = name_gen_obj.get_instance_method_name(operation.method.name.lower()) + + # Collect the Endpoint instance method details + instance_method_details[instance_method_name] = {'params':parsed_parameters, + 'base_api_param_string':base_api_param_string, + 'instance_method_param_string': instance_method_param_string, + 'http_method': operation.method.name.lower(), + 'endpoint':self.path.url} + self.instance_methods.append(instance_method_details) + self.logger.info(f"Parsed {operation.method.name} for {self.path.url}") + except Exception as failed_to_parse_err: + self.logger.debug(f"Failed to parse {operation.method.name} for {self.path.url} due to {failed_to_parse_err}, skipping it") + continue + + + # pylint: disable=inconsistent-return-statements + def get_function_param_type(self, type_str: str) -> Union[str, None]: + "Translate the datatype in spec to corresponding Python type" + if type_str.lower() == 'boolean': + return 'bool' + if type_str.lower() == 'integer': + return 'int' + if type_str.lower() == 'number': + return 'float' + if type_str.lower() == 'string': + return 'str' + if type_str.lower() == 'array': + return 'list' + if type_str.lower() == 'object': + return 'dict' + + + def parse_parameters(self, parameters: list[specification.Parameter]) -> tuple[list, list]: + """ + Create class parameters for Endpoint module + This function will parse: + 1. Query Parameters + 2. Path Parameters + """ + query_params = [] + path_params = [] + + # Loop through list[specification.Parameter] to identify: 1.Query, 2.Path params + for parameter in parameters: + if parameter.location.name.lower() == "path": + name = parameter.name + param_type = self.get_function_param_type(parameter.schema.type.name) + if (name, param_type,) not in path_params: + path_params.append((name, param_type)) + elif parameter.location.name.lower() == "query": + name = parameter.name + param_type = self.get_function_param_type(parameter.schema.type.name) + if (name, param_type,) not in query_params: + query_params.append((name, param_type)) + return (query_params, path_params,) + + + def get_name_type_nested_prop(self, prop) -> list: + "Get the name & type for nested property" + nested_param_list = [] + for nested_prop in prop.schema.properties: + nested_name = nested_prop.name + nested_param_type = self.get_function_param_type(nested_prop.schema.type.name) + nested_param_list.append(nested_name, nested_param_type) + return nested_param_list + + # pylint: disable=too-many-branches, too-complex + def parse_request_body(self, request_body: specification.RequestBody) -> tuple[str, list, str]: + """ + Parse the requestBody from the spec and return a list of json & data params + This function will parse dict inside a JSON/Form param to only one level only + i.e this function will identify another_dict in this example: + json_param = { + another_dict: { + 'nested_key': 'nested_value'}, + 'key':'value + } + but will not identify nested_dict in: + json_param = { + another_dict:{ + 'nested_dict':{ + 'nested_key': 'nested_value'} + } + }, + 'key':'value + } + """ + requestbody_type = None + requestbody_param = [] + content_schema_type = None + + # Parse the request_body in the spec + # Identify the content type of the request_body + # This module currently supports: 1.json, 2.form content types + for content in request_body.content: + if content.type.name.lower() == "json": + if content.schema.type.name.lower() == 'object': + for prop in content.schema.properties: + name = prop.name + requestbody_type = self.get_function_param_type(prop.schema.type.name) + if requestbody_type == 'dict': + nested_dict_param = {} + nested_param_list = self.get_name_type_nested_prop(prop) + nested_dict_param[name] = nested_param_list + requestbody_param.append(nested_dict_param) + else: + requestbody_param.append((name, requestbody_type)) + requestbody_type = "json" + if content.schema.type.name.lower() == 'array': + for prop in content.schema.items.properties: + name = prop.name + requestbody_type = self.get_function_param_type(prop.schema.type.name) + if requestbody_type == 'dict': + nested_dict_param = {} + nested_param_list = self.get_name_type_nested_prop(prop) + nested_dict_param[name] = nested_param_list + requestbody_param.append(nested_dict_param) + else: + requestbody_param.append((name, requestbody_type)) + requestbody_type = "json" + content_schema_type = content.schema.type.name.lower() + if content.type.name.lower() == "form": + if content.schema.type.name.lower() == 'object': + for prop in content.schema.properties: + name = prop.name + requestbody_type = self.get_function_param_type(prop.schema.type.name) + if requestbody_type == 'dict': + nested_dict_param = {} + nested_param_list = self.get_name_type_nested_prop(prop) + nested_dict_param[name] = nested_param_list + requestbody_param.append(nested_dict_param) + else: + requestbody_param.append((name, requestbody_type)) + requestbody_type = "data" + content_schema_type = content.schema.type.name.lower() + return (requestbody_type, requestbody_param, content_schema_type,) + + +class OpenAPISpecParser(): + "OpenAPI Specification Parser Object" + + + # pylint: disable=too-few-public-methods + def __init__(self, spec_file: TextIO, logger_obj) -> None: + "Init Spec Parser Obj" + + self.logger = logger_obj + # Generate Final dict usable against a Jinja2 template from the OpenAPI Spec + self._fdict = {} + """ + _fdict structure: + { + module_name1:{ + class_name1:{ + instance_methods: [], + url_method_name: str + } + } + module_name2:{ + class_name1:{ + instance_methods: [], + url_method_name: str + } + } + } + """ + try: # <- Outer level try-catch to prevent exception chaining + spec_dict, _ = read_from_filename(spec_file) + validate_spec(spec_dict) + self.logger.success(f"Successfully validated spec file - {spec_file}") + try: + self.parsed_spec = parse(spec_file) + # Loop through all paths and parse them using OpenAPIPathParser obj + # Collect the: 1.Module, 2.Class, 3.url_method_name, + # 4.base_api_param_string, 5.instance_method_param_string, + # 6.instance_method_name name + for path in self.parsed_spec.paths: + p_path = OpenAPIPathParser(path, logger_obj) + if p_path.module_name: + if self._fdict.get(p_path.module_name): + if self._fdict[p_path.module_name].get(p_path.class_name): + if self._fdict[p_path.module_name][p_path.class_name].get('instance_methods'): + for instance_method in p_path.instance_methods: + self._fdict[p_path.module_name][p_path.class_name]['instance_methods'].append(instance_method) + else: + self._fdict[p_path.module_name][p_path.class_name]= {'instance_methods': p_path.instance_methods} + else: + self._fdict[p_path.module_name][p_path.class_name]={'instance_methods': p_path.instance_methods} + + else: + self._fdict[p_path.module_name]= {p_path.class_name:{'instance_methods': p_path.instance_methods}} + self._fdict[p_path.module_name][p_path.class_name]['url_method_name'] = p_path.url_method_name + except Exception as err: + self.logger.error(err) + except osv.validation.exceptions.OpenAPIValidationError as val_err: + self.logger.error(f"Validation failed for {spec_file}") + self.logger.error(val_err) + except Exception as gen_err: + self.logger.error(f"Failed to parse spec {spec_file}") + self.logger.error(gen_err) + else: + self.logger.success(f"Successfully parsed spec file {spec_file}") + + + @property + def parsed_dict(self): + "Parsed dict for Jinja2 template from OpenAPI spec" + return self._fdict diff --git a/api_auto_generator/templates/endpoint_template.jinja2 b/api_auto_generator/templates/endpoint_template.jinja2 new file mode 100644 index 00000000..ac795e88 --- /dev/null +++ b/api_auto_generator/templates/endpoint_template.jinja2 @@ -0,0 +1,77 @@ +{#- This template is used to generate Endpoints file for the API Test Automation Framework -#} +""" +This Endpoint file is generated using the api_auto_generator/endpoint_module_generator.py module +""" +from .base_api import BaseAPI + + +class {{class_name}}(BaseAPI): + + + def {{class_content['url_method_name']}}(self, suffix=''): + "Append endpoint to base URI" + return self.base_url + suffix + +{% for function in class_content['instance_methods'] -%} {#- No need to enclose paths in {{}} in for step -#} +{%- for function_name, function_value in function.items() %} + def {{function_name}}({{function_value['instance_method_param_string']}}): + """ + Run {{function_value['http_method']}} request against {{function_value['endpoint']}} + :parameters: + {%- if function_value['params']['query_params'] %} + :params: dict + {%- for query_param in function_value['params']['query_params'] %} + :{{query_param[0]}}: {{query_param[1]}} + {%- endfor %} + {%- endif %} + {%- if function_value['params']['path_params'] %} + {%- for path_param in function_value['params']['path_params'] %} + :{{path_param[0]}}: {{path_param[1]}} + {%- endfor %} + {%- endif %} + {%- if function_value['params']['json_params'] %} + :json: dict + {%- if function_value['params']['content_schema_type'] == 'array' %} + :list: + {%- endif %} + {%- for json_param in function_value['params']['json_params'] %} + {%- if json_param is mapping %} + {%- for json_key, json_value in json_param.items() %} + :{{json_key}}: dict + {%- for nested_json_value in json_value %} + :{{nested_json_value[0]}}: {{nested_json_value[1]}} + {%- endfor %} + {%- endfor %} + {%- else %} + :{{json_param[0]}}: {{json_param[1]}} + {%- endif %} + {%- endfor %} + {%- endif %} + {%- if function_value['params']['data_params'] %} + :data: dict + {%- if function_value['params']['content_schema_type'] == 'array' %} + :list: + {%- endif %} + {%- for data_param in function_value['params']['data_params'] %} + {%- if data_param is mapping %} + {%- for data_key, data_value in data_param.items() %} + :{{data_key}}: dict + {%- for nested_data_value in data_value %} + :{{nested_data_value[0]}}: {{nested_data_value[1]}} + {%- endfor %} + {%- endfor %} + {%- else %} + :{{data_param[0]}}: {{data_param[1]}} + {%- endif %} + {%- endfor %} + {%- endif %} + """ + url = self.{{class_content['url_method_name']}}(f"{{function_value['endpoint']}}") + json_response = self.make_request(method='{{function_value["http_method"]}}', url=url{{function_value['base_api_param_string']}}) + return { + 'url' : url, + 'response' : json_response['json_response'] + } +{% endfor %} +{% endfor %} + \ No newline at end of file diff --git a/endpoints/base_api.py b/endpoints/base_api.py index a4724ab6..b8c7ca23 100644 --- a/endpoints/base_api.py +++ b/endpoints/base_api.py @@ -11,6 +11,33 @@ class BaseAPI: session_object = requests.Session() base_url = None + def make_request(self, + method, + url, + headers=None, + auth=None, + params=None, + data=None, + json=None): + "Generic method to make HTTP request" + headers = headers if headers else {} + try: + response = self.session_object.request(method=method, + url=url, + headers=headers, + auth=auth, + params=params, + data=data, + json=json) + response.raise_for_status() + except HTTPError as http_err: + print(f"{method} request failed: {http_err}") + except ConnectionError: + print(f"\033[1;31mFailed to connect to {url}. Check if the API server is up.\033[1;m") + except RequestException as err: + print(f"\033[1;31mAn error occurred: {err}\033[1;m") + return response + def get(self, url, headers=None): "Get request" headers = headers if headers else {} diff --git a/endpoints/cars_api_endpoints.py b/endpoints/cars_api_endpoints.py index e6858457..9edfefa9 100644 --- a/endpoints/cars_api_endpoints.py +++ b/endpoints/cars_api_endpoints.py @@ -14,7 +14,7 @@ def add_car(self,data,headers): "Adds a new car" try: url = self.cars_url('/add') - json_response = self.post(url,json=data,headers=headers) + json_response = self.make_request(method="post",url=url,json=data,headers=headers) except Exception as err: # pylint: disable=broad-exception-caught print(f"Python says: {err}") json_response = None @@ -27,7 +27,7 @@ def get_cars(self,headers): "gets list of cars" try: url = self.cars_url() - json_response = self.get(url,headers=headers) + json_response = self.make_request(method="get",url=url,headers=headers) except Exception as err: # pylint: disable=broad-exception-caught print(f"Python says: {err}") json_response = None diff --git a/requirements.txt b/requirements.txt index 95276124..ff050836 100644 --- a/requirements.txt +++ b/requirements.txt @@ -20,6 +20,8 @@ axe_selenium_python==2.1.6 pytest-snapshot==0.9.0 beautifulsoup4>=4.12.3 openai==1.12.0 +openapi3-parser==1.1.17 +jinja2==3.1.3 pytesseract==0.3.10 pytest-asyncio==0.23.7 prettytable==3.10.2