group file functions to utils.FileUtils

httprunner/__about__.py

@@ -1,7 +1,7 @@
 __title__ = 'HttpRunner'
 __description__ = 'One-stop solution for HTTP(S) testing.'
 __url__ = 'https://github.com/HttpRunner/HttpRunner'
-__version__ = '1.3.11'
+__version__ = '1.3.12'
 __author__ = 'debugtalk'
 __author_email__ = 'mail@debugtalk.com'
 __license__ = 'MIT'

httprunner/testcase.py

@@ -2,7 +2,6 @@
 
 import ast
 import collections
-import csv
 import io
 import itertools
 import json
@@ -10,8 +9,8 @@ import os
 import random
 import re
 
-import yaml
 from httprunner import exception, logger, utils
+from httprunner.utils import FileUtils
 from httprunner.compat import OrderedDict, numeric_types
 
 variable_regexp = r"\$([\w_]+)"
@@ -25,72 +24,6 @@ test_def_overall_dict = {
 testcases_cache_mapping = {}
 
 
-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 exception.JSONDecodeError:
-            err_msg = u"JSONDecodeError: JSON file format error: {}".format(json_file)
-            logger.log_error(err_msg)
-            raise exception.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 exception.FileNotFoundError("{} 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 extract_variables(content):
     """ extract all variable names from content, which is in format $variable
     @param (str) content
@@ -180,7 +113,7 @@ def load_test_dependencies():
 
     # load api definitions
     api_def_folder = os.path.join(os.getcwd(), "tests", "api")
-    api_files = utils.load_folder_files(api_def_folder)
+    api_files = FileUtils.load_folder_files(api_def_folder)
 
     for test_file in api_files:
         testset = load_test_file(test_file)
@@ -188,7 +121,7 @@ def load_test_dependencies():
 
     # load suite definitions
     suite_def_folder = os.path.join(os.getcwd(), "tests", "suite")
-    suite_files = utils.load_folder_files(suite_def_folder)
+    suite_files = FileUtils.load_folder_files(suite_def_folder)
 
     for suite_file in suite_files:
         suite = load_test_file(suite_file)
@@ -230,7 +163,7 @@ def load_testsets_by_path(path):
         return testcases_cache_mapping[path]
 
     if os.path.isdir(path):
-        files_list = utils.load_folder_files(path)
+        files_list = FileUtils.load_folder_files(path)
         testcases_list = load_testsets_by_path(files_list)
 
     elif os.path.isfile(path):
@@ -496,7 +429,7 @@ def load_test_file(file_path):
         "api": {},
         "testcases": []
     }
-    tests_list = load_file(file_path)
+    tests_list = FileUtils.load_file(file_path)
 
     for item in tests_list:
         for key in item:
@@ -639,21 +572,6 @@ def substitute_variables_with_mapping(content, mapping):
 
     return content
 
-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 exception.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 exception.FileFormatError(err_msg)
-
 def gen_cartesian_product(*args):
     """ generate cartesian product for lists
     @param
@@ -812,7 +730,7 @@ class TestcaseParser(object):
             os.path.dirname(self.file_path),
             "{}".format(csv_file_name)
         )
-        csv_content_list = load_file(parameter_file_path)
+        csv_content_list = FileUtils.load_file(parameter_file_path)
 
         if fetch_method.lower() == "random":
             random.shuffle(csv_content_list)

httprunner/utils.py

@@ -1,6 +1,7 @@
 # encoding: utf-8
 
 import copy
+import csv
 import hashlib
 import hmac
 import imp
@@ -14,6 +15,7 @@ import string
 import types
 from datetime import datetime
 
+import yaml
 from httprunner import exception, logger
 from httprunner.compat import OrderedDict, is_py2, is_py3
 from requests.structures import CaseInsensitiveDict
@@ -41,7 +43,91 @@ def remove_prefix(text, prefix):
         return text[len(prefix):]
     return text
 
-def load_folder_files(folder_path, recursive=True):
+
+class FileUtils(object):
+
+    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 exception.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 exception.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)
+            FileUtils._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 exception.JSONDecodeError:
+                err_msg = u"JSONDecodeError: JSON file format error: {}".format(json_file)
+                logger.log_error(err_msg)
+                raise exception.FileFormatError(err_msg)
+
+            FileUtils._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 exception.FileNotFoundError("{} 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 []
+
+    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
@@ -50,7 +136,7 @@ def load_folder_files(folder_path, recursive=True):
         if isinstance(folder_path, (list, set)):
             files = []
             for path in set(folder_path):
-            files.extend(load_folder_files(path, recursive))
+                files.extend(FileUtils.load_folder_files(path, recursive))
 
             return files
 
@@ -77,6 +163,7 @@ def load_folder_files(folder_path, recursive=True):
 
         return file_list
 
+
 def query_json(json_content, query, delimiter='.'):
     """ Do an xpath-like query with json_content.
     @param (json_content) json_content

tests/test_context.py

@@ -2,8 +2,9 @@ import os
 import time
 
 import requests
-from httprunner import exception, response, runner, testcase, utils
+from httprunner import exception, response, runner, testcase
 from httprunner.context import Context
+from httprunner.utils import FileUtils, gen_md5
 from tests.base import ApiServerUnittest
 
 
@@ -12,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 = testcase.load_file(testcase_file_path)
+        self.testcases = FileUtils.load_file(testcase_file_path)
 
     def test_context_init_functions(self):
         self.assertIn("get_timestamp", self.context.testset_functions_config)
@@ -158,7 +159,7 @@ class VariableBindsUnittest(ApiServerUnittest):
             self.assertIn("authorization", context_variables)
             self.assertEqual(len(context_variables["authorization"]), 32)
             authorization = context_variables["authorization"]
-            self.assertEqual(utils.gen_md5(TOKEN, data, random), authorization)
+            self.assertEqual(gen_md5(TOKEN, data, random), authorization)
 
     def test_import_module_items(self):
         testcase1 = {
@@ -192,7 +193,7 @@ class VariableBindsUnittest(ApiServerUnittest):
             self.assertIn("authorization", context_variables)
             self.assertEqual(len(context_variables["authorization"]), 32)
             authorization = context_variables["authorization"]
-            self.assertEqual(utils.gen_md5(TOKEN, data, random), authorization)
+            self.assertEqual(gen_md5(TOKEN, data, random), authorization)
             self.assertIn("SECRET_KEY", context_variables)
             SECRET_KEY = context_variables["SECRET_KEY"]
             self.assertEqual(SECRET_KEY, "DebugTalk")

tests/test_runner.py

@@ -1,7 +1,8 @@
 import os
 import time
 
-from httprunner import HttpRunner, exception, runner, testcase, utils
+from httprunner import HttpRunner, exception, runner, testcase
+from httprunner.utils import FileUtils, deep_update_dict
 from tests.base import ApiServerUnittest
 
 
@@ -25,7 +26,7 @@ class TestRunner(ApiServerUnittest):
 
     def test_run_single_testcase(self):
         for testcase_file_path in self.testcase_file_path_list:
-            testcases = testcase.load_file(testcase_file_path)
+            testcases = FileUtils.load_file(testcase_file_path)
 
             config_dict = {
                 "path": testcase_file_path
@@ -157,7 +158,7 @@ class TestRunner(ApiServerUnittest):
         testset = testsets[0]
         config_dict_headers = testset["config"]["request"]["headers"]
         test_dict_headers = testset["testcases"][0]["request"]["headers"]
-        headers = utils.deep_update_dict(
+        headers = deep_update_dict(
             config_dict_headers,
             test_dict_headers
         )
@@ -166,7 +167,7 @@ class TestRunner(ApiServerUnittest):
     def test_bugfix_type_match(self):
         testcase_file_path = os.path.join(
             os.getcwd(), 'tests/data/test_bugfix.yml')
-        testcases = testcase.load_file(testcase_file_path)
+        testcases = FileUtils.load_file(testcase_file_path)
         config_dict = {
             "path": testcase_file_path
         }

tests/test_testcase.py

@@ -9,59 +9,6 @@ from httprunner.exception import (ApiNotFound, FileFormatError,
 
 class TestcaseParserUnittest(unittest.TestCase):
 
-    def test_load_testcases_bad_filepath(self):
-        testcase_file_path = os.path.join(os.getcwd(), 'tests/data/demo')
-        with self.assertRaises(FileNotFoundError):
-            testcase.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 = testcase.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 = testcase.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 = testcase.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 = testcase.load_file(csv_file_path)
-        self.assertEqual(
-            csv_content,
-            [
-                {'username': 'test1', 'password': '111111'},
-                {'username': 'test2', 'password': '222222'},
-                {'username': 'test3', 'password': '333333'}
-            ]
-        )
-
     def test_cartesian_product_one(self):
         parameters_content_list = [
             [
@@ -175,55 +122,6 @@ class TestcaseParserUnittest(unittest.TestCase):
             3 * 2 * 3
         )
 
-    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(FileFormatError):
-            testcase._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(FileFormatError):
-            testcase._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(FileFormatError):
-            testcase._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(FileFormatError):
-            testcase._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(FileFormatError):
-            testcase._load_json_file(json_tmp_file)
-
-        os.remove(json_tmp_file)
-
     def test_extract_variables(self):
         self.assertEqual(
             testcase.extract_variables("$var"),

tests/test_utils.py

@@ -1,11 +1,146 @@
 import os
 import shutil
+import unittest
 
 from httprunner import exception, 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(exception.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(exception.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(exception.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(exception.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(exception.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(FileNotFoundError):
+            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_1 = FileUtils.load_folder_files(folder)
+        api_file = os.path.join(os.getcwd(), 'tests', 'api', 'demo.yml')
+        self.assertEqual(files_1[0], api_file)
+
+        files_2 = FileUtils.load_folder_files(folder)
+        api_file = os.path.join(os.getcwd(), 'tests', 'api', 'demo.yml')
+        self.assertEqual(files_2[0], api_file)
+        self.assertEqual(len(files_1), len(files_2))
+
+        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):
@@ -16,33 +151,6 @@ class TestUtils(ApiServerUnittest):
             "/post/123"
         )
 
-    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 = utils.load_folder_files(folder, recursive=False)
-        self.assertNotIn(file2, files)
-
-        files = utils.load_folder_files(folder)
-        self.assertIn(file2, files)
-        self.assertNotIn(file1, files)
-
-        files_1 = utils.load_folder_files(folder)
-        api_file = os.path.join(os.getcwd(), 'tests', 'api', 'demo.yml')
-        self.assertEqual(files_1[0], api_file)
-
-        files_2 = utils.load_folder_files(folder)
-        api_file = os.path.join(os.getcwd(), 'tests', 'api', 'demo.yml')
-        self.assertEqual(files_2[0], api_file)
-        self.assertEqual(len(files_1), len(files_2))
-
-        files = utils.load_folder_files("not_existed_foulder", recursive=False)
-        self.assertEqual([], files)
-
-        files = utils.load_folder_files(file2, recursive=False)
-        self.assertEqual([], files)
-
     def test_query_json(self):
         json_content = {
             "ids": [1, 2, 3, 4],