GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-05-15 04:19:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

relocate file loader

Browse Source
This commit is contained in:
debugtalk
2018-07-29 22:40:00 +08:00
parent 4dadef4ad9
commit 5738c368c8
6 changed files with 150 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
httprunner/loader.py Normal file
View File

@@ -0,0 +1,136 @@
import csv
import io
import json
import os
import yaml
from httprunner import exceptions, logger
###############################################################################
## file loader
###############################################################################
def _check_format(file_path, content):
""" check testcase format if valid
"""
# TODO: replace with JSON schema validation
if not content:
# testcase file content is empty
err_msg = u"Testcase file content is empty: {}".format(file_path)
logger.log_error(err_msg)
raise exceptions.FileFormatError(err_msg)
elif not isinstance(content, (list, dict)):
# testcase file content does not match testcase format
err_msg = u"Testcase file content format invalid: {}".format(file_path)
logger.log_error(err_msg)
raise exceptions.FileFormatError(err_msg)
def load_yaml_file(yaml_file):
""" load yaml file and check file content format
"""
with io.open(yaml_file, 'r', encoding='utf-8') as stream:
yaml_content = yaml.load(stream)
_check_format(yaml_file, yaml_content)
return yaml_content
def load_json_file(json_file):
""" load json file and check file content format
"""
with io.open(json_file, encoding='utf-8') as data_file:
try:
json_content = json.load(data_file)
except exceptions.JSONDecodeError:
err_msg = u"JSONDecodeError: JSON file format error: {}".format(json_file)
logger.log_error(err_msg)
raise exceptions.FileFormatError(err_msg)
_check_format(json_file, json_content)
return json_content
def load_csv_file(csv_file):
""" load csv file and check file content format
@param
csv_file: csv file path
e.g. csv file content:
username,password
test1,111111
test2,222222
test3,333333
@return
list of parameter, each parameter is in dict format
e.g.
[
{'username': 'test1', 'password': '111111'},
{'username': 'test2', 'password': '222222'},
{'username': 'test3', 'password': '333333'}
]
"""
csv_content_list = []
with io.open(csv_file, encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
csv_content_list.append(row)
return csv_content_list
def load_file(file_path):
if not os.path.isfile(file_path):
raise exceptions.FileNotFound("{} does not exist.".format(file_path))
file_suffix = os.path.splitext(file_path)[1].lower()
if file_suffix == '.json':
return load_json_file(file_path)
elif file_suffix in ['.yaml', '.yml']:
return load_yaml_file(file_path)
elif file_suffix == ".csv":
return load_csv_file(file_path)
else:
# '' or other suffix
err_msg = u"Unsupported file format: {}".format(file_path)
logger.log_warning(err_msg)
return []
def load_folder_files(folder_path, recursive=True):
""" load folder path, return all files in list format.
@param
folder_path: specified folder path to load
recursive: if True, will load files recursively
"""
if isinstance(folder_path, (list, set)):
files = []
for path in set(folder_path):
files.extend(load_folder_files(path, recursive))
return files
if not os.path.exists(folder_path):
return []
file_list = []
for dirpath, dirnames, filenames in os.walk(folder_path):
filenames_list = []
for filename in filenames:
if not filename.endswith(('.yml', '.yaml', '.json')):
continue
filenames_list.append(filename)
for filename in filenames_list:
file_path = os.path.join(dirpath, filename)
file_list.append(file_path)
if not recursive:
break
return file_list

15
httprunner/testcase.py
View File

@@ -9,10 +9,9 @@ import os
import random
import re
from httprunner import exceptions, logger, utils
from httprunner import exceptions, loader, logger, utils
from httprunner.compat import (OrderedDict, basestring, builtin_str,
numeric_types, str)
from httprunner.utils import FileUtils
variable_regexp = r"\$([\w_]+)"
function_regexp = r"\$\{([\w_]+\([\$\w\.\-_ =,]*\))\}"
@@ -119,12 +118,12 @@ class TestcaseLoader(object):
# TODO: cache api and suite loading
# load api definitions
api_def_folder = os.path.join(os.getcwd(), "tests", "api")
for test_file in FileUtils.load_folder_files(api_def_folder):
for test_file in loader.load_folder_files(api_def_folder):
TestcaseLoader.load_api_file(test_file)
# load suite definitions
suite_def_folder = os.path.join(os.getcwd(), "tests", "suite")
for suite_file in FileUtils.load_folder_files(suite_def_folder):
for suite_file in loader.load_folder_files(suite_def_folder):
suite = TestcaseLoader.load_test_file(suite_file)
if "def" not in suite["config"]:
raise exceptions.ParamsError("def missed in suite file: {}!".format(suite_file))
@@ -155,7 +154,7 @@ class TestcaseLoader(object):
}
]
"""
api_items = FileUtils.load_file(file_path)
api_items = loader.load_file(file_path)
if not isinstance(api_items, list):
raise exceptions.FileFormatError("API format error: {}".format(file_path))
@@ -217,7 +216,7 @@ class TestcaseLoader(object):
},
"testcases": [] # TODO: rename to tests
}
for item in FileUtils.load_file(file_path):
for item in loader.load_file(file_path):
if not isinstance(item, dict) or len(item) != 1:
raise exceptions.FileFormatError("Testcase format error: {}".format(file_path))
@@ -371,7 +370,7 @@ class TestcaseLoader(object):
return TestcaseLoader.testcases_cache_mapping[path]
if os.path.isdir(path):
files_list = FileUtils.load_folder_files(path)
files_list = loader.load_folder_files(path)
testcases_list = TestcaseLoader.load_testsets_by_path(files_list)
elif os.path.isfile(path):
@@ -792,7 +791,7 @@ class TestcaseParser(object):
os.path.dirname(self.file_path),
"{}".format(csv_file_name)
)
csv_content_list = FileUtils.load_file(parameter_file_path)
csv_content_list = loader.load_file(parameter_file_path)
if fetch_method.lower() == "random":
random.shuffle(csv_content_list)

129
httprunner/utils.py
View File

@@ -1,7 +1,6 @@
# encoding: utf-8
import copy
import csv
import hashlib
import hmac
import imp
@@ -10,12 +9,10 @@ import io
import json
import os.path
import random
import re
import string
import types
from datetime import datetime
import yaml
from httprunner import exceptions, logger
from httprunner.compat import OrderedDict, basestring, is_py2, is_py3, str
from requests.structures import CaseInsensitiveDict
@@ -44,132 +41,6 @@ def remove_prefix(text, prefix):
return text
class FileUtils(object):
@staticmethod
def _check_format(file_path, content):
""" check testcase format if valid
"""
if not content:
# testcase file content is empty
err_msg = u"Testcase file content is empty: {}".format(file_path)
logger.log_error(err_msg)
raise exceptions.FileFormatError(err_msg)
elif not isinstance(content, (list, dict)):
# testcase file content does not match testcase format
err_msg = u"Testcase file content format invalid: {}".format(file_path)
logger.log_error(err_msg)
raise exceptions.FileFormatError(err_msg)
@staticmethod
def _load_yaml_file(yaml_file):
""" load yaml file and check file content format
"""
with io.open(yaml_file, 'r', encoding='utf-8') as stream:
yaml_content = yaml.load(stream)
FileUtils._check_format(yaml_file, yaml_content)
return yaml_content
@staticmethod
def _load_json_file(json_file):
""" load json file and check file content format
"""
with io.open(json_file, encoding='utf-8') as data_file:
try:
json_content = json.load(data_file)
except exceptions.JSONDecodeError:
err_msg = u"JSONDecodeError: JSON file format error: {}".format(json_file)
logger.log_error(err_msg)
raise exceptions.FileFormatError(err_msg)
FileUtils._check_format(json_file, json_content)
return json_content
@staticmethod
def _load_csv_file(csv_file):
""" load csv file and check file content format
@param
csv_file: csv file path
e.g. csv file content:
username,password
test1,111111
test2,222222
test3,333333
@return
list of parameter, each parameter is in dict format
e.g.
[
{'username': 'test1', 'password': '111111'},
{'username': 'test2', 'password': '222222'},
{'username': 'test3', 'password': '333333'}
]
"""
csv_content_list = []
with io.open(csv_file, encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
csv_content_list.append(row)
return csv_content_list
@staticmethod
def load_file(file_path):
if not os.path.isfile(file_path):
raise exceptions.FileNotFound("{} does not exist.".format(file_path))
file_suffix = os.path.splitext(file_path)[1].lower()
if file_suffix == '.json':
return FileUtils._load_json_file(file_path)
elif file_suffix in ['.yaml', '.yml']:
return FileUtils._load_yaml_file(file_path)
elif file_suffix == ".csv":
return FileUtils._load_csv_file(file_path)
else:
# '' or other suffix
err_msg = u"Unsupported file format: {}".format(file_path)
logger.log_warning(err_msg)
return []
@staticmethod
def load_folder_files(folder_path, recursive=True):
""" load folder path, return all files in list format.
@param
folder_path: specified folder path to load
recursive: if True, will load files recursively
"""
if isinstance(folder_path, (list, set)):
files = []
for path in set(folder_path):
files.extend(FileUtils.load_folder_files(path, recursive))
return files
if not os.path.exists(folder_path):
return []
file_list = []
for dirpath, dirnames, filenames in os.walk(folder_path):
filenames_list = []
for filename in filenames:
if not filename.endswith(('.yml', '.yaml', '.json')):
continue
filenames_list.append(filename)
for filename in filenames_list:
file_path = os.path.join(dirpath, filename)
file_list.append(file_path)
if not recursive:
break
return file_list
def query_json(json_content, query, delimiter='.'):
""" Do an xpath-like query with json_content.
@param (dict/list/string) json_content

6
tests/test_context.py
View File

@@ -2,9 +2,9 @@ import os
import time
import requests
from httprunner import exceptions, response, runner, testcase
from httprunner import exceptions, loader, response, runner, testcase
from httprunner.context import Context
from httprunner.utils import FileUtils, gen_md5
from httprunner.utils import gen_md5
from tests.base import ApiServerUnittest
@@ -13,7 +13,7 @@ class VariableBindsUnittest(ApiServerUnittest):
def setUp(self):
self.context = Context()
testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo_binds.yml')
self.testcases = FileUtils.load_file(testcase_file_path)
self.testcases = loader.load_file(testcase_file_path)
def test_context_init_functions(self):
self.assertIn("get_timestamp", self.context.testset_functions_config)

8
tests/test_runner.py
View File

@@ -1,9 +1,9 @@
import os
import time
from httprunner import HttpRunner, exceptions, runner
from httprunner import HttpRunner, exceptions, loader, runner
from httprunner.testcase import TestcaseLoader
from httprunner.utils import FileUtils, deep_update_dict
from httprunner.utils import deep_update_dict
from tests.base import ApiServerUnittest
@@ -27,7 +27,7 @@ class TestRunner(ApiServerUnittest):
def test_run_single_testcase(self):
for testcase_file_path in self.testcase_file_path_list:
testcases = FileUtils.load_file(testcase_file_path)
testcases = loader.load_file(testcase_file_path)
config_dict = {
"path": testcase_file_path
@@ -393,7 +393,7 @@ class TestRunner(ApiServerUnittest):
def test_bugfix_type_match(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/test_bugfix.yml')
testcases = FileUtils.load_file(testcase_file_path)
testcases = loader.load_file(testcase_file_path)
config_dict = {
"path": testcase_file_path
}

129
tests/test_utils.py
View File

@@ -4,138 +4,9 @@ import unittest
from httprunner import exceptions, utils
from httprunner.compat import OrderedDict
from httprunner.utils import FileUtils
from tests.base import ApiServerUnittest
class TestFileUtils(unittest.TestCase):
def test_load_yaml_file_file_format_error(self):
yaml_tmp_file = "tests/data/tmp.yml"
# create empty yaml file
with open(yaml_tmp_file, 'w') as f:
f.write("")
with self.assertRaises(exceptions.FileFormatError):
FileUtils._load_yaml_file(yaml_tmp_file)
os.remove(yaml_tmp_file)
# create invalid format yaml file
with open(yaml_tmp_file, 'w') as f:
f.write("abc")
with self.assertRaises(exceptions.FileFormatError):
FileUtils._load_yaml_file(yaml_tmp_file)
os.remove(yaml_tmp_file)
def test_load_json_file_file_format_error(self):
json_tmp_file = "tests/data/tmp.json"
# create empty file
with open(json_tmp_file, 'w') as f:
f.write("")
with self.assertRaises(exceptions.FileFormatError):
FileUtils._load_json_file(json_tmp_file)
os.remove(json_tmp_file)
# create empty json file
with open(json_tmp_file, 'w') as f:
f.write("{}")
with self.assertRaises(exceptions.FileFormatError):
FileUtils._load_json_file(json_tmp_file)
os.remove(json_tmp_file)
# create invalid format json file
with open(json_tmp_file, 'w') as f:
f.write("abc")
with self.assertRaises(exceptions.FileFormatError):
FileUtils._load_json_file(json_tmp_file)
os.remove(json_tmp_file)
def test_load_testcases_bad_filepath(self):
testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo')
with self.assertRaises(exceptions.FileNotFound):
FileUtils.load_file(testcase_file_path)
def test_load_json_testcases(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.json')
testcases = FileUtils.load_file(testcase_file_path)
self.assertEqual(len(testcases), 3)
test = testcases[0]["test"]
self.assertIn('name', test)
self.assertIn('request', test)
self.assertIn('url', test['request'])
self.assertIn('method', test['request'])
def test_load_yaml_testcases(self):
testcase_file_path = os.path.join(
os.getcwd(), 'tests/data/demo_testset_hardcode.yml')
testcases = FileUtils.load_file(testcase_file_path)
self.assertEqual(len(testcases), 3)
test = testcases[0]["test"]
self.assertIn('name', test)
self.assertIn('request', test)
self.assertIn('url', test['request'])
self.assertIn('method', test['request'])
def test_load_csv_file_one_parameter(self):
csv_file_path = os.path.join(
os.getcwd(), 'tests/data/user_agent.csv')
csv_content = FileUtils.load_file(csv_file_path)
self.assertEqual(
csv_content,
[
{'user_agent': 'iOS/10.1'},
{'user_agent': 'iOS/10.2'},
{'user_agent': 'iOS/10.3'}
]
)
def test_load_csv_file_multiple_parameters(self):
csv_file_path = os.path.join(
os.getcwd(), 'tests/data/account.csv')
csv_content = FileUtils.load_file(csv_file_path)
self.assertEqual(
csv_content,
[
{'username': 'test1', 'password': '111111'},
{'username': 'test2', 'password': '222222'},
{'username': 'test3', 'password': '333333'}
]
)
def test_load_folder_files(self):
folder = os.path.join(os.getcwd(), 'tests')
file1 = os.path.join(os.getcwd(), 'tests', 'test_utils.py')
file2 = os.path.join(os.getcwd(), 'tests', 'data', 'demo_binds.yml')
files = FileUtils.load_folder_files(folder, recursive=False)
self.assertNotIn(file2, files)
files = FileUtils.load_folder_files(folder)
self.assertIn(file2, files)
self.assertNotIn(file1, files)
files = FileUtils.load_folder_files(folder)
api_file = os.path.join(os.getcwd(), 'tests', 'api', 'basic.yml')
self.assertIn(api_file, files)
files = FileUtils.load_folder_files("not_existed_foulder", recursive=False)
self.assertEqual([], files)
files = FileUtils.load_folder_files(file2, recursive=False)
self.assertEqual([], files)
class TestUtils(ApiServerUnittest):
def test_remove_prefix(self):