GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-06-30 20:11:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #342 from HttpRunner/loader

Browse Source 
refactor load testcases
This commit is contained in:
debugtalk
2018-08-12 15:37:46 +08:00
committed by GitHub
parent b250c462bb b665ad79ef
commit 799fcd9386
14 changed files with 716 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files

641
httprunner/loader.py
View File

@@ -9,11 +9,23 @@ import yaml
from httprunner import exceptions, logger, parser, validator from httprunner import exceptions, logger, parser, validator
from httprunner.compat import OrderedDict from httprunner.compat import OrderedDict
project_mapping = {
"debugtalk": {},
"env": {},
"def-api": {},
"def-testcase": {}
}
""" dict: save project loaded api/testcases definitions, environments and debugtalk.py module.
"""
testcases_cache_mapping = {}
############################################################################### ###############################################################################
## file loader ## file loader
############################################################################### ###############################################################################
def _check_format(file_path, content): def _check_format(file_path, content):
""" check testcase format if valid """ check testcase format if valid
""" """
@@ -102,10 +114,14 @@ def load_file(file_path):
def load_folder_files(folder_path, recursive=True): def load_folder_files(folder_path, recursive=True):
""" load folder path, return all files in list format. """ load folder path, return all files endswith yml/yaml/json in list.
@param
folder_path: specified folder path to load Args:
recursive: if True, will load files recursively folder_path (str): specified folder path to load
recursive (bool): load files recursively if True
Returns:
list: files endswith yml/yaml/json
""" """
if isinstance(folder_path, (list, set)): if isinstance(folder_path, (list, set)):
files = [] files = []
@@ -140,6 +156,24 @@ def load_folder_files(folder_path, recursive=True):
def load_dot_env_file(path): def load_dot_env_file(path):
""" load .env file """ load .env file
Args:
path (str): .env file path.
If path is None, it will find .env file in current working directory.
Returns:
dict: environment variables mapping
{
"UserName": "debugtalk",
"Password": "123456",
"PROJECT_KEY": "ABCDEFGH"
}
Raises:
exceptions.FileNotFound: If specified env file is not exist.
exceptions.FileFormatError: If env file format is invalid.
""" """
if not path: if not path:
path = os.path.join(os.getcwd(), ".env") path = os.path.join(os.getcwd(), ".env")
@@ -163,6 +197,7 @@ def load_dot_env_file(path):
env_variables_mapping[variable.strip()] = value.strip() env_variables_mapping[variable.strip()] = value.strip()
project_mapping["env"] = env_variables_mapping
return env_variables_mapping return env_variables_mapping
@@ -289,7 +324,10 @@ def load_debugtalk_module(start_path=None):
} }
imported_module = importlib.import_module(module_name) imported_module = importlib.import_module(module_name)
return load_python_module(imported_module) loaded_module = load_python_module(imported_module)
project_mapping["debugtalk"] = loaded_module
return loaded_module
def get_module_item(module_mapping, item_type, item_name): def get_module_item(module_mapping, item_type, item_name):
@@ -326,88 +364,15 @@ def get_module_item(module_mapping, item_type, item_name):
############################################################################### ###############################################################################
## suite loader ## testcase loader
############################################################################### ###############################################################################
overall_def_dict = {
"api": {},
"suite": {}
}
testcases_cache_mapping = {}
def _load_test_dependencies():
""" load all api and suite definitions.
default api folder is "$CWD/tests/api/".
default suite folder is "$CWD/tests/suite/".
"""
# TODO: cache api and suite loading
# load api definitions
api_def_folder = os.path.join(os.getcwd(), "tests", "api")
for test_file in load_folder_files(api_def_folder):
_load_api_file(test_file)
# load suite definitions
suite_def_folder = os.path.join(os.getcwd(), "tests", "suite")
for suite_file in load_folder_files(suite_def_folder):
suite = _load_test_file(suite_file)
if "def" not in suite["config"]:
raise exceptions.ParamsError("def missed in suite file: {}!".format(suite_file))
call_func = suite["config"]["def"]
function_meta = parser.parse_function(call_func)
suite["function_meta"] = function_meta
overall_def_dict["suite"][function_meta["func_name"]] = suite
def _load_api_file(file_path):
""" load api definition from file and store in overall_def_dict["api"]
api file should be in format below:
[
{
"api": {
"def": "api_login",
"request": {},
"validate": []
}
},
{
"api": {
"def": "api_logout",
"request": {},
"validate": []
}
}
]
"""
api_items = load_file(file_path)
if not isinstance(api_items, list):
raise exceptions.FileFormatError("API format error: {}".format(file_path))
for api_item in api_items:
if not isinstance(api_item, dict) or len(api_item) != 1:
raise exceptions.FileFormatError("API format error: {}".format(file_path))
key, api_dict = api_item.popitem()
if key != "api" or not isinstance(api_dict, dict) or "def" not in api_dict:
raise exceptions.FileFormatError("API format error: {}".format(file_path))
api_def = api_dict.pop("def")
function_meta = parser.parse_function(api_def)
func_name = function_meta["func_name"]
if func_name in overall_def_dict["api"]:
logger.log_warning("API definition duplicated: {}".format(func_name))
api_dict["function_meta"] = function_meta
overall_def_dict["api"][func_name] = api_dict
def _load_test_file(file_path): def _load_test_file(file_path):
""" load testcase file or testsuite file """ load testcase file or testsuite file
@param file_path: absolute valid file path
file_path should be in format below: Args:
file_path (str): absolute valid file path. file_path should be in the following format:
[ [
{ {
"config": { "config": {
@@ -423,6 +388,13 @@ def _load_test_file(file_path):
"validate": [] "validate": []
} }
}, },
{
"test": {
"name": "add product to cart",
"suite": "create_and_check()",
"validate": []
}
},
{ {
"test": { "test": {
"name": "checkout cart", "name": "checkout cart",
@@ -431,19 +403,24 @@ def _load_test_file(file_path):
} }
} }
] ]
@return testset dict
{ Returns:
"config": {}, dict: testcase dict
"testcases": [testcase11, testcase12] {
} "config": {},
"teststeps": [teststep11, teststep12]
}
""" """
testset = { testcase = {
"config": { "config": {
"path": file_path "path": file_path
}, },
"testcases": [] # TODO: rename to tests "teststeps": []
} }
for item in load_file(file_path): for item in load_file(file_path):
# TODO: add json schema validation
if not isinstance(item, dict) or len(item) != 1: if not isinstance(item, dict) or len(item) != 1:
raise exceptions.FileFormatError("Testcase format error: {}".format(file_path)) raise exceptions.FileFormatError("Testcase format error: {}".format(file_path))
@@ -452,43 +429,69 @@ def _load_test_file(file_path):
raise exceptions.FileFormatError("Testcase format error: {}".format(file_path)) raise exceptions.FileFormatError("Testcase format error: {}".format(file_path))
if key == "config": if key == "config":
testset["config"].update(test_block) testcase["config"].update(test_block)
elif key == "test": elif key == "test":
def extend_api_definition(block):
ref_call = block["api"]
def_block = _get_block_by_name(ref_call, "def-api")
_extend_block(block, def_block)
# reference api
if "api" in test_block: if "api" in test_block:
ref_call = test_block["api"] extend_api_definition(test_block)
def_block = _get_block_by_name(ref_call, "api") testcase["teststeps"].append(test_block)
_override_block(def_block, test_block)
testset["testcases"].append(test_block) # reference testcase
elif "suite" in test_block: elif "suite" in test_block: # TODO: replace suite with testcase
ref_call = test_block["suite"] ref_call = test_block["suite"]
block = _get_block_by_name(ref_call, "suite") block = _get_block_by_name(ref_call, "def-testcase")
testset["testcases"].extend(block["testcases"]) # TODO: bugfix lost block config variables
for teststep in block["teststeps"]:
if "api" in teststep:
extend_api_definition(teststep)
testcase["teststeps"].append(teststep)
# define directly
else: else:
testset["testcases"].append(test_block) testcase["teststeps"].append(test_block)
else: else:
logger.log_warning( logger.log_warning(
"unexpected block key: {}. block key should only be 'config' or 'test'.".format(key) "unexpected block key: {}. block key should only be 'config' or 'test'.".format(key)
) )
return testset return testcase
def _get_block_by_name(ref_call, ref_type): def _get_block_by_name(ref_call, ref_type):
""" get test content by reference name """ get test content by reference name.
@params:
ref_call: e.g. api_v1_Account_Login_POST($UserName, $Password) Args:
ref_type: "api" or "suite" ref_call (str): call function.
e.g. api_v1_Account_Login_POST($UserName, $Password)
ref_type (enum): "def-api" or "def-testcase"
Returns:
dict: api/testcase definition.
Raises:
exceptions.ParamsError: call args number is not equal to defined args number.
""" """
function_meta = parser.parse_function(ref_call) function_meta = parser.parse_function(ref_call)
func_name = function_meta["func_name"] func_name = function_meta["func_name"]
call_args = function_meta["args"] call_args = function_meta["args"]
block = _get_test_definition(func_name, ref_type) block = _get_test_definition(func_name, ref_type)
def_args = block.get("function_meta").get("args", []) def_args = block.get("function_meta", {}).get("args", [])
if len(call_args) != len(def_args): if len(call_args) != len(def_args):
raise exceptions.ParamsError("call args mismatch defined args!") err_msg = "{}: call args number is not equal to defined args number!\n".format(func_name)
err_msg += "defined args: {}\n".format(def_args)
err_msg += "reference args: {}".format(call_args)
logger.log_error(err_msg)
raise exceptions.ParamsError(err_msg)
args_mapping = {} args_mapping = {}
for index, item in enumerate(def_args): for index, item in enumerate(def_args):
@@ -505,80 +508,104 @@ def _get_block_by_name(ref_call, ref_type):
def _get_test_definition(name, ref_type): def _get_test_definition(name, ref_type):
""" get expected api or testcase. """ get expected api or testcase.
@params:
name: api or testcase name Args:
ref_type: "api" or "suite" name (str): api or testcase name
@return ref_type (enum): "def-api" or "def-testcase"
expected api info if found, otherwise raise ApiNotFound exception
Returns:
dict: expected api/testcase info if found.
Raises:
exceptions.ApiNotFound: api not found
exceptions.TestcaseNotFound: testcase not found
""" """
block = overall_def_dict.get(ref_type, {}).get(name) block = project_mapping.get(ref_type, {}).get(name)
if not block: if not block:
err_msg = "{} not found!".format(name) err_msg = "{} not found!".format(name)
if ref_type == "api": if ref_type == "def-api":
raise exceptions.ApiNotFound(err_msg) raise exceptions.ApiNotFound(err_msg)
else: else:
# ref_type == "suite": # ref_type == "def-testcase":
raise exceptions.TestcaseNotFound(err_msg) raise exceptions.TestcaseNotFound(err_msg)
return block return block
def _override_block(def_block, current_block): def _extend_block(ref_block, def_block):
""" override def_block with current_block """ extend ref_block with def_block.
@param def_block:
{ Args:
"name": "get token", def_block (dict): api definition dict.
"request": {...}, ref_block (dict): reference block
"validate": [{'eq': ['status_code', 200]}]
} Returns:
@param current_block: dict: extended reference block.
{
"name": "get token", Examples:
"extract": [{"token": "content.token"}], >>> def_block = {
"validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}] "name": "get token 1",
} "request": {...},
@return "validate": [{'eq': ['status_code', 200]}]
{ }
"name": "get token", >>> ref_block = {
"request": {...}, "name": "get token 2",
"extract": [{"token": "content.token"}], "extract": [{"token": "content.token"}],
"validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}] "validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}]
} }
>>> _extend_block(def_block, ref_block)
{
"name": "get token 2",
"request": {...},
"extract": [{"token": "content.token"}],
"validate": [{'eq': ['status_code', 201]}, {'len_eq': ['content.token', 16]}]
}
""" """
# TODO: override variables
def_validators = def_block.get("validate") or def_block.get("validators", []) def_validators = def_block.get("validate") or def_block.get("validators", [])
current_validators = current_block.get("validate") or current_block.get("validators", []) ref_validators = ref_block.get("validate") or ref_block.get("validators", [])
def_extrators = def_block.get("extract") \ def_extrators = def_block.get("extract") \
or def_block.get("extractors") \ or def_block.get("extractors") \
or def_block.get("extract_binds", []) or def_block.get("extract_binds", [])
current_extractors = current_block.get("extract") \ ref_extractors = ref_block.get("extract") \
or current_block.get("extractors") \ or ref_block.get("extractors") \
or current_block.get("extract_binds", []) or ref_block.get("extract_binds", [])
current_block.update(def_block) ref_block.update(def_block)
current_block["validate"] = _merge_validator( ref_block["validate"] = _merge_validator(
def_validators, def_validators,
current_validators ref_validators
) )
current_block["extract"] = _merge_extractor( ref_block["extract"] = _merge_extractor(
def_extrators, def_extrators,
current_extractors ref_extractors
) )
def _get_validators_mapping(validators): def _convert_validators_to_mapping(validators):
""" get validators mapping from api or test validators """ convert validators list to mapping.
@param (list) validators:
[ Args:
{"check": "v1", "expect": 201, "comparator": "eq"}, validators (list): validators in list
{"check": {"b": 1}, "expect": 200, "comparator": "eq"}
] Returns:
@return dict: validators mapping, use (check, comparator) as key.
{
("v1", "eq"): {"check": "v1", "expect": 201, "comparator": "eq"}, Examples:
('{"b": 1}', "eq"): {"check": {"b": 1}, "expect": 200, "comparator": "eq"} >>> validators = [
} {"check": "v1", "expect": 201, "comparator": "eq"},
{"check": {"b": 1}, "expect": 200, "comparator": "eq"}
]
>>> _convert_validators_to_mapping(validators)
{
("v1", "eq"): {"check": "v1", "expect": 201, "comparator": "eq"},
('{"b": 1}', "eq"): {"check": {"b": 1}, "expect": 200, "comparator": "eq"}
}
""" """
validators_mapping = {} validators_mapping = {}
@@ -596,48 +623,66 @@ def _get_validators_mapping(validators):
return validators_mapping return validators_mapping
def _merge_validator(def_validators, current_validators): def _merge_validator(def_validators, ref_validators):
""" merge def_validators with current_validators """ merge def_validators with ref_validators.
@params:
def_validators: [{'eq': ['v1', 200]}, {"check": "s2", "expect": 16, "comparator": "len_eq"}] Args:
current_validators: [{"check": "v1", "expect": 201}, {'len_eq': ['s3', 12]}] def_validators (list):
@return: ref_validators (list):
[
{"check": "v1", "expect": 201, "comparator": "eq"}, Returns:
{"check": "s2", "expect": 16, "comparator": "len_eq"}, list: merged validators
{"check": "s3", "expect": 12, "comparator": "len_eq"}
] Examples:
>>> def_validators = [{'eq': ['v1', 200]}, {"check": "s2", "expect": 16, "comparator": "len_eq"}]
>>> ref_validators = [{"check": "v1", "expect": 201}, {'len_eq': ['s3', 12]}]
>>> _merge_validator(def_validators, ref_validators)
[
{"check": "v1", "expect": 201, "comparator": "eq"},
{"check": "s2", "expect": 16, "comparator": "len_eq"},
{"check": "s3", "expect": 12, "comparator": "len_eq"}
]
""" """
if not def_validators: if not def_validators:
return current_validators return ref_validators
elif not current_validators: elif not ref_validators:
return def_validators return def_validators
else: else:
api_validators_mapping = _get_validators_mapping(def_validators) def_validators_mapping = _convert_validators_to_mapping(def_validators)
test_validators_mapping = _get_validators_mapping(current_validators) ref_validators_mapping = _convert_validators_to_mapping(ref_validators)
api_validators_mapping.update(test_validators_mapping) def_validators_mapping.update(ref_validators_mapping)
return list(api_validators_mapping.values()) return list(def_validators_mapping.values())
def _merge_extractor(def_extrators, current_extractors): def _merge_extractor(def_extrators, ref_extractors):
""" merge def_extrators with current_extractors """ merge def_extrators with ref_extractors
@params:
def_extrators: [{"var1": "val1"}, {"var2": "val2"}] Args:
current_extractors: [{"var1": "val111"}, {"var3": "val3"}] def_extrators (list): [{"var1": "val1"}, {"var2": "val2"}]
@return: ref_extractors (list): [{"var1": "val111"}, {"var3": "val3"}]
[
{"var1": "val111"}, Returns:
{"var2": "val2"}, list: merged extractors
{"var3": "val3"}
] Examples:
>>> def_extrators = [{"var1": "val1"}, {"var2": "val2"}]
>>> ref_extractors = [{"var1": "val111"}, {"var3": "val3"}]
>>> _merge_extractor(def_extrators, ref_extractors)
[
{"var1": "val111"},
{"var2": "val2"},
{"var3": "val3"}
]
""" """
if not def_extrators: if not def_extrators:
return current_extractors return ref_extractors
elif not current_extractors: elif not ref_extractors:
return def_extrators return def_extrators
else: else:
@@ -650,7 +695,7 @@ def _merge_extractor(def_extrators, current_extractors):
var_name = list(api_extrator.keys())[0] var_name = list(api_extrator.keys())[0]
extractor_dict[var_name] = api_extrator[var_name] extractor_dict[var_name] = api_extrator[var_name]
for test_extrator in current_extractors: for test_extrator in ref_extractors:
if len(test_extrator) != 1: if len(test_extrator) != 1:
logger.log_warning("incorrect extractor: {}".format(test_extrator)) logger.log_warning("incorrect extractor: {}".format(test_extrator))
continue continue
@@ -665,17 +710,213 @@ def _merge_extractor(def_extrators, current_extractors):
return extractor_list return extractor_list
def load_folder_content(folder_path):
""" load api/testcases/testsuites definitions from folder.
Args:
folder_path (str): api/testcases/testsuites files folder.
Returns:
dict: api definition mapping.
{
"tests/api/basic.yml": [
{"api": {"def": "api_login", "request": {}, "validate": []}},
{"api": {"def": "api_logout", "request": {}, "validate": []}}
]
}
"""
items_mapping = {}
for file_path in load_folder_files(folder_path):
items_mapping[file_path] = load_file(file_path)
return items_mapping
def load_api_folder(api_folder_path=None):
""" load api definitions from api folder.
Args:
api_folder_path (str): api files folder.
api file should be in the following format:
[
{
"api": {
"def": "api_login",
"request": {},
"validate": []
}
},
{
"api": {
"def": "api_logout",
"request": {},
"validate": []
}
}
]
Returns:
dict: api definition mapping.
{
"api_login": {
"function_meta": {"func_name": "api_login", "args": [], "kwargs": {}}
"request": {}
},
"api_logout": {
"function_meta": {"func_name": "api_logout", "args": [], "kwargs": {}}
"request": {}
}
}
"""
api_definition_mapping = {}
api_folder_path = api_folder_path or os.path.join(os.getcwd(), "api")
api_items_mapping = load_folder_content(api_folder_path)
for api_file_path, api_items in api_items_mapping.items():
# TODO: add JSON schema validation
for api_item in api_items:
key, api_dict = api_item.popitem()
api_def = api_dict.pop("def")
function_meta = parser.parse_function(api_def)
func_name = function_meta["func_name"]
if func_name in api_definition_mapping:
logger.log_warning("API definition duplicated: {}".format(func_name))
api_dict["function_meta"] = function_meta
api_definition_mapping[func_name] = api_dict
project_mapping["def-api"] = api_definition_mapping
return api_definition_mapping
def load_test_folder(test_folder_path=None):
""" load testcases definitions from folder.
Args:
test_folder_path (str): testcases files folder.
testcase file should be in the following format:
[
{
"config": {
"def": "create_and_check",
"request": {},
"validate": []
}
},
{
"test": {
"api": "get_user",
"validate": []
}
}
]
Returns:
dict: testcases definition mapping.
{
"create_and_check": [
{"config": {}},
{"test": {}},
{"test": {}}
],
"tests/testcases/create_and_get.yml": [
{"config": {}},
{"test": {}},
{"test": {}}
]
}
"""
test_definition_mapping = {}
# TODO: replace suite with testcases
test_folder_path = test_folder_path or os.path.join(os.getcwd(), "suite")
test_items_mapping = load_folder_content(test_folder_path)
for test_file_path, items in test_items_mapping.items():
# TODO: add JSON schema validation
testcase = {
"config": {
"path": test_file_path
},
"teststeps": []
}
for item in items:
key, block = item.popitem()
if key == "config":
testcase["config"].update(block)
if "def" not in block:
test_definition_mapping[test_file_path] = testcase
continue
testcase_def = block.pop("def")
function_meta = parser.parse_function(testcase_def)
func_name = function_meta["func_name"]
if func_name in test_definition_mapping:
logger.log_warning("API definition duplicated: {}".format(func_name))
testcase["function_meta"] = function_meta
test_definition_mapping[func_name] = testcase
else:
# key == "test":
testcase["teststeps"].append(block)
project_mapping["def-testcase"] = test_definition_mapping
return test_definition_mapping
def load_project_tests(folder_path=None):
""" load api, testcases and debugtalk.py module.
Args:
folder_path (str): folder path.
If not set, defautls to current working directory.
Returns:
dict: project tests mapping.
"""
folder_path = folder_path or os.getcwd()
load_debugtalk_module(folder_path)
load_api_folder(os.path.join(folder_path, "api"))
load_test_folder(os.path.join(folder_path, "suite"))
return project_mapping
def load_testcases(path): def load_testcases(path):
""" load testcases from file path """ load testcases from file path
@param path: path could be in several type
- absolute/relative file path Args:
- absolute/relative folder path path (str): testcase file/foler path.
- list/set container with file(s) and/or folder(s) path could be in several types:
@return testcases list, each testcase is corresponding to a file - absolute/relative file path
- absolute/relative folder path
- list/set container with file(s) and/or folder(s)
Returns:
list: testcases list, each testcase is corresponding to a file
[ [
testcase_dict_1, testcase_dict_1,
testcase_dict_2 testcase_dict_2
] ]
""" """
if isinstance(path, (list, set)): if isinstance(path, (list, set)):
testcases_list = [] testcases_list = []
@@ -701,7 +942,7 @@ def load_testcases(path):
elif os.path.isfile(path): elif os.path.isfile(path):
try: try:
testcase = _load_test_file(path) testcase = _load_test_file(path)
if testcase["testcases"]: if testcase["teststeps"]:
testcases_list = [testcase] testcases_list = [testcase]
else: else:
testcases_list = [] testcases_list = []
@@ -715,15 +956,3 @@ def load_testcases(path):
testcases_cache_mapping[path] = testcases_list testcases_cache_mapping[path] = testcases_list
return testcases_list return testcases_list
def load(path):
""" main interface for loading testcases
@param (str) path: testcase file/folder path
@return (list) testcases list
"""
if validator.is_testcases(path):
return path
_load_test_dependencies()
return load_testcases(path)

2
httprunner/locusts.py
View File

@@ -40,7 +40,7 @@ def gen_locustfile(testcase_file_path):
"templates", "templates",
"locustfile_template" "locustfile_template"
) )
testcases = loader.load(testcase_file_path) testcases = loader.load_testcases(testcase_file_path)
host = testcases[0].get("config", {}).get("request", {}).get("base_url", "") host = testcases[0].get("config", {}).get("request", {}).get("base_url", "")
with io.open(template_path, encoding='utf-8') as template: with io.open(template_path, encoding='utf-8') as template:

177
httprunner/task.py
View File

@@ -4,7 +4,8 @@ import copy
import sys import sys
import unittest import unittest
from httprunner import context, exceptions, loader, logger, runner, utils from httprunner import (context, exceptions, loader, logger, runner, utils,
validator)
from httprunner.compat import is_py3 from httprunner.compat import is_py3
from httprunner.report import (HtmlTestResult, get_platform, get_summary, from httprunner.report import (HtmlTestResult, get_platform, get_summary,
render_html_report) render_html_report)
@@ -33,39 +34,39 @@ class TestCase(unittest.TestCase):
class TestSuite(unittest.TestSuite): class TestSuite(unittest.TestSuite):
""" create test suite with a testset, it may include one or several testcases. """ create test suite with a testcase, it may include one or several teststeps.
each suite should initialize a separate Runner() with testset config. each suite should initialize a separate Runner() with testcase config.
@param
(dict) testset Args:
testcase (dict): testcase dict
{ {
"name": "testset description",
"config": { "config": {
"name": "testset description", "name": "testcase description",
"parameters": {}, "parameters": {},
"variables": [], "variables": [],
"request": {}, "request": {},
"output": [] "output": []
}, },
"testcases": [ "teststeps": [
{ {
"name": "testcase description", "name": "teststep1 description",
"parameters": {}, "parameters": {},
"variables": [], # optional, override "variables": [], # optional, override
"request": {}, "request": {},
"extract": {}, # optional "extract": {}, # optional
"validate": {} # optional "validate": {} # optional
}, },
testcase12 teststep2
] ]
} }
(dict) variables_mapping: variables_mapping (dict): passed in variables mapping, it will override variables in config block.
passed in variables mapping, it will override variables in config block
""" """
def __init__(self, testset, variables_mapping=None, http_client_session=None): def __init__(self, testcase, variables_mapping=None, http_client_session=None):
super(TestSuite, self).__init__() super(TestSuite, self).__init__()
self.test_runner_list = [] self.test_runner_list = []
self.config = testset.get("config", {}) self.config = testcase.get("config", {})
self.output_variables_list = self.config.get("output", []) self.output_variables_list = self.config.get("output", [])
self.testset_file_path = self.config.get("path") self.testset_file_path = self.config.get("path")
config_dict_parameters = self.config.get("parameters", []) config_dict_parameters = self.config.get("parameters", [])
@@ -79,22 +80,22 @@ class TestSuite(unittest.TestSuite):
config_dict_parameters config_dict_parameters
) )
self.testcase_parser = context.TestcaseParser() self.testcase_parser = context.TestcaseParser()
testcases = testset.get("testcases", []) teststeps = testcase.get("teststeps", [])
for config_variables in config_parametered_variables_list: for config_variables in config_parametered_variables_list:
# config level # config level
self.config["variables"] = config_variables self.config["variables"] = config_variables
test_runner = runner.Runner(self.config, http_client_session) test_runner = runner.Runner(self.config, http_client_session)
for testcase_dict in testcases: for teststep_dict in teststeps:
testcase_dict = copy.copy(testcase_dict) teststep_dict = copy.copy(teststep_dict)
# testcase level # testcase level
testcase_parametered_variables_list = self._get_parametered_variables( testcase_parametered_variables_list = self._get_parametered_variables(
testcase_dict.get("variables", []), teststep_dict.get("variables", []),
testcase_dict.get("parameters", []) teststep_dict.get("parameters", [])
) )
for testcase_variables in testcase_parametered_variables_list: for testcase_variables in testcase_parametered_variables_list:
testcase_dict["variables"] = testcase_variables teststep_dict["variables"] = testcase_variables
# eval testcase name with bind variables # eval testcase name with bind variables
variables = utils.override_variables_binds( variables = utils.override_variables_binds(
@@ -103,13 +104,13 @@ class TestSuite(unittest.TestSuite):
) )
self.testcase_parser.update_binded_variables(variables) self.testcase_parser.update_binded_variables(variables)
try: try:
testcase_name = self.testcase_parser.eval_content_with_bindings(testcase_dict["name"]) testcase_name = self.testcase_parser.eval_content_with_bindings(teststep_dict["name"])
except (AssertionError, exceptions.ParamsError): except (AssertionError, exceptions.ParamsError):
logger.log_warning("failed to eval testcase name: {}".format(testcase_dict["name"])) logger.log_warning("failed to eval testcase name: {}".format(teststep_dict["name"]))
testcase_name = testcase_dict["name"] testcase_name = teststep_dict["name"]
self.test_runner_list.append((test_runner, variables)) self.test_runner_list.append((test_runner, variables))
self._add_test_to_suite(testcase_name, test_runner, testcase_dict) self._add_test_to_suite(testcase_name, test_runner, teststep_dict)
def _get_parametered_variables(self, variables, parameters): def _get_parametered_variables(self, variables, parameters):
""" parameterize varaibles with parameters """ parameterize varaibles with parameters
@@ -159,38 +160,47 @@ class TestSuite(unittest.TestSuite):
return outputs return outputs
def init_test_suites(path_or_testsets, mapping=None, http_client_session=None): def init_test_suites(path_or_testcases, mapping=None, http_client_session=None):
""" initialize TestSuite list with testset path or testset dict """ initialize TestSuite list with testcase path or testcase(s).
@params
testsets (dict/list): testset or list of testset Args:
testset_dict path_or_testcases (str/dict/list): testcase file path or testcase dict or testcases list
testcase_dict
or or
[ [
testset_dict_1, testcase_dict_1,
testset_dict_2, testcase_dict_2,
{ {
"config": {}, "config": {},
"api": {}, "teststeps": [teststep11, teststep12]
"testcases": [testcase11, testcase12]
} }
] ]
mapping (dict):
passed in variables mapping, it will override variables in config block mapping (dict): passed in variables mapping, it will override variables in config block.
http_client_session (instance): requests.Session(), or locusts.client.Session() instance.
Returns:
list: TestSuite() instance list.
""" """
testsets = loader.load(path_or_testsets) if validator.is_testcases(path_or_testcases):
testcases = path_or_testcases
else:
testcases = loader.load_testcases(path_or_testcases)
# TODO: move comparator uniform here # TODO: move comparator uniform here
mapping = mapping or {} mapping = mapping or {}
if not testsets: if not testcases:
raise exceptions.TestcaseNotFound raise exceptions.TestcaseNotFound
if isinstance(testsets, dict): if isinstance(testcases, dict):
testsets = [testsets] testcases = [testcases]
test_suite_list = [] test_suite_list = []
for testset in testsets: for testcase in testcases:
test_suite = TestSuite(testset, mapping, http_client_session) test_suite = TestSuite(testcase, mapping, http_client_session)
test_suite_list.append(test_suite) test_suite_list.append(test_suite)
return test_suite_list return test_suite_list
@@ -199,39 +209,55 @@ def init_test_suites(path_or_testsets, mapping=None, http_client_session=None):
class HttpRunner(object): class HttpRunner(object):
def __init__(self, **kwargs): def __init__(self, **kwargs):
""" initialize test runner """ initialize HttpRunner.
@param (dict) kwargs: key-value arguments used to initialize TextTestRunner
- resultclass: HtmlTestResult or TextTestResult Args:
- failfast: False/True, stop the test run on the first error or failure. kwargs (dict): key-value arguments used to initialize TextTestRunner.
- dot_env_path: .env file path Commonly used arguments:
resultclass (class): HtmlTestResult or TextTestResult
failfast (bool): False/True, stop the test run on the first error or failure.
dot_env_path (str): .env file path.
Attributes:
project_mapping (dict): save project loaded api/testcases, environments and debugtalk.py module.
""" """
dot_env_path = kwargs.pop("dot_env_path", None) dot_env_path = kwargs.pop("dot_env_path", None)
utils.set_os_environ(loader.load_dot_env_file(dot_env_path)) loader.load_dot_env_file(dot_env_path)
loader.load_project_tests("tests") # TODO: remove tests
self.project_mapping = loader.project_mapping
utils.set_os_environ(self.project_mapping["env"])
kwargs.setdefault("resultclass", HtmlTestResult) kwargs.setdefault("resultclass", HtmlTestResult)
self.runner = unittest.TextTestRunner(**kwargs) self.runner = unittest.TextTestRunner(**kwargs)
def run(self, path_or_testsets, mapping=None): def run(self, path_or_testcases, mapping=None):
""" start to run test with varaibles mapping """ start to run test with varaibles mapping.
@param path_or_testsets: YAML/JSON testset file path or testset list
path: path could be in several type Args:
- absolute/relative file path path_or_testcases (str/list/dict): YAML/JSON testcase file path or testcase list
- absolute/relative folder path path: path could be in several type
- list/set container with file(s) and/or folder(s) - absolute/relative file path
testsets: testset or list of testset - absolute/relative folder path
- (dict) testset_dict - list/set container with file(s) and/or folder(s)
- (list) list of testset_dict testcases: testcase dict or list of testcases
[ - (dict) testset_dict
testset_dict_1, - (list) list of testset_dict
testset_dict_2 [
] testset_dict_1,
@param (dict) mapping: testset_dict_2
if mapping specified, it will override variables in config block ]
mapping (dict): if mapping specified, it will override variables in config block.
Returns:
instance: HttpRunner() instance
""" """
try: try:
test_suite_list = init_test_suites(path_or_testsets, mapping) test_suite_list = init_test_suites(path_or_testcases, mapping)
except exceptions.TestcaseNotFound: except exceptions.TestcaseNotFound:
logger.log_error("Testcases not found in {}".format(path_or_testsets)) logger.log_error("Testcases not found in {}".format(path_or_testcases))
sys.exit(1) sys.exit(1)
self.summary = { self.summary = {
@@ -243,8 +269,7 @@ class HttpRunner(object):
} }
def accumulate_stat(origin_stat, new_stat): def accumulate_stat(origin_stat, new_stat):
""" accumulate new_stat to origin_stat """accumulate new_stat to origin_stat."""
"""
for key in new_stat: for key in new_stat:
if key not in origin_stat: if key not in origin_stat:
origin_stat[key] = new_stat[key] origin_stat[key] = new_stat[key]
@@ -272,11 +297,15 @@ class HttpRunner(object):
return self return self
def gen_html_report(self, html_report_name=None, html_report_template=None): def gen_html_report(self, html_report_name=None, html_report_template=None):
""" generate html report and return report path """ generate html report and return report path.
@param (str) html_report_name:
output html report file name Args:
@param (str) html_report_template: html_report_name (str): output html report file name
report template file path, template should be in Jinja2 format html_report_template (str): report template file path, template should be in Jinja2 format
Returns:
str: generated html report path
""" """
return render_html_report( return render_html_report(
self.summary, self.summary,
@@ -287,8 +316,8 @@ class HttpRunner(object):
class LocustTask(object): class LocustTask(object):
def __init__(self, path_or_testsets, locust_client, mapping=None): def __init__(self, path_or_testcases, locust_client, mapping=None):
self.test_suite_list = init_test_suites(path_or_testsets, mapping, locust_client) self.test_suite_list = init_test_suites(path_or_testcases, mapping, locust_client)
def run(self): def run(self):
for test_suite in self.test_suite_list: for test_suite in self.test_suite_list:

18
httprunner/utils.py
View File

@@ -1,34 +1,16 @@
# encoding: utf-8 # encoding: utf-8
import copy import copy
import hashlib
import hmac
import io import io
import itertools import itertools
import json import json
import os.path import os.path
import random
import string import string
from datetime import datetime from datetime import datetime
from httprunner import exceptions, logger from httprunner import exceptions, logger
from httprunner.compat import OrderedDict, basestring, is_py2 from httprunner.compat import OrderedDict, basestring, is_py2
SECRET_KEY = "DebugTalk"
def gen_random_string(str_len):
return ''.join(
random.choice(string.ascii_letters + string.digits) for _ in range(str_len))
def gen_md5(*str_args):
return hashlib.md5("".join(str_args).encode('utf-8')).hexdigest()
def get_sign(*args):
content = ''.join(args).encode('ascii')
sign_key = SECRET_KEY.encode('ascii')
sign = hmac.new(sign_key, content, hashlib.sha1).hexdigest()
return sign
def remove_prefix(text, prefix): def remove_prefix(text, prefix):
""" remove prefix from text """ remove prefix from text

49
httprunner/validator.py
View File

@@ -6,35 +6,48 @@ TODO: refactor with JSON schema validate
""" """
def is_testcase(data_structure): def is_testcase(data_structure):
""" check if data_structure is a testcase """ check if data_structure is a testcase.
testcase should always be in the following data structure:
{ Args:
"name": "desc1", data_structure (dict): testcase should always be in the following data structure:
"config": {},
"api": {}, {
"testcases": [testcase11, testcase12] "name": "desc1",
} "config": {},
"api": {},
"testcases": [testcase11, testcase12]
}
Returns:
bool: True if data_structure is valid testcase, otherwise False.
""" """
if not isinstance(data_structure, dict): if not isinstance(data_structure, dict):
return False return False
if "name" not in data_structure or "testcases" not in data_structure: if "name" not in data_structure or "teststeps" not in data_structure:
return False return False
if not isinstance(data_structure["testcases"], list): if not isinstance(data_structure["teststeps"], list):
return False return False
return True return True
def is_testcases(data_structure): def is_testcases(data_structure):
""" check if data_structure is testcase or testcases list """ check if data_structure is testcase or testcases list.
testsets should always be in the following data structure:
testset_dict Args:
or data_structure (dict): testcase(s) should always be in the following data structure:
[
testset_dict_1, testcase_dict
testset_dict_2 or
] [
testcase_dict_1,
testcase_dict_2
]
Returns:
bool: True if data_structure is valid testcase(s), otherwise False.
""" """
if not isinstance(data_structure, list): if not isinstance(data_structure, list):
return is_testcase(data_structure) return is_testcase(data_structure)

31
tests/api_server.py
View File

@@ -1,9 +1,23 @@
import hashlib import hashlib
import hmac
import json import json
from functools import wraps from functools import wraps
from httprunner import utils
from flask import Flask, make_response, request from flask import Flask, make_response, request
from httprunner.built_in import gen_random_string
try:
from httpbin import app as httpbin_app
HTTPBIN_HOST = "127.0.0.1"
HTTPBIN_PORT = 3458
except ImportError:
httpbin_app = None
HTTPBIN_HOST = "httpbin.org"
HTTPBIN_PORT = 80
FLASK_APP_PORT = 5000
HTTPBIN_SERVER = "http://{}:{}".format(HTTPBIN_HOST, HTTPBIN_PORT)
SECRET_KEY = "DebugTalk"
app = Flask(__name__) app = Flask(__name__)
@@ -31,6 +45,17 @@ data structure:
""" """
token_dict = {} token_dict = {}
def get_sign(*args):
content = ''.join(args).encode('ascii')
sign_key = SECRET_KEY.encode('ascii')
sign = hmac.new(sign_key, content, hashlib.sha1).hexdigest()
return sign
def gen_md5(*args):
return hashlib.md5("".join(args).encode('utf-8')).hexdigest()
def validate_request(func): def validate_request(func):
@wraps(func) @wraps(func)
@@ -74,7 +99,7 @@ def get_token():
data = request.get_json() data = request.get_json()
sign = data.get('sign', "") sign = data.get('sign', "")
expected_sign = utils.get_sign(user_agent, device_sn, os_platform, app_version) expected_sign = get_sign(user_agent, device_sn, os_platform, app_version)
if expected_sign != sign: if expected_sign != sign:
result = { result = {
@@ -83,7 +108,7 @@ def get_token():
} }
response = make_response(json.dumps(result), 403) response = make_response(json.dumps(result), 403)
else: else:
token = utils.gen_random_string(16) token = gen_random_string(16)
token_dict[device_sn] = token token_dict[device_sn] = token
result = { result = {

21
tests/base.py
View File

@@ -3,26 +3,17 @@ import time
import unittest import unittest
import requests import requests
from httprunner import utils from tests.api_server import FLASK_APP_PORT, HTTPBIN_HOST, HTTPBIN_PORT
from tests.api_server import app as flask_app from tests.api_server import app as flask_app
from tests.api_server import gen_md5, gen_random_string, get_sign, httpbin_app
try:
from httpbin import app as httpbin_app
HTTPBIN_HOST = "127.0.0.1"
HTTPBIN_PORT = 3458
except ImportError:
HTTPBIN_HOST = "httpbin.org"
HTTPBIN_PORT = 80
FLASK_APP_PORT = 5000
HTTPBIN_SERVER = "http://{}:{}".format(HTTPBIN_HOST, HTTPBIN_PORT)
def run_flask(): def run_flask():
flask_app.run(port=FLASK_APP_PORT) flask_app.run(port=FLASK_APP_PORT)
def run_httpbin(): def run_httpbin():
if HTTPBIN_HOST == "127.0.0.1": if httpbin_app:
httpbin_app.run(host=HTTPBIN_HOST, port=HTTPBIN_PORT) httpbin_app.run(host=HTTPBIN_HOST, port=HTTPBIN_PORT)
@@ -59,7 +50,7 @@ class ApiServerUnittest(unittest.TestCase):
'app_version': app_version 'app_version': app_version
} }
data = { data = {
'sign': utils.get_sign(user_agent, device_sn, os_platform, app_version) 'sign': get_sign(user_agent, device_sn, os_platform, app_version)
} }
resp = self.api_client.post(url, json=data, headers=headers) resp = self.api_client.post(url, json=data, headers=headers)
@@ -71,7 +62,7 @@ class ApiServerUnittest(unittest.TestCase):
def get_authenticated_headers(self): def get_authenticated_headers(self):
user_agent = 'iOS/10.3' user_agent = 'iOS/10.3'
device_sn = utils.gen_random_string(15) device_sn = gen_random_string(15)
os_platform = 'ios' os_platform = 'ios'
app_version = '2.8.6' app_version = '2.8.6'

25
tests/debugtalk.py
View File

@@ -1,34 +1,13 @@
import hashlib
import hmac
import json import json
import os import os
import random import random
import string import string
import time import time
from tests.base import HTTPBIN_SERVER from tests.api_server import HTTPBIN_SERVER, SECRET_KEY, gen_md5, get_sign
try:
import urllib
except NameError:
import urllib.parse as urllib
SECRET_KEY = "DebugTalk"
BASE_URL = "http://127.0.0.1:5000" BASE_URL = "http://127.0.0.1:5000"
def get_sign(*args):
content = ''.join(args).encode('ascii')
sign_key = SECRET_KEY.encode('ascii')
sign = hmac.new(sign_key, content, hashlib.sha1).hexdigest()
return sign
get_sign_lambda = lambda *args: hmac.new(
'DebugTalk'.encode('ascii'),
''.join(args).encode('ascii'),
hashlib.sha1).hexdigest()
def gen_md5(*args):
return hashlib.md5("".join(args).encode('utf-8')).hexdigest()
def sum_status_code(status_code, expect_sum): def sum_status_code(status_code, expect_sum):
""" sum status code digits """ sum status code digits
@@ -62,8 +41,6 @@ def get_account():
{"username": "user2", "password": "222222"} {"username": "user2", "password": "222222"}
] ]
SECRET_KEY = "DebugTalk"
def gen_random_string(str_len): def gen_random_string(str_len):
random_char_list = [] random_char_list = []
for _ in range(str_len): for _ in range(str_len):

2
tests/test_context.py
View File

@@ -4,7 +4,6 @@ import unittest
import requests import requests
from httprunner import context, exceptions, loader, parser, response, runner from httprunner import context, exceptions, loader, parser, response, runner
from httprunner.utils import gen_md5
from tests.base import ApiServerUnittest from tests.base import ApiServerUnittest
@@ -120,6 +119,7 @@ class TestContext(ApiServerUnittest):
{"authorization": "${gen_md5($TOKEN, $data, $random)}"} {"authorization": "${gen_md5($TOKEN, $data, $random)}"}
] ]
from tests import debugtalk from tests import debugtalk
from tests.debugtalk import gen_md5
self.context.import_module_items(debugtalk) self.context.import_module_items(debugtalk)
self.context.bind_variables(variables) self.context.bind_variables(variables)
context_variables = self.context.testcase_variables_mapping context_variables = self.context.testcase_variables_mapping

7
tests/test_httprunner.py
View File

@@ -2,7 +2,8 @@ import os
import shutil import shutil
from httprunner import HttpRunner from httprunner import HttpRunner
from tests.base import HTTPBIN_SERVER, ApiServerUnittest from tests.api_server import HTTPBIN_SERVER
from tests.base import ApiServerUnittest
class TestHttpRunner(ApiServerUnittest): class TestHttpRunner(ApiServerUnittest):
@@ -22,7 +23,7 @@ class TestHttpRunner(ApiServerUnittest):
'output': ['token'] 'output': ['token']
}, },
'api': {}, 'api': {},
'testcases': [ 'teststeps': [
{ {
'name': '/api/get-token', 'name': '/api/get-token',
'request': { 'request': {
@@ -113,7 +114,7 @@ class TestHttpRunner(ApiServerUnittest):
testsets = [ testsets = [
{ {
"name": "post data", "name": "post data",
"testcases": [ "teststeps": [
{ {
"name": "post data", "name": "post data",
"request": { "request": {

161
tests/test_loader.py
View File

@@ -1,7 +1,7 @@
import os import os
import unittest import unittest
from httprunner import exceptions, loader, validator from httprunner import exceptions, loader, task, validator
class TestFileLoader(unittest.TestCase): class TestFileLoader(unittest.TestCase):
@@ -197,17 +197,17 @@ class TestModuleLoader(unittest.TestCase):
from httprunner import utils from httprunner import utils
module_mapping = loader.load_python_module(utils) module_mapping = loader.load_python_module(utils)
gen_md5 = loader.get_module_item(module_mapping, "functions", "gen_md5") get_uniform_comparator = loader.get_module_item(
self.assertTrue(validator.is_function(("gen_md5", gen_md5))) module_mapping, "functions", "get_uniform_comparator")
self.assertEqual(gen_md5("abc"), "900150983cd24fb0d6963f7d28e17f72") self.assertTrue(validator.is_function(("get_uniform_comparator", get_uniform_comparator)))
self.assertEqual(get_uniform_comparator("=="), "equals")
with self.assertRaises(exceptions.FunctionNotFound): with self.assertRaises(exceptions.FunctionNotFound):
loader.get_module_item(module_mapping, "functions", "gen_md4") loader.get_module_item(module_mapping, "functions", "gen_md4")
def test_get_module_item_variables(self): def test_get_module_item_variables(self):
from httprunner import utils from tests import debugtalk
module_mapping = loader.load_python_module(utils) module_mapping = loader.load_python_module(debugtalk)
SECRET_KEY = loader.get_module_item(module_mapping, "variables", "SECRET_KEY") SECRET_KEY = loader.get_module_item(module_mapping, "variables", "SECRET_KEY")
self.assertTrue(validator.is_variable(("SECRET_KEY", SECRET_KEY))) self.assertTrue(validator.is_variable(("SECRET_KEY", SECRET_KEY)))
@@ -219,60 +219,34 @@ class TestModuleLoader(unittest.TestCase):
class TestSuiteLoader(unittest.TestCase): class TestSuiteLoader(unittest.TestCase):
def setUp(self): @classmethod
loader.overall_def_dict = { def setUpClass(cls):
"api": {}, project_dir = os.path.join(os.getcwd(), "tests")
"suite": {} loader.load_project_tests(project_dir)
}
def test_load_test_dependencies(self):
loader._load_test_dependencies()
overall_def_dict = loader.overall_def_dict
self.assertIn("get_token", overall_def_dict["api"])
self.assertIn("create_and_check", overall_def_dict["suite"])
def test_load_api_file(self):
loader._load_api_file("tests/api/basic.yml")
overall_api_def_dict = loader.overall_def_dict["api"]
self.assertIn("get_token",overall_api_def_dict)
self.assertEqual("/api/get-token", overall_api_def_dict["get_token"]["request"]["url"])
self.assertIn("$user_agent", overall_api_def_dict["get_token"]["function_meta"]["args"])
self.assertEqual(len(overall_api_def_dict["get_token"]["validate"]), 3)
def test_load_test_file_suite(self):
loader._load_api_file("tests/api/basic.yml")
testset = loader._load_test_file("tests/suite/create_and_get.yml")
self.assertEqual(testset["config"]["name"], "create user and check result.")
self.assertEqual(len(testset["testcases"]), 3)
self.assertEqual(testset["testcases"][0]["name"], "make sure user $uid does not exist")
self.assertEqual(testset["testcases"][0]["request"]["url"], "/api/users/$uid")
def test_load_test_file_testcase(self): def test_load_test_file_testcase(self):
loader._load_test_dependencies() testcase = loader._load_test_file("tests/testcases/smoketest.yml")
testset = loader._load_test_file("tests/testcases/smoketest.yml") self.assertEqual(testcase["config"]["name"], "smoketest")
self.assertEqual(testset["config"]["name"], "smoketest") self.assertEqual(testcase["config"]["path"], "tests/testcases/smoketest.yml")
self.assertEqual(testset["config"]["path"], "tests/testcases/smoketest.yml") self.assertIn("device_sn", testcase["config"]["variables"][0])
self.assertIn("device_sn", testset["config"]["variables"][0]) self.assertEqual(len(testcase["teststeps"]), 8)
self.assertEqual(len(testset["testcases"]), 8) self.assertEqual(testcase["teststeps"][0]["name"], "get token")
self.assertEqual(testset["testcases"][0]["name"], "get token")
def test_get_block_by_name(self): def test_get_block_by_name(self):
loader._load_test_dependencies()
ref_call = "get_user($uid, $token)" ref_call = "get_user($uid, $token)"
block = loader._get_block_by_name(ref_call, "api") block = loader._get_block_by_name(ref_call, "def-api")
self.assertEqual(block["request"]["url"], "/api/users/$uid") self.assertEqual(block["request"]["url"], "/api/users/$uid")
self.assertEqual(block["function_meta"]["func_name"], "get_user") self.assertEqual(block["function_meta"]["func_name"], "get_user")
self.assertEqual(block["function_meta"]["args"], ['$uid', '$token']) self.assertEqual(block["function_meta"]["args"], ['$uid', '$token'])
def test_get_block_by_name_args_mismatch(self): def test_get_block_by_name_args_mismatch(self):
loader._load_test_dependencies()
ref_call = "get_user($uid, $token, $var)" ref_call = "get_user($uid, $token, $var)"
with self.assertRaises(exceptions.ParamsError): with self.assertRaises(exceptions.ParamsError):
loader._get_block_by_name(ref_call, "api") loader._get_block_by_name(ref_call, "def-api")
def test_override_block(self): def test_override_block(self):
loader._load_test_dependencies() def_block = loader._get_block_by_name(
def_block = loader._get_block_by_name("get_token($user_agent, $device_sn, $os_platform, $app_version)", "api") "get_token($user_agent, $device_sn, $os_platform, $app_version)", "def-api")
test_block = { test_block = {
"name": "override block", "name": "override block",
"variables": [ "variables": [
@@ -286,28 +260,26 @@ class TestSuiteLoader(unittest.TestCase):
] ]
} }
loader._override_block(def_block, test_block) loader._extend_block(test_block, def_block)
self.assertEqual(test_block["name"], "override block") self.assertEqual(test_block["name"], "override block")
self.assertIn({'check': 'status_code', 'expect': 201, 'comparator': 'eq'}, test_block["validate"]) self.assertIn({'check': 'status_code', 'expect': 201, 'comparator': 'eq'}, test_block["validate"])
self.assertIn({'check': 'content.token', 'comparator': 'len_eq', 'expect': 32}, test_block["validate"]) self.assertIn({'check': 'content.token', 'comparator': 'len_eq', 'expect': 32}, test_block["validate"])
def test_get_test_definition_api(self): def test_get_test_definition_api(self):
loader._load_test_dependencies() api_def = loader._get_test_definition("get_headers", "def-api")
api_def = loader._get_test_definition("get_headers", "api")
self.assertEqual(api_def["request"]["url"], "/headers") self.assertEqual(api_def["request"]["url"], "/headers")
self.assertEqual(len(api_def["setup_hooks"]), 2) self.assertEqual(len(api_def["setup_hooks"]), 2)
self.assertEqual(len(api_def["teardown_hooks"]), 1) self.assertEqual(len(api_def["teardown_hooks"]), 1)
with self.assertRaises(exceptions.ApiNotFound): with self.assertRaises(exceptions.ApiNotFound):
loader._get_test_definition("get_token_XXX", "api") loader._get_test_definition("get_token_XXX", "def-api")
def test_get_test_definition_suite(self): def test_get_test_definition_suite(self):
loader._load_test_dependencies() api_def = loader._get_test_definition("create_and_check", "def-testcase")
api_def = loader._get_test_definition("create_and_check", "suite")
self.assertEqual(api_def["config"]["name"], "create user and check result.") self.assertEqual(api_def["config"]["name"], "create user and check result.")
with self.assertRaises(exceptions.TestcaseNotFound): with self.assertRaises(exceptions.TestcaseNotFound):
loader._get_test_definition("create_and_check_XXX", "suite") loader._get_test_definition("create_and_check_XXX", "def-testcase")
def test_merge_validator(self): def test_merge_validator(self):
def_validators = [ def_validators = [
@@ -376,7 +348,7 @@ class TestSuiteLoader(unittest.TestCase):
self.assertEqual(len(testset_list), 1) self.assertEqual(len(testset_list), 1)
self.assertIn("path", testset_list[0]["config"]) self.assertIn("path", testset_list[0]["config"])
self.assertEqual(testset_list[0]["config"]["path"], path) self.assertEqual(testset_list[0]["config"]["path"], path)
self.assertEqual(len(testset_list[0]["testcases"]), 3) self.assertEqual(len(testset_list[0]["teststeps"]), 3)
testsets_list.extend(testset_list) testsets_list.extend(testset_list)
# relative file path # relative file path
@@ -385,7 +357,7 @@ class TestSuiteLoader(unittest.TestCase):
self.assertEqual(len(testset_list), 1) self.assertEqual(len(testset_list), 1)
self.assertIn("path", testset_list[0]["config"]) self.assertIn("path", testset_list[0]["config"])
self.assertIn(path, testset_list[0]["config"]["path"]) self.assertIn(path, testset_list[0]["config"]["path"])
self.assertEqual(len(testset_list[0]["testcases"]), 3) self.assertEqual(len(testset_list[0]["teststeps"]), 3)
testsets_list.extend(testset_list) testsets_list.extend(testset_list)
# list/set container with file(s) # list/set container with file(s)
@@ -395,20 +367,19 @@ class TestSuiteLoader(unittest.TestCase):
] ]
testset_list = loader.load_testcases(path) testset_list = loader.load_testcases(path)
self.assertEqual(len(testset_list), 2) self.assertEqual(len(testset_list), 2)
self.assertEqual(len(testset_list[0]["testcases"]), 3) self.assertEqual(len(testset_list[0]["teststeps"]), 3)
self.assertEqual(len(testset_list[1]["testcases"]), 3) self.assertEqual(len(testset_list[1]["teststeps"]), 3)
testsets_list.extend(testset_list) testsets_list.extend(testset_list)
self.assertEqual(len(testsets_list), 4) self.assertEqual(len(testsets_list), 4)
for testset in testsets_list: for testset in testsets_list:
for test in testset["testcases"]: for test in testset["teststeps"]:
self.assertIn('name', test) self.assertIn('name', test)
self.assertIn('request', test) self.assertIn('request', test)
self.assertIn('url', test['request']) self.assertIn('url', test['request'])
self.assertIn('method', test['request']) self.assertIn('method', test['request'])
def test_load_testcases_by_path_folder(self): def test_load_testcases_by_path_folder(self):
loader._load_test_dependencies()
# absolute folder path # absolute folder path
path = os.path.join(os.getcwd(), 'tests/data') path = os.path.join(os.getcwd(), 'tests/data')
testset_list_1 = loader.load_testcases(path) testset_list_1 = loader.load_testcases(path)
@@ -447,12 +418,74 @@ class TestSuiteLoader(unittest.TestCase):
loader.load_testcases(path) loader.load_testcases(path)
def test_load_testcases_by_path_layered(self): def test_load_testcases_by_path_layered(self):
loader._load_test_dependencies()
path = os.path.join( path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_layer.yml') os.getcwd(), 'tests/data/demo_testset_layer.yml')
testsets_list = loader.load_testcases(path) testsets_list = loader.load_testcases(path)
self.assertIn("variables", testsets_list[0]["config"]) self.assertIn("variables", testsets_list[0]["config"])
self.assertIn("request", testsets_list[0]["config"]) self.assertIn("request", testsets_list[0]["config"])
self.assertIn("request", testsets_list[0]["testcases"][0]) self.assertIn("request", testsets_list[0]["teststeps"][0])
self.assertIn("url", testsets_list[0]["testcases"][0]["request"]) self.assertIn("url", testsets_list[0]["teststeps"][0]["request"])
self.assertIn("validate", testsets_list[0]["testcases"][0]) self.assertIn("validate", testsets_list[0]["teststeps"][0])
def test_load_folder_content(self):
path = os.path.join(os.getcwd(), "tests", "api")
items_mapping = loader.load_folder_content(path)
file_path = os.path.join(os.getcwd(), "tests", "api", "basic.yml")
self.assertIn(file_path, items_mapping)
self.assertIsInstance(items_mapping[file_path], list)
def test_load_api_folder(self):
path = os.path.join(os.getcwd(), "tests", "api")
api_definition_mapping = loader.load_api_folder(path)
self.assertIn("get_token", api_definition_mapping)
self.assertIn("request", api_definition_mapping["get_token"])
self.assertIn("function_meta", api_definition_mapping["get_token"])
def test_load_testcases_folder(self):
path = os.path.join(os.getcwd(), "tests", "suite")
testcases_definition_mapping = loader.load_test_folder(path)
self.assertIn("setup_and_reset", testcases_definition_mapping)
self.assertIn("create_and_check", testcases_definition_mapping)
self.assertEqual(
testcases_definition_mapping["setup_and_reset"]["config"]["name"],
"setup and reset all."
)
self.assertEqual(
testcases_definition_mapping["setup_and_reset"]["function_meta"]["func_name"],
"setup_and_reset"
)
def test_load_testsuites_folder(self):
path = os.path.join(os.getcwd(), "tests", "testcases")
testsuites_definition_mapping = loader.load_test_folder(path)
testsute_path = os.path.join(os.getcwd(), "tests", "testcases", "smoketest.yml")
self.assertIn(
testsute_path,
testsuites_definition_mapping
)
self.assertEqual(
testsuites_definition_mapping[testsute_path]["config"]["name"],
"smoketest"
)
def test_load_project_tests(self):
project_dir = os.path.join(os.getcwd(), "tests")
project_tests = loader.load_project_tests(project_dir)
self.assertEqual(project_tests["debugtalk"]["variables"]["SECRET_KEY"], "DebugTalk")
self.assertIn("get_token", project_tests["def-api"])
self.assertIn("setup_and_reset", project_tests["def-testcase"])
def test_loader(self):
hrunner = task.HttpRunner(dot_env_path="tests/data/test.env")
self.assertEqual(hrunner.project_mapping["env"]["PROJECT_KEY"], "ABCDEFGH")
self.assertIn("debugtalk", hrunner.project_mapping)
self.assertIn("setup_and_reset", hrunner.project_mapping["def-testcase"])
self.assertEqual(
hrunner.project_mapping["debugtalk"]["variables"]["SECRET_KEY"],
"DebugTalk"
)
self.assertIn("get_sign", hrunner.project_mapping["debugtalk"]["functions"])
self.assertIn("get_token", hrunner.project_mapping["def-api"])
self.assertIn("setup_and_reset", hrunner.project_mapping["def-testcase"])

3
tests/test_response.py
View File

@@ -1,7 +1,8 @@
import requests import requests
from httprunner import built_in, exceptions, loader, response from httprunner import built_in, exceptions, loader, response
from httprunner.compat import basestring, bytes from httprunner.compat import basestring, bytes
from tests.base import HTTPBIN_SERVER, ApiServerUnittest from tests.api_server import HTTPBIN_SERVER
from tests.base import ApiServerUnittest
class TestResponse(ApiServerUnittest): class TestResponse(ApiServerUnittest):

11
tests/test_runner.py
View File

@@ -3,7 +3,8 @@ import time
from httprunner import HttpRunner, exceptions, loader, runner from httprunner import HttpRunner, exceptions, loader, runner
from httprunner.utils import deep_update_dict from httprunner.utils import deep_update_dict
from tests.base import HTTPBIN_SERVER, ApiServerUnittest from tests.api_server import HTTPBIN_SERVER
from tests.base import ApiServerUnittest
class TestRunner(ApiServerUnittest): class TestRunner(ApiServerUnittest):
@@ -169,7 +170,7 @@ class TestRunner(ApiServerUnittest):
"config": { "config": {
'path': 'tests/httpbin/hooks.yml', 'path': 'tests/httpbin/hooks.yml',
}, },
"testcases": [ "teststeps": [
{ {
"name": "test teardown hooks", "name": "test teardown hooks",
"request": { "request": {
@@ -205,7 +206,7 @@ class TestRunner(ApiServerUnittest):
"config": { "config": {
'path': 'tests/httpbin/hooks.yml', 'path': 'tests/httpbin/hooks.yml',
}, },
"testcases": [ "teststeps": [
{ {
"name": "test teardown hooks", "name": "test teardown hooks",
"request": { "request": {
@@ -235,7 +236,7 @@ class TestRunner(ApiServerUnittest):
"config": { "config": {
'path': 'tests/httpbin/hooks.yml', 'path': 'tests/httpbin/hooks.yml',
}, },
"testcases": [ "teststeps": [
{ {
"name": "test teardown hooks", "name": "test teardown hooks",
"request": { "request": {
@@ -383,7 +384,7 @@ class TestRunner(ApiServerUnittest):
testsets = loader.load_testcases(testcase_file_path) testsets = loader.load_testcases(testcase_file_path)
testset = testsets[0] testset = testsets[0]
config_dict_headers = testset["config"]["request"]["headers"] config_dict_headers = testset["config"]["request"]["headers"]
test_dict_headers = testset["testcases"][0]["request"]["headers"] test_dict_headers = testset["teststeps"][0]["request"]["headers"]
headers = deep_update_dict( headers = deep_update_dict(
config_dict_headers, config_dict_headers,
test_dict_headers test_dict_headers

2
tests/test_task.py
View File

@@ -37,7 +37,7 @@ class TestTask(ApiServerUnittest):
'output': ['token'] 'output': ['token']
}, },
'api': {}, 'api': {},
'testcases': [ 'teststeps': [
{ {
'name': '/api/get-token', 'name': '/api/get-token',
'request': { 'request': {