GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-05-14 12:57:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: make step extensible to support implementing new protocols and test types for python version

Browse Source
This commit is contained in:
debugtalk
2022-04-01 22:47:47 +08:00
parent e5f994f936
commit d36c1cb987
13 changed files with 842 additions and 802 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/CHANGELOG.md
View File

@@ -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**

12
hrp/step_request.go
View File

@@ -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

4
hrp/step_testcase.go
View File

@@ -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

6
httprunner/__init__.py
View File

@@ -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__",

60
httprunner/config.py Normal file
View File

@@ -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,
)

24
httprunner/models.py
View File

@@ -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

26
httprunner/parser.py
View File

@@ -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)

24
httprunner/response.py
View File

@@ -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__})"

459
httprunner/runner.py
View File

@@ -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

40
httprunner/step.py Normal file
View File

@@ -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)

462
httprunner/step_request.py Normal file
View File

@@ -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)

110
httprunner/step_testcase.py Normal file
View File

@@ -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)

415
httprunner/testcase.py
View File

@@ -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