From 1a586482f02350fa6ee03ea515efe20a6d2b49bf Mon Sep 17 00:00:00 2001 From: debugtalk Date: Fri, 1 Apr 2022 22:47:47 +0800 Subject: [PATCH] refactor: make step extensible to support implementing new protocols and test types for python version --- docs/CHANGELOG.md | 2 +- hrp/step_request.go | 12 +- hrp/step_testcase.go | 4 +- httprunner/__init__.py | 6 +- httprunner/config.py | 60 +++++ httprunner/models.py | 24 +- httprunner/parser.py | 26 +- httprunner/response.py | 24 +- httprunner/runner.py | 459 ++++++++--------------------------- httprunner/step.py | 40 ++++ httprunner/step_request.py | 462 ++++++++++++++++++++++++++++++++++++ httprunner/step_testcase.py | 110 +++++++++ httprunner/testcase.py | 415 -------------------------------- 13 files changed, 842 insertions(+), 802 deletions(-) create mode 100644 httprunner/config.py create mode 100644 httprunner/step.py create mode 100644 httprunner/step_request.py create mode 100644 httprunner/step_testcase.py delete mode 100644 httprunner/testcase.py diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 3f986f07..d938d777 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -3,6 +3,7 @@ ## v4.0.0-alpha - refactor: merge [hrp] into httprunner v4, which will include golang and python dual engine +- refactor: redesign `IStep` to make step extensible to support implementing new protocols and test types **go version** @@ -11,7 +12,6 @@ - change: integrate [sentry sdk][sentry sdk] for panic reporting and analysis - change: lock funplugin version when creating scaffold project - fix: call referenced api/testcase with relative path -- refactor: redesign `IStep` to make step extensible to support implementing new protocols and test types **python version** diff --git a/hrp/step_request.go b/hrp/step_request.go index 6639740d..78b205c4 100644 --- a/hrp/step_request.go +++ b/hrp/step_request.go @@ -257,6 +257,12 @@ func runStepRequest(r *SessionRunner, step *TStep) (stepResult *StepResult, err } }() + // override step variables + stepVariables, err := r.MergeStepVariables(step.Variables) + if err != nil { + return + } + sessionData := newSessionData() parser := r.GetParser() config := r.GetConfig() @@ -264,12 +270,6 @@ func runStepRequest(r *SessionRunner, step *TStep) (stepResult *StepResult, err rb := newRequestBuilder(parser, config, step.Request) rb.req.Method = string(step.Request.Method) - // override step variables - stepVariables, err := r.MergeStepVariables(step.Variables) - if err != nil { - return - } - err = rb.prepareUrlParams(stepVariables) if err != nil { return diff --git a/hrp/step_testcase.go b/hrp/step_testcase.go index 9d96256f..ea412058 100644 --- a/hrp/step_testcase.go +++ b/hrp/step_testcase.go @@ -77,9 +77,9 @@ func (s *StepTestCaseWithOptionalArgs) Run(r *SessionRunner) (*StepResult, error return stepResult, err } summary := sessionRunner.GetSummary() - stepResult.Data = summary + stepResult.Data = summary.Records // export testcase export variables - stepResult.ExportVars = sessionRunner.summary.InOut.ExportVars + stepResult.ExportVars = summary.InOut.ExportVars stepResult.Success = true // update extracted variables diff --git a/httprunner/__init__.py b/httprunner/__init__.py index 6f10a6f5..5c6dcfe5 100644 --- a/httprunner/__init__.py +++ b/httprunner/__init__.py @@ -1,10 +1,12 @@ __version__ = "4.0.0-alpha" __description__ = "One-stop solution for HTTP(S) testing." -# import firstly for monkey patch if needed +from httprunner.config import Config from httprunner.parser import parse_parameters as Parameters from httprunner.runner import HttpRunner -from httprunner.testcase import Config, Step, RunRequest, RunTestCase +from httprunner.step import Step +from httprunner.step_request import RunRequest +from httprunner.step_testcase import RunTestCase __all__ = [ "__version__", diff --git a/httprunner/config.py b/httprunner/config.py new file mode 100644 index 00000000..2a449ddb --- /dev/null +++ b/httprunner/config.py @@ -0,0 +1,60 @@ +import inspect +from typing import Text + +from httprunner.models import TConfig + + +class Config(object): + def __init__(self, name: Text): + self.__name = name + self.__variables = {} + self.__base_url = "" + self.__verify = False + self.__export = [] + self.__weight = 1 + + caller_frame = inspect.stack()[1] + self.__path = caller_frame.filename + + @property + def name(self) -> Text: + return self.__name + + @property + def path(self) -> Text: + return self.__path + + @property + def weight(self) -> int: + return self.__weight + + def variables(self, **variables) -> "Config": + self.__variables.update(variables) + return self + + def base_url(self, base_url: Text) -> "Config": + self.__base_url = base_url + return self + + def verify(self, verify: bool) -> "Config": + self.__verify = verify + return self + + def export(self, *export_var_name: Text) -> "Config": + self.__export.extend(export_var_name) + return self + + def locust_weight(self, weight: int) -> "Config": + self.__weight = weight + return self + + def struct(self) -> TConfig: + return TConfig( + name=self.__name, + base_url=self.__base_url, + verify=self.__verify, + variables=self.__variables, + export=list(set(self.__export)), + path=self.__path, + weight=self.__weight, + ) diff --git a/httprunner/models.py b/httprunner/models.py index e149e407..95d2fb30 100644 --- a/httprunner/models.py +++ b/httprunner/models.py @@ -156,15 +156,35 @@ class SessionData(BaseModel): class StepData(BaseModel): """teststep data, each step maybe corresponding to one request or one testcase""" + name: Text = "" # teststep name + step_type: Text = "" # teststep type, request or testcase success: bool = False - name: Text = "" # teststep name data: Union[SessionData, List['StepData']] = None + elapsed: float = 0.0 # teststep elapsed time + content_size: float = 0 # response content size export_vars: VariablesMapping = {} + attachment: Text = "" # teststep attachment + - StepData.update_forward_refs() +class IStep(object): + + def name(self) -> str: + raise NotImplementedError + + def type(self) -> str: + raise NotImplementedError + + def struct(self) -> TStep: + raise NotImplementedError + + def run(self, runner) -> StepData: + # runner: HttpRunner + raise NotImplementedError + + class TestCaseSummary(BaseModel): name: Text success: bool diff --git a/httprunner/parser.py b/httprunner/parser.py index 457d8806..9f32d721 100644 --- a/httprunner/parser.py +++ b/httprunner/parser.py @@ -1,14 +1,14 @@ import ast import builtins -import re import os -from typing import Any, Set, Text, Callable, List, Dict, Union +import re +from typing import Any, Callable, Dict, List, Set, Text from loguru import logger from sentry_sdk import capture_exception -from httprunner import loader, utils, exceptions -from httprunner.models import VariablesMapping, FunctionsMapping +from httprunner import exceptions, loader, utils +from httprunner.models import FunctionsMapping, VariablesMapping absolute_http_url_regexp = re.compile(r"^https?://", re.I) @@ -572,3 +572,21 @@ def parse_parameters(parameters: Dict,) -> List[Dict]: parsed_parameters_list.append(parameter_content_list) return utils.gen_cartesian_product(*parsed_parameters_list) + + +class Parser(object): + + def __init__(self, functions_mapping: FunctionsMapping = None) -> None: + self.functions_mapping = functions_mapping + + def parse_string(self, raw_string: Text, variables_mapping: VariablesMapping) -> Any: + return parse_string(raw_string, variables_mapping, self.functions_mapping) + + def parse_variables(self, variables_mapping: VariablesMapping) -> VariablesMapping: + return parse_variables_mapping(variables_mapping, self.functions_mapping) + + def parse_data(self, raw_data: Any, variables_mapping: VariablesMapping = None) -> Any: + return parse_data(raw_data, variables_mapping, self.functions_mapping) + + def get_mapping_function(self, func_name: Text) -> Callable: + return get_mapping_function(func_name, self.functions_mapping) diff --git a/httprunner/response.py b/httprunner/response.py index 7fd43bc5..fb711dc9 100644 --- a/httprunner/response.py +++ b/httprunner/response.py @@ -7,8 +7,8 @@ from loguru import logger from httprunner import exceptions from httprunner.exceptions import ValidationFailure, ParamsError -from httprunner.models import VariablesMapping, Validators, FunctionsMapping -from httprunner.parser import parse_data, parse_string_value, get_mapping_function +from httprunner.models import VariablesMapping, Validators +from httprunner.parser import parse_string_value, Parser def get_uniform_comparator(comparator: Text): @@ -115,7 +115,7 @@ def uniform_validator(validator): class ResponseObject(object): - def __init__(self, resp_obj: requests.Response): + def __init__(self, resp_obj: requests.Response, parser: Parser): """ initialize with a requests.Response object Args: @@ -123,6 +123,7 @@ class ResponseObject(object): """ self.resp_obj = resp_obj + self.parser = parser self.validation_results: Dict = {} def __getattr__(self, key): @@ -170,7 +171,6 @@ class ResponseObject(object): def extract(self, extractors: Dict[Text, Text], variables_mapping: VariablesMapping = None, - functions_mapping: FunctionsMapping = None, ) -> Dict[Text, Any]: if not extractors: return {} @@ -179,8 +179,8 @@ class ResponseObject(object): for key, field in extractors.items(): if '$' in field: # field contains variable or function - field = parse_data( - field, variables_mapping, functions_mapping + field = self.parser.parse_data( + field, variables_mapping ) field_value = self._search_jmespath(field) extract_mapping[key] = field_value @@ -192,11 +192,9 @@ class ResponseObject(object): self, validators: Validators, variables_mapping: VariablesMapping = None, - functions_mapping: FunctionsMapping = None, ): variables_mapping = variables_mapping or {} - functions_mapping = functions_mapping or {} self.validation_results = {} if not validators: @@ -216,8 +214,8 @@ class ResponseObject(object): check_item = u_validator["check"] if "$" in check_item: # check_item is variable or function - check_item = parse_data( - check_item, variables_mapping, functions_mapping + check_item = self.parser.parse_data( + check_item, variables_mapping ) check_item = parse_string_value(check_item) @@ -229,17 +227,17 @@ class ResponseObject(object): # comparator assert_method = u_validator["assert"] - assert_func = get_mapping_function(assert_method, functions_mapping) + assert_func = self.parser.get_mapping_function(assert_method) # expect item expect_item = u_validator["expect"] # parse expected value with config/teststep/extracted variables - expect_value = parse_data(expect_item, variables_mapping, functions_mapping) + expect_value = self.parser.parse_data(expect_item, variables_mapping) # message message = u_validator["message"] # parse message with config/teststep/extracted variables - message = parse_data(message, variables_mapping, functions_mapping) + message = self.parser.parse_data(message, variables_mapping) validate_msg = f"assert {check_item} {assert_method} {expect_value}({type(expect_value).__name__})" diff --git a/httprunner/runner.py b/httprunner/runner.py index 3c5600ad..af9b6ee7 100644 --- a/httprunner/runner.py +++ b/httprunner/runner.py @@ -2,7 +2,7 @@ import os import time import uuid from datetime import datetime -from typing import List, Dict, Text +from typing import Dict, List, Text try: import allure @@ -13,41 +13,29 @@ except ModuleNotFoundError: from loguru import logger -from httprunner import utils, exceptions from httprunner.client import HttpSession -from httprunner.exceptions import ValidationFailure, ParamsError -from httprunner.ext.uploader import prepare_upload_step -from httprunner.loader import load_project_meta, load_testcase_file -from httprunner.parser import build_url, parse_data, parse_variables_mapping -from httprunner.response import ResponseObject -from httprunner.testcase import Config, Step +from httprunner.config import Config +from httprunner.exceptions import ParamsError +from httprunner.loader import load_project_meta +from httprunner.models import (ProjectMeta, StepData, TConfig, TestCaseInOut, + TestCaseSummary, TestCaseTime, VariablesMapping) +from httprunner.parser import Parser from httprunner.utils import merge_variables -from httprunner.models import ( - TConfig, - TStep, - VariablesMapping, - StepData, - TestCaseSummary, - TestCaseTime, - TestCaseInOut, - ProjectMeta, - TestCase, - Hooks, -) class HttpRunner(object): config: Config - teststeps: List[Step] + teststeps: List[object] # list of Step + + parser: Parser = None + session: HttpSession = None + case_id: Text = "" + root_dir: Text = "" - success: bool = False # indicate testcase execution result __config: TConfig - __teststeps: List[TStep] __project_meta: ProjectMeta = None - __case_id: Text = "" __export: List[Text] = [] __step_datas: List[StepData] = [] - __session: HttpSession = None __session_variables: VariablesMapping = {} # time __start_at: float = 0 @@ -55,29 +43,34 @@ class HttpRunner(object): # log __log_path: Text = "" - def __init_tests__(self): - self.__config = self.config.perform() - self.__teststeps = [] - for step in self.teststeps: - self.__teststeps.append(step.perform()) + def __init(self): + self.__config = self.config.struct() + self.__session_variables = {} + self.__start_at = 0 + self.__duration = 0 - @property - def raw_testcase(self) -> TestCase: - if not hasattr(self, "__config"): - self.__init_tests__() + self.__project_meta = self.__project_meta or load_project_meta( + self.__config.path + ) + self.case_id = self.case_id or str(uuid.uuid4()) + self.root_dir = self.root_dir or self.__project_meta.RootDir + self.__log_path = os.path.join( + self.root_dir, "logs", f"{self.case_id}.run.log" + ) - return TestCase(config=self.__config, teststeps=self.__teststeps) - - def with_project_meta(self, project_meta: ProjectMeta) -> "HttpRunner": - self.__project_meta = project_meta - return self + self.__step_datas.clear() + self.session = self.session or HttpSession() + self.parser = self.parser or Parser(self.__project_meta.functions) def with_session(self, session: HttpSession) -> "HttpRunner": - self.__session = session + self.session = session return self + def get_config(self) -> TConfig: + return self.__config + def with_case_id(self, case_id: Text) -> "HttpRunner": - self.__case_id = case_id + self.case_id = case_id return self def with_variables(self, variables: VariablesMapping) -> "HttpRunner": @@ -88,302 +81,24 @@ class HttpRunner(object): self.__export = export return self - def __call_hooks(self, hooks: Hooks, step_variables: VariablesMapping, hook_msg: Text): - """ call hook actions. - - Args: - hooks (list): each hook in hooks list maybe in two format. - - format1 (str): only call hook functions. - ${func()} - format2 (dict): assignment, the value returned by hook function will be assigned to variable. - {"var": "${func()}"} - - step_variables: current step variables to call hook, include two special variables - - request: parsed request dict - response: ResponseObject for current response - - hook_msg: setup/teardown request/testcase - - """ - logger.info(f"call hook actions: {hook_msg}") - - if not isinstance(hooks, List): - logger.error(f"Invalid hooks format: {hooks}") - return - - for hook in hooks: - if isinstance(hook, Text): - # format 1: ["${func()}"] - logger.debug(f"call hook function: {hook}") - parse_data(hook, step_variables, self.__project_meta.functions) - elif isinstance(hook, Dict) and len(hook) == 1: - # format 2: {"var": "${func()}"} - var_name, hook_content = list(hook.items())[0] - hook_content_eval = parse_data( - hook_content, step_variables, self.__project_meta.functions - ) - logger.debug( - f"call hook function: {hook_content}, got value: {hook_content_eval}" - ) - logger.debug(f"assign variable: {var_name} = {hook_content_eval}") - step_variables[var_name] = hook_content_eval - else: - logger.error(f"Invalid hook format: {hook}") - - def __run_step_request(self, step: TStep) -> StepData: - """run teststep: request""" - step_data = StepData(name=step.name) - - # parse - functions = self.__project_meta.functions - prepare_upload_step(step, functions) - request_dict = step.request.dict() - request_dict.pop("upload", None) - parsed_request_dict = parse_data( - request_dict, step.variables, functions - ) - parsed_request_dict["headers"].setdefault( - "HRUN-Request-ID", - f"HRUN-{self.__case_id}-{str(int(time.time() * 1000))[-6:]}", - ) - step.variables["request"] = parsed_request_dict - - # setup hooks - if step.setup_hooks: - self.__call_hooks(step.setup_hooks, step.variables, "setup request") - - # prepare arguments - method = parsed_request_dict.pop("method") - url_path = parsed_request_dict.pop("url") - url = build_url(self.__config.base_url, url_path) - parsed_request_dict["verify"] = self.__config.verify - parsed_request_dict["json"] = parsed_request_dict.pop("req_json", {}) - - # request - resp = self.__session.request(method, url, **parsed_request_dict) - resp_obj = ResponseObject(resp) - step.variables["response"] = resp_obj - - # teardown hooks - if step.teardown_hooks: - self.__call_hooks(step.teardown_hooks, step.variables, "teardown request") - - def log_req_resp_details(): - err_msg = "\n{} DETAILED REQUEST & RESPONSE {}\n".format("*" * 32, "*" * 32) - - # log request - err_msg += "====== request details ======\n" - err_msg += f"url: {url}\n" - err_msg += f"method: {method}\n" - headers = parsed_request_dict.pop("headers", {}) - err_msg += f"headers: {headers}\n" - for k, v in parsed_request_dict.items(): - v = utils.omit_long_data(v) - err_msg += f"{k}: {repr(v)}\n" - - err_msg += "\n" - - # log response - err_msg += "====== response details ======\n" - err_msg += f"status_code: {resp.status_code}\n" - err_msg += f"headers: {resp.headers}\n" - err_msg += f"body: {repr(resp.text)}\n" - logger.error(err_msg) - - # extract - extractors = step.extract - extract_mapping = resp_obj.extract(extractors, step.variables, functions) - step_data.export_vars = extract_mapping - - variables_mapping = step.variables - variables_mapping.update(extract_mapping) - - # validate - validators = step.validators - session_success = False - try: - resp_obj.validate( - validators, variables_mapping, functions - ) - session_success = True - except ValidationFailure: - session_success = False - log_req_resp_details() - # log testcase duration before raise ValidationFailure - self.__duration = time.time() - self.__start_at - raise - finally: - self.success = session_success - step_data.success = session_success - - if hasattr(self.__session, "data"): - # httprunner.client.HttpSession, not locust.clients.HttpSession - # save request & response meta data - self.__session.data.success = session_success - self.__session.data.validators = resp_obj.validation_results - - # save step data - step_data.data = self.__session.data - - return step_data - - def __run_step_testcase(self, step: TStep) -> StepData: - """run teststep: referenced testcase""" - step_data = StepData(name=step.name) - step_variables = step.variables - step_export = step.export - - # setup hooks - if step.setup_hooks: - self.__call_hooks(step.setup_hooks, step_variables, "setup testcase") - - if hasattr(step.testcase, "config") and hasattr(step.testcase, "teststeps"): - testcase_cls = step.testcase - case_result = ( - testcase_cls() - .with_session(self.__session) - .with_case_id(self.__case_id) - .with_variables(step_variables) - .with_export(step_export) - .run() - ) - - elif isinstance(step.testcase, Text): - if os.path.isabs(step.testcase): - ref_testcase_path = step.testcase - else: - ref_testcase_path = os.path.join( - self.__project_meta.RootDir, step.testcase - ) - - case_result = ( - HttpRunner() - .with_session(self.__session) - .with_case_id(self.__case_id) - .with_variables(step_variables) - .with_export(step_export) - .run_path(ref_testcase_path) - ) - - else: - raise exceptions.ParamsError( - f"Invalid teststep referenced testcase: {step.dict()}" - ) - - # teardown hooks - if step.teardown_hooks: - self.__call_hooks(step.teardown_hooks, step.variables, "teardown testcase") - - step_data.data = case_result.get_step_datas() # list of step data - step_data.export_vars = case_result.get_export_variables() - step_data.success = case_result.success - self.success = case_result.success - - if step_data.export_vars: - logger.info(f"export variables: {step_data.export_vars}") - - return step_data - - def __run_step(self, step: TStep) -> Dict: - """run teststep, teststep maybe a request or referenced testcase""" - logger.info(f"run step begin: {step.name} >>>>>>") - - if step.request: - step_data = self.__run_step_request(step) - elif step.testcase: - step_data = self.__run_step_testcase(step) - else: - raise ParamsError( - f"teststep is neither a request nor a referenced testcase: {step.dict()}" - ) - - self.__step_datas.append(step_data) - logger.info(f"run step end: {step.name} <<<<<<\n") - return step_data.export_vars - - def __parse_config(self, config: TConfig): - config.variables.update(self.__session_variables) - config.variables = parse_variables_mapping( - config.variables, self.__project_meta.functions - ) - config.name = parse_data( - config.name, config.variables, self.__project_meta.functions - ) - config.base_url = parse_data( - config.base_url, config.variables, self.__project_meta.functions + def __parse_config(self, param: Dict = None) -> None: + # parse config variables + self.__config.variables.update(self.__session_variables) + if param: + self.__config.variables.update(param) + self.__config.variables = self.parser.parse_variables( + self.__config.variables ) - def run_testcase(self, testcase: TestCase) -> "HttpRunner": - """run specified testcase - - Examples: - >>> testcase_obj = TestCase(config=TConfig(...), teststeps=[TStep(...)]) - >>> HttpRunner().with_project_meta(project_meta).run_testcase(testcase_obj) - - """ - self.__config = testcase.config - self.__teststeps = testcase.teststeps - - # prepare - self.__project_meta = self.__project_meta or load_project_meta( - self.__config.path + # parse config name + self.__config.name = self.parser.parse_data( + self.__config.name, self.__config.variables ) - self.__parse_config(self.__config) - self.__start_at = time.time() - self.__step_datas: List[StepData] = [] - self.__session = self.__session or HttpSession() - # save extracted variables of teststeps - extracted_variables: VariablesMapping = {} - # run teststeps - for step in self.__teststeps: - # override variables - # step variables > extracted variables from previous steps - step.variables = merge_variables(step.variables, extracted_variables) - # step variables > testcase config variables - step.variables = merge_variables(step.variables, self.__config.variables) - - # parse variables - step.variables = parse_variables_mapping( - step.variables, self.__project_meta.functions - ) - - # run step - if USE_ALLURE: - with allure.step(f"step: {step.name}"): - extract_mapping = self.__run_step(step) - else: - extract_mapping = self.__run_step(step) - - # save extracted variables to session variables - extracted_variables.update(extract_mapping) - - self.__session_variables.update(extracted_variables) - self.__duration = time.time() - self.__start_at - return self - - def run_path(self, path: Text) -> "HttpRunner": - if not os.path.isfile(path): - raise exceptions.ParamsError(f"Invalid testcase path: {path}") - - testcase_obj = load_testcase_file(path) - return self.run_testcase(testcase_obj) - - def run(self) -> "HttpRunner": - """ run current testcase - - Examples: - >>> TestCaseRequestWithFunctions().run() - - """ - self.__init_tests__() - testcase_obj = TestCase(config=self.__config, teststeps=self.__teststeps) - return self.run_testcase(testcase_obj) - - def get_step_datas(self) -> List[StepData]: - return self.__step_datas + # parse config base url + self.__config.base_url = self.parser.parse_data( + self.__config.base_url, self.__config.variables + ) def get_export_variables(self) -> Dict: # override testcase export vars with step export @@ -403,10 +118,17 @@ class HttpRunner(object): """get testcase result summary""" start_at_timestamp = self.__start_at start_at_iso_format = datetime.utcfromtimestamp(start_at_timestamp).isoformat() + + summary_success = True + for step_data in self.__step_datas: + if not step_data.success: + summary_success = False + break + return TestCaseSummary( name=self.__config.name, - success=self.success, - case_id=self.__case_id, + success=summary_success, + case_id=self.case_id, time=TestCaseTime( start_at=self.__start_at, start_at_iso_format=start_at_iso_format, @@ -420,40 +142,63 @@ class HttpRunner(object): step_datas=self.__step_datas, ) - def test_start(self, param: Dict = None) -> "HttpRunner": + def merge_step_variables(self, variables: VariablesMapping) -> VariablesMapping: + # override variables + # step variables > extracted variables from previous steps + variables = merge_variables(variables, self.__session_variables) + # step variables > testcase config variables + variables = merge_variables(variables, self.__config.variables) + + # parse variables + return self.parser.parse_variables(variables) + + def __run_step(self, step) -> Dict: + """run teststep, step maybe any kind that implements IStep interface + + Args: + step (Step): teststep + + """ + logger.info(f"run step begin: {step.name()} >>>>>>") + + # run step + if USE_ALLURE: + with allure.step(f"step: {step.name()}"): + step_result = step.run(self) + else: + step_result = step.run(self) + + # save extracted variables to session variables + self.__session_variables.update(step_result.export_vars) + # update testcase summary + self.__step_datas.append(step_result) + + logger.info(f"run step end: {step.name()} <<<<<<\n") + + def test_start(self, param: Dict = None): """main entrance, discovered by pytest""" - self.__init_tests__() - self.__project_meta = self.__project_meta or load_project_meta( - self.__config.path - ) - self.__case_id = self.__case_id or str(uuid.uuid4()) - self.__log_path = self.__log_path or os.path.join( - self.__project_meta.RootDir, "logs", f"{self.__case_id}.run.log" - ) + self.__init() log_handler = logger.add(self.__log_path, level="DEBUG") - # parse config name - config_variables = self.__config.variables - if param: - config_variables.update(param) - config_variables.update(self.__session_variables) - self.__config.name = parse_data( - self.__config.name, config_variables, self.__project_meta.functions - ) + self.__parse_config(param) if USE_ALLURE: # update allure report meta allure.dynamic.title(self.__config.name) - allure.dynamic.description(f"TestCase ID: {self.__case_id}") + allure.dynamic.description(f"TestCase ID: {self.case_id}") logger.info( - f"Start to run testcase: {self.__config.name}, TestCase ID: {self.__case_id}" + f"Start to run testcase: {self.__config.name}, TestCase ID: {self.case_id}" ) + self.__start_at = time.time() try: - return self.run_testcase( - TestCase(config=self.__config, teststeps=self.__teststeps) - ) + # run step in sequential order + for step in self.teststeps: + self.__run_step(step) finally: logger.remove(log_handler) logger.info(f"generate testcase log: {self.__log_path}") + + self.__duration = time.time() - self.__start_at + return self diff --git a/httprunner/step.py b/httprunner/step.py new file mode 100644 index 00000000..f6bc7c90 --- /dev/null +++ b/httprunner/step.py @@ -0,0 +1,40 @@ +from typing import Union + +from httprunner.models import StepData, TRequest, TStep, TestCase +from httprunner.runner import HttpRunner +from httprunner.step_request import RequestWithOptionalArgs, StepRequestExtraction, StepRequestValidation +from httprunner.step_testcase import StepRefCase + + +class Step(object): + + def __init__( + self, + step: Union[ + StepRequestValidation, + StepRequestExtraction, + RequestWithOptionalArgs, + StepRefCase, + ], + ): + self.__step = step + + @property + def request(self) -> TRequest: + return self.__step.struct().request + + @property + def testcase(self) -> TestCase: + return self.__step.struct().testcase + + def struct(self) -> TStep: + return self.__step.struct() + + def name(self) -> str: + return self.__step.name() + + def type(self) -> str: + return self.__step.type() + + def run(self, runner: HttpRunner) -> StepData: + return self.__step.run(runner) diff --git a/httprunner/step_request.py b/httprunner/step_request.py new file mode 100644 index 00000000..2633bcbd --- /dev/null +++ b/httprunner/step_request.py @@ -0,0 +1,462 @@ +import time +from typing import Any, Dict, List, Text, Union + +from loguru import logger + +from httprunner import utils +from httprunner.exceptions import ValidationFailure +from httprunner.ext.uploader import prepare_upload_step +from httprunner.models import (Hooks, IStep, MethodEnum, StepData, TRequest, + TStep, VariablesMapping) +from httprunner.parser import build_url +from httprunner.response import ResponseObject +from httprunner.runner import HttpRunner + + +def call_hooks(runner: HttpRunner, hooks: Hooks, step_variables: VariablesMapping, hook_msg: Text): + """ call hook actions. + + Args: + hooks (list): each hook in hooks list maybe in two format. + + format1 (str): only call hook functions. + ${func()} + format2 (dict): assignment, the value returned by hook function will be assigned to variable. + {"var": "${func()}"} + + step_variables: current step variables to call hook, include two special variables + + request: parsed request dict + response: ResponseObject for current response + + hook_msg: setup/teardown request/testcase + + """ + logger.info(f"call hook actions: {hook_msg}") + + if not isinstance(hooks, List): + logger.error(f"Invalid hooks format: {hooks}") + return + + for hook in hooks: + if isinstance(hook, Text): + # format 1: ["${func()}"] + logger.debug(f"call hook function: {hook}") + runner.parser.parse_data(hook, step_variables) + elif isinstance(hook, Dict) and len(hook) == 1: + # format 2: {"var": "${func()}"} + var_name, hook_content = list(hook.items())[0] + hook_content_eval = runner.parser.parse_data( + hook_content, step_variables + ) + logger.debug( + f"call hook function: {hook_content}, got value: {hook_content_eval}" + ) + logger.debug(f"assign variable: {var_name} = {hook_content_eval}") + step_variables[var_name] = hook_content_eval + else: + logger.error(f"Invalid hook format: {hook}") + + +def run_step_request(runner: HttpRunner, step: TStep) -> StepData: + """run teststep: request""" + step_data = StepData( + name=step.name, + success=False, + ) + + step.variables = runner.merge_step_variables(step.variables) + + # parse + functions = runner.parser.functions_mapping + prepare_upload_step(step, functions) + request_dict = step.request.dict() + request_dict.pop("upload", None) + parsed_request_dict = runner.parser.parse_data( + request_dict, step.variables + ) + parsed_request_dict["headers"].setdefault( + "HRUN-Request-ID", + f"HRUN-{runner.case_id}-{str(int(time.time() * 1000))[-6:]}", + ) + step.variables["request"] = parsed_request_dict + + # setup hooks + if step.setup_hooks: + call_hooks(runner, step.setup_hooks, step.variables, "setup request") + + # prepare arguments + config = runner.get_config() + method = parsed_request_dict.pop("method") + url_path = parsed_request_dict.pop("url") + url = build_url(config.base_url, url_path) + parsed_request_dict["verify"] = config.verify + parsed_request_dict["json"] = parsed_request_dict.pop("req_json", {}) + + # request + resp = runner.session.request(method, url, **parsed_request_dict) + resp_obj = ResponseObject(resp, runner.parser) + step.variables["response"] = resp_obj + + # teardown hooks + if step.teardown_hooks: + call_hooks(runner, step.teardown_hooks, step.variables, "teardown request") + + def log_req_resp_details(): + err_msg = "\n{} DETAILED REQUEST & RESPONSE {}\n".format("*" * 32, "*" * 32) + + # log request + err_msg += "====== request details ======\n" + err_msg += f"url: {url}\n" + err_msg += f"method: {method}\n" + headers = parsed_request_dict.pop("headers", {}) + err_msg += f"headers: {headers}\n" + for k, v in parsed_request_dict.items(): + v = utils.omit_long_data(v) + err_msg += f"{k}: {repr(v)}\n" + + err_msg += "\n" + + # log response + err_msg += "====== response details ======\n" + err_msg += f"status_code: {resp.status_code}\n" + err_msg += f"headers: {resp.headers}\n" + err_msg += f"body: {repr(resp.text)}\n" + logger.error(err_msg) + + # extract + extractors = step.extract + extract_mapping = resp_obj.extract(extractors, step.variables) + step_data.export_vars = extract_mapping + + variables_mapping = step.variables + variables_mapping.update(extract_mapping) + + # validate + validators = step.validators + try: + resp_obj.validate( + validators, variables_mapping + ) + step_data.success = True + except ValidationFailure: + log_req_resp_details() + # log testcase duration before raise ValidationFailure + step_data.elapsed = time.time() - runner.__start_at + raise + finally: + session_data = runner.session.data + session_data.success = step_data.success + session_data.validators = resp_obj.validation_results + + # save step data + step_data.data = session_data + + return step_data + + +class StepRequestValidation(IStep): + def __init__(self, step: TStep): + self.__step = step + + def assert_equal( + self, jmes_path: Text, expected_value: Any, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"equal": [jmes_path, expected_value, message]} + ) + return self + + def assert_not_equal( + self, jmes_path: Text, expected_value: Any, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"not_equal": [jmes_path, expected_value, message]} + ) + return self + + def assert_greater_than( + self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"greater_than": [jmes_path, expected_value, message]} + ) + return self + + def assert_less_than( + self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"less_than": [jmes_path, expected_value, message]} + ) + return self + + def assert_greater_or_equals( + self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"greater_or_equals": [jmes_path, expected_value, message]} + ) + return self + + def assert_less_or_equals( + self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"less_or_equals": [jmes_path, expected_value, message]} + ) + return self + + def assert_length_equal( + self, jmes_path: Text, expected_value: int, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"length_equal": [jmes_path, expected_value, message]} + ) + return self + + def assert_length_greater_than( + self, jmes_path: Text, expected_value: int, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"length_greater_than": [jmes_path, expected_value, message]} + ) + return self + + def assert_length_less_than( + self, jmes_path: Text, expected_value: int, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"length_less_than": [jmes_path, expected_value, message]} + ) + return self + + def assert_length_greater_or_equals( + self, jmes_path: Text, expected_value: int, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"length_greater_or_equals": [jmes_path, expected_value, message]} + ) + return self + + def assert_length_less_or_equals( + self, jmes_path: Text, expected_value: int, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"length_less_or_equals": [jmes_path, expected_value, message]} + ) + return self + + def assert_string_equals( + self, jmes_path: Text, expected_value: Any, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"string_equals": [jmes_path, expected_value, message]} + ) + return self + + def assert_startswith( + self, jmes_path: Text, expected_value: Text, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"startswith": [jmes_path, expected_value, message]} + ) + return self + + def assert_endswith( + self, jmes_path: Text, expected_value: Text, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"endswith": [jmes_path, expected_value, message]} + ) + return self + + def assert_regex_match( + self, jmes_path: Text, expected_value: Text, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"regex_match": [jmes_path, expected_value, message]} + ) + return self + + def assert_contains( + self, jmes_path: Text, expected_value: Any, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"contains": [jmes_path, expected_value, message]} + ) + return self + + def assert_contained_by( + self, jmes_path: Text, expected_value: Any, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"contained_by": [jmes_path, expected_value, message]} + ) + return self + + def assert_type_match( + self, jmes_path: Text, expected_value: Any, message: Text = "" + ) -> "StepRequestValidation": + self.__step.validators.append( + {"type_match": [jmes_path, expected_value, message]} + ) + return self + + def struct(self) -> TStep: + return self.__step + + def name(self) -> Text: + return self.__step.name + + def type(self) -> Text: + return f"request-{self.__step.request.method}" + + def run(self, runner: HttpRunner): + return run_step_request(runner, self.__step) + + +class StepRequestExtraction(IStep): + def __init__(self, step: TStep): + self.__step = step + + def with_jmespath(self, jmes_path: Text, var_name: Text) -> "StepRequestExtraction": + self.__step.extract[var_name] = jmes_path + return self + + # def with_regex(self): + # # TODO: extract response html with regex + # pass + # + # def with_jsonpath(self): + # # TODO: extract response json with jsonpath + # pass + + def validate(self) -> StepRequestValidation: + return StepRequestValidation(self.__step) + + def struct(self) -> TStep: + return self.__step + + def name(self) -> Text: + return self.__step.name + + def type(self) -> Text: + return f"request-{self.__step.request.method}" + + def run(self, runner: HttpRunner): + return run_step_request(runner, self.__step) + + +class RequestWithOptionalArgs(IStep): + def __init__(self, step: TStep): + self.__step = step + + def with_params(self, **params) -> "RequestWithOptionalArgs": + self.__step.request.params.update(params) + return self + + def with_headers(self, **headers) -> "RequestWithOptionalArgs": + self.__step.request.headers.update(headers) + return self + + def with_cookies(self, **cookies) -> "RequestWithOptionalArgs": + self.__step.request.cookies.update(cookies) + return self + + def with_data(self, data) -> "RequestWithOptionalArgs": + self.__step.request.data = data + return self + + def with_json(self, req_json) -> "RequestWithOptionalArgs": + self.__step.request.req_json = req_json + return self + + def set_timeout(self, timeout: float) -> "RequestWithOptionalArgs": + self.__step.request.timeout = timeout + return self + + def set_verify(self, verify: bool) -> "RequestWithOptionalArgs": + self.__step.request.verify = verify + return self + + def set_allow_redirects(self, allow_redirects: bool) -> "RequestWithOptionalArgs": + self.__step.request.allow_redirects = allow_redirects + return self + + def upload(self, **file_info) -> "RequestWithOptionalArgs": + self.__step.request.upload.update(file_info) + return self + + def teardown_hook( + self, hook: Text, assign_var_name: Text = None + ) -> "RequestWithOptionalArgs": + if assign_var_name: + self.__step.teardown_hooks.append({assign_var_name: hook}) + else: + self.__step.teardown_hooks.append(hook) + + return self + + def extract(self) -> StepRequestExtraction: + return StepRequestExtraction(self.__step) + + def validate(self) -> StepRequestValidation: + return StepRequestValidation(self.__step) + + def struct(self) -> TStep: + return self.__step + + def name(self) -> Text: + return self.__step.name + + def type(self) -> Text: + return f"request-{self.__step.request.method}" + + def run(self, runner: HttpRunner): + return run_step_request(runner, self.__step) + + +class RunRequest(object): + def __init__(self, name: Text): + self.__step = TStep(name=name) + + def with_variables(self, **variables) -> "RunRequest": + self.__step.variables.update(variables) + return self + + def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunRequest": + if assign_var_name: + self.__step.setup_hooks.append({assign_var_name: hook}) + else: + self.__step.setup_hooks.append(hook) + + return self + + def get(self, url: Text) -> RequestWithOptionalArgs: + self.__step.request = TRequest(method=MethodEnum.GET, url=url) + return RequestWithOptionalArgs(self.__step) + + def post(self, url: Text) -> RequestWithOptionalArgs: + self.__step.request = TRequest(method=MethodEnum.POST, url=url) + return RequestWithOptionalArgs(self.__step) + + def put(self, url: Text) -> RequestWithOptionalArgs: + self.__step.request = TRequest(method=MethodEnum.PUT, url=url) + return RequestWithOptionalArgs(self.__step) + + def head(self, url: Text) -> RequestWithOptionalArgs: + self.__step.request = TRequest(method=MethodEnum.HEAD, url=url) + return RequestWithOptionalArgs(self.__step) + + def delete(self, url: Text) -> RequestWithOptionalArgs: + self.__step.request = TRequest(method=MethodEnum.DELETE, url=url) + return RequestWithOptionalArgs(self.__step) + + def options(self, url: Text) -> RequestWithOptionalArgs: + self.__step.request = TRequest(method=MethodEnum.OPTIONS, url=url) + return RequestWithOptionalArgs(self.__step) + + def patch(self, url: Text) -> RequestWithOptionalArgs: + self.__step.request = TRequest(method=MethodEnum.PATCH, url=url) + return RequestWithOptionalArgs(self.__step) diff --git a/httprunner/step_testcase.py b/httprunner/step_testcase.py new file mode 100644 index 00000000..0882428d --- /dev/null +++ b/httprunner/step_testcase.py @@ -0,0 +1,110 @@ +import os +from typing import Text, Callable + +from loguru import logger +from httprunner import exceptions +from httprunner.loader import load_testcase_file + +from httprunner.step_request import call_hooks +from httprunner.runner import HttpRunner +from httprunner.models import ( + TStep, + StepData +) + + +def run_step_testcase(runner: HttpRunner, step: TStep) -> StepData: + """run teststep: referenced testcase""" + step_data = StepData(name=step.name) + step_variables = step.variables + step_export = step.export + + # setup hooks + if step.setup_hooks: + call_hooks(runner, step.setup_hooks, step_variables, "setup testcase") + + # TODO: override testcase with current step name/variables/export + + ref_case_runner = HttpRunner() + ref_case_runner.config = step.testcase.config + ref_case_runner.teststeps = step.testcase.teststeps + ref_case_runner.with_session(runner.session) \ + .with_case_id(runner.case_id) \ + .with_variables(step_variables) \ + .with_export(step_export) \ + .test_start() + + # teardown hooks + if step.teardown_hooks: + call_hooks(runner, step.teardown_hooks, step.variables, "teardown testcase") + + summary = ref_case_runner.get_summary() + step_data.data = summary.step_datas # list of step data + step_data.export_vars = summary.in_out.export_vars + step_data.success = summary.success + + if step_data.export_vars: + logger.info(f"export variables: {step_data.export_vars}") + + return step_data + + +class StepRefCase(object): + def __init__(self, step: TStep): + self.__step = step + + def teardown_hook(self, hook: Text, assign_var_name: Text = None) -> "StepRefCase": + if assign_var_name: + self.__step.teardown_hooks.append({assign_var_name: hook}) + else: + self.__step.teardown_hooks.append(hook) + + return self + + def export(self, *var_name: Text) -> "StepRefCase": + self.__step.export.extend(var_name) + return self + + def struct(self) -> TStep: + return self.__step + + def name(self) -> Text: + return self.__step.name + + def type(self) -> Text: + return f"request-{self.__step.request.method}" + + def run(self, runner: HttpRunner): + return run_step_testcase(runner, self.__step) + + +class RunTestCase(object): + def __init__(self, name: Text): + self.__step = TStep(name=name) + + def with_variables(self, **variables) -> "RunTestCase": + self.__step.variables.update(variables) + return self + + def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunTestCase": + if assign_var_name: + self.__step.setup_hooks.append({assign_var_name: hook}) + else: + self.__step.setup_hooks.append(hook) + + return self + + def call(self, testcase: Callable) -> StepRefCase: + if hasattr(testcase, "config") and hasattr(testcase, "teststeps"): + self.__step.testcase = testcase + elif isinstance(testcase, Text): + if not os.path.isfile(testcase): + raise exceptions.ParamsError(f"Invalid testcase path: {testcase}") + + self.__step.testcase = load_testcase_file(testcase) + else: + raise exceptions.ParamsError( + f"Invalid teststep referenced testcase: {testcase}" + ) + + return StepRefCase(self.__step) diff --git a/httprunner/testcase.py b/httprunner/testcase.py deleted file mode 100644 index 86d1554f..00000000 --- a/httprunner/testcase.py +++ /dev/null @@ -1,415 +0,0 @@ -import inspect -from typing import Text, Any, Union, Callable - -from httprunner.models import ( - TConfig, - TStep, - TRequest, - MethodEnum, - TestCase, -) - - -class Config(object): - def __init__(self, name: Text): - self.__name = name - self.__variables = {} - self.__base_url = "" - self.__verify = False - self.__export = [] - self.__weight = 1 - - caller_frame = inspect.stack()[1] - self.__path = caller_frame.filename - - @property - def name(self) -> Text: - return self.__name - - @property - def path(self) -> Text: - return self.__path - - @property - def weight(self) -> int: - return self.__weight - - def variables(self, **variables) -> "Config": - self.__variables.update(variables) - return self - - def base_url(self, base_url: Text) -> "Config": - self.__base_url = base_url - return self - - def verify(self, verify: bool) -> "Config": - self.__verify = verify - return self - - def export(self, *export_var_name: Text) -> "Config": - self.__export.extend(export_var_name) - return self - - def locust_weight(self, weight: int) -> "Config": - self.__weight = weight - return self - - def perform(self) -> TConfig: - return TConfig( - name=self.__name, - base_url=self.__base_url, - verify=self.__verify, - variables=self.__variables, - export=list(set(self.__export)), - path=self.__path, - weight=self.__weight, - ) - - -class StepRequestValidation(object): - def __init__(self, step_context: TStep): - self.__step_context = step_context - - def assert_equal( - self, jmes_path: Text, expected_value: Any, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"equal": [jmes_path, expected_value, message]} - ) - return self - - def assert_not_equal( - self, jmes_path: Text, expected_value: Any, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"not_equal": [jmes_path, expected_value, message]} - ) - return self - - def assert_greater_than( - self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"greater_than": [jmes_path, expected_value, message]} - ) - return self - - def assert_less_than( - self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"less_than": [jmes_path, expected_value, message]} - ) - return self - - def assert_greater_or_equals( - self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"greater_or_equals": [jmes_path, expected_value, message]} - ) - return self - - def assert_less_or_equals( - self, jmes_path: Text, expected_value: Union[int, float], message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"less_or_equals": [jmes_path, expected_value, message]} - ) - return self - - def assert_length_equal( - self, jmes_path: Text, expected_value: int, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"length_equal": [jmes_path, expected_value, message]} - ) - return self - - def assert_length_greater_than( - self, jmes_path: Text, expected_value: int, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"length_greater_than": [jmes_path, expected_value, message]} - ) - return self - - def assert_length_less_than( - self, jmes_path: Text, expected_value: int, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"length_less_than": [jmes_path, expected_value, message]} - ) - return self - - def assert_length_greater_or_equals( - self, jmes_path: Text, expected_value: int, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"length_greater_or_equals": [jmes_path, expected_value, message]} - ) - return self - - def assert_length_less_or_equals( - self, jmes_path: Text, expected_value: int, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"length_less_or_equals": [jmes_path, expected_value, message]} - ) - return self - - def assert_string_equals( - self, jmes_path: Text, expected_value: Any, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"string_equals": [jmes_path, expected_value, message]} - ) - return self - - def assert_startswith( - self, jmes_path: Text, expected_value: Text, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"startswith": [jmes_path, expected_value, message]} - ) - return self - - def assert_endswith( - self, jmes_path: Text, expected_value: Text, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"endswith": [jmes_path, expected_value, message]} - ) - return self - - def assert_regex_match( - self, jmes_path: Text, expected_value: Text, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"regex_match": [jmes_path, expected_value, message]} - ) - return self - - def assert_contains( - self, jmes_path: Text, expected_value: Any, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"contains": [jmes_path, expected_value, message]} - ) - return self - - def assert_contained_by( - self, jmes_path: Text, expected_value: Any, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"contained_by": [jmes_path, expected_value, message]} - ) - return self - - def assert_type_match( - self, jmes_path: Text, expected_value: Any, message: Text = "" - ) -> "StepRequestValidation": - self.__step_context.validators.append( - {"type_match": [jmes_path, expected_value, message]} - ) - return self - - def perform(self) -> TStep: - return self.__step_context - - -class StepRequestExtraction(object): - def __init__(self, step_context: TStep): - self.__step_context = step_context - - def with_jmespath(self, jmes_path: Text, var_name: Text) -> "StepRequestExtraction": - self.__step_context.extract[var_name] = jmes_path - return self - - # def with_regex(self): - # # TODO: extract response html with regex - # pass - # - # def with_jsonpath(self): - # # TODO: extract response json with jsonpath - # pass - - def validate(self) -> StepRequestValidation: - return StepRequestValidation(self.__step_context) - - def perform(self) -> TStep: - return self.__step_context - - -class RequestWithOptionalArgs(object): - def __init__(self, step_context: TStep): - self.__step_context = step_context - - def with_params(self, **params) -> "RequestWithOptionalArgs": - self.__step_context.request.params.update(params) - return self - - def with_headers(self, **headers) -> "RequestWithOptionalArgs": - self.__step_context.request.headers.update(headers) - return self - - def with_cookies(self, **cookies) -> "RequestWithOptionalArgs": - self.__step_context.request.cookies.update(cookies) - return self - - def with_data(self, data) -> "RequestWithOptionalArgs": - self.__step_context.request.data = data - return self - - def with_json(self, req_json) -> "RequestWithOptionalArgs": - self.__step_context.request.req_json = req_json - return self - - def set_timeout(self, timeout: float) -> "RequestWithOptionalArgs": - self.__step_context.request.timeout = timeout - return self - - def set_verify(self, verify: bool) -> "RequestWithOptionalArgs": - self.__step_context.request.verify = verify - return self - - def set_allow_redirects(self, allow_redirects: bool) -> "RequestWithOptionalArgs": - self.__step_context.request.allow_redirects = allow_redirects - return self - - def upload(self, **file_info) -> "RequestWithOptionalArgs": - self.__step_context.request.upload.update(file_info) - return self - - def teardown_hook( - self, hook: Text, assign_var_name: Text = None - ) -> "RequestWithOptionalArgs": - if assign_var_name: - self.__step_context.teardown_hooks.append({assign_var_name: hook}) - else: - self.__step_context.teardown_hooks.append(hook) - - return self - - def extract(self) -> StepRequestExtraction: - return StepRequestExtraction(self.__step_context) - - def validate(self) -> StepRequestValidation: - return StepRequestValidation(self.__step_context) - - def perform(self) -> TStep: - return self.__step_context - - -class RunRequest(object): - def __init__(self, name: Text): - self.__step_context = TStep(name=name) - - def with_variables(self, **variables) -> "RunRequest": - self.__step_context.variables.update(variables) - return self - - def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunRequest": - if assign_var_name: - self.__step_context.setup_hooks.append({assign_var_name: hook}) - else: - self.__step_context.setup_hooks.append(hook) - - return self - - def get(self, url: Text) -> RequestWithOptionalArgs: - self.__step_context.request = TRequest(method=MethodEnum.GET, url=url) - return RequestWithOptionalArgs(self.__step_context) - - def post(self, url: Text) -> RequestWithOptionalArgs: - self.__step_context.request = TRequest(method=MethodEnum.POST, url=url) - return RequestWithOptionalArgs(self.__step_context) - - def put(self, url: Text) -> RequestWithOptionalArgs: - self.__step_context.request = TRequest(method=MethodEnum.PUT, url=url) - return RequestWithOptionalArgs(self.__step_context) - - def head(self, url: Text) -> RequestWithOptionalArgs: - self.__step_context.request = TRequest(method=MethodEnum.HEAD, url=url) - return RequestWithOptionalArgs(self.__step_context) - - def delete(self, url: Text) -> RequestWithOptionalArgs: - self.__step_context.request = TRequest(method=MethodEnum.DELETE, url=url) - return RequestWithOptionalArgs(self.__step_context) - - def options(self, url: Text) -> RequestWithOptionalArgs: - self.__step_context.request = TRequest(method=MethodEnum.OPTIONS, url=url) - return RequestWithOptionalArgs(self.__step_context) - - def patch(self, url: Text) -> RequestWithOptionalArgs: - self.__step_context.request = TRequest(method=MethodEnum.PATCH, url=url) - return RequestWithOptionalArgs(self.__step_context) - - -class StepRefCase(object): - def __init__(self, step_context: TStep): - self.__step_context = step_context - - def teardown_hook(self, hook: Text, assign_var_name: Text = None) -> "StepRefCase": - if assign_var_name: - self.__step_context.teardown_hooks.append({assign_var_name: hook}) - else: - self.__step_context.teardown_hooks.append(hook) - - return self - - def export(self, *var_name: Text) -> "StepRefCase": - self.__step_context.export.extend(var_name) - return self - - def perform(self) -> TStep: - return self.__step_context - - -class RunTestCase(object): - def __init__(self, name: Text): - self.__step_context = TStep(name=name) - - def with_variables(self, **variables) -> "RunTestCase": - self.__step_context.variables.update(variables) - return self - - def setup_hook(self, hook: Text, assign_var_name: Text = None) -> "RunTestCase": - if assign_var_name: - self.__step_context.setup_hooks.append({assign_var_name: hook}) - else: - self.__step_context.setup_hooks.append(hook) - - return self - - def call(self, testcase: Callable) -> StepRefCase: - self.__step_context.testcase = testcase - return StepRefCase(self.__step_context) - - def perform(self) -> TStep: - return self.__step_context - - -class Step(object): - def __init__( - self, - step_context: Union[ - StepRequestValidation, - StepRequestExtraction, - RequestWithOptionalArgs, - RunTestCase, - StepRefCase, - ], - ): - self.__step_context = step_context.perform() - - @property - def request(self) -> TRequest: - return self.__step_context.request - - @property - def testcase(self) -> TestCase: - return self.__step_context.testcase - - def perform(self) -> TStep: - return self.__step_context