From 3ec2a2f29e6b93cb7b4368106b080338d200e273 Mon Sep 17 00:00:00 2001 From: debugtalk Date: Tue, 21 Apr 2020 21:31:09 +0800 Subject: [PATCH] feat: implement HttpRunner main interface v3 --- httprunner/v3/api.py | 256 ++++++++++++++++++++++++++++++++++++++ httprunner/v3/api_test.py | 17 +++ 2 files changed, 273 insertions(+) create mode 100644 httprunner/v3/api.py create mode 100644 httprunner/v3/api_test.py diff --git a/httprunner/v3/api.py b/httprunner/v3/api.py new file mode 100644 index 00000000..e77f8a30 --- /dev/null +++ b/httprunner/v3/api.py @@ -0,0 +1,256 @@ +import os +import sys +import unittest +from typing import List, Tuple + +from loguru import logger + +from httprunner import report, loader, utils, exceptions, __version__ +from httprunner.v3.runner import TestCaseRunner +from httprunner.v3.schema import TestsMapping + + +class HttpRunner(object): + """ Developer Interface: Main Interface + Usage: + + from httprunner.api import HttpRunner + runner = HttpRunner( + failfast=True, + save_tests=True, + log_level="INFO", + log_file="test.log" + ) + summary = runner.run(path_or_tests) + + """ + + def __init__(self, failfast=False, save_tests=False, log_level="WARNING", log_file=None): + """ initialize HttpRunner. + + Args: + failfast (bool): stop the test run on the first error or failure. + save_tests (bool): save loaded/parsed tests to JSON file. + log_level (str): logging level. + log_file (str): log file path. + + """ + self.exception_stage = "initialize HttpRunner()" + kwargs = { + "failfast": failfast, + "resultclass": report.HtmlTestResult + } + + logger.remove() + log_level = log_level.upper() + logger.add(sys.stdout, level=log_level) + if log_file: + logger.add(log_file, level=log_level) + + self.unittest_runner = unittest.TextTestRunner(**kwargs) + self.test_loader = unittest.TestLoader() + self.save_tests = save_tests + self._summary = None + self.test_path = None + + def _prepare_tests(self, tests: TestsMapping) -> List[unittest.TestSuite]: + def _add_test(test_runner: TestCaseRunner): + """ add test to testcase. + """ + def test(self): + try: + test_runner.run() + except exceptions.MyBaseFailure as ex: + self.fail(str(ex)) + finally: + self.meta_datas = test_runner.meta_datas + + test.__doc__ = test_runner.config.name + return test + + project_meta = tests.project_mapping + testcases = tests.testcases + + prepared_testcases: List[unittest.TestSuite] = [] + + for testcase in testcases: + testcase.config.variables.update(project_meta.variables) + testcase.config.functions.update(project_meta.functions) + + test_runner = TestCaseRunner().init(testcase) + + TestSequense = type('TestSequense', (unittest.TestCase,), {}) + test_method = _add_test(test_runner) + setattr(TestSequense, "test_method_name", test_method) + + loaded_testcase = self.test_loader.loadTestsFromTestCase(TestSequense) + setattr(loaded_testcase, "config", testcase.config) + # setattr(loaded_testcase, "teststeps", testcase.teststeps) + # setattr(loaded_testcase, "runner", test_runner) + prepared_testcases.append(loaded_testcase) + + return prepared_testcases + + def _run_suite(self, prepared_testcases: List[unittest.TestSuite]) -> List[Tuple]: + """ run prepared testcases + """ + tests_results: List[Tuple] = [] + + for index, testcase in enumerate(prepared_testcases): + log_handler = None + if self.save_tests: + logs_file_abs_path = utils.prepare_log_file_abs_path( + self.test_path, f"testcase_{index+1}.log" + ) + log_handler = logger.add(logs_file_abs_path, level="DEBUG") + + logger.info(f"Start to run testcase: {testcase.config.name}") + + result = self.unittest_runner.run(testcase) + if result.wasSuccessful(): + tests_results.append((testcase, result)) + else: + tests_results.insert(0, (testcase, result)) + + if self.save_tests and log_handler: + logger.remove(log_handler) + + return tests_results + + def _aggregate(self, tests_results: List[Tuple]): + """ aggregate results + + Args: + tests_results (list): list of (testcase, result) + + """ + summary = { + "success": True, + "stat": { + "testcases": { + "total": len(tests_results), + "success": 0, + "fail": 0 + }, + "teststeps": {} + }, + "time": {}, + "platform": report.get_platform(), + "details": [] + } + + for index, tests_result in enumerate(tests_results): + testcase, result = tests_result + testcase_summary = report.get_summary(result) + + if testcase_summary["success"]: + summary["stat"]["testcases"]["success"] += 1 + else: + summary["stat"]["testcases"]["fail"] += 1 + + summary["success"] &= testcase_summary["success"] + testcase_summary["name"] = testcase.config.name + testcase_summary["in_out"] = { + "in": testcase.config.variables, + "out": testcase.config.export + } + + report.aggregate_stat(summary["stat"]["teststeps"], testcase_summary["stat"]) + report.aggregate_stat(summary["time"], testcase_summary["time"]) + + if self.save_tests: + logs_file_abs_path = utils.prepare_log_file_abs_path( + self.test_path, f"testcase_{index+1}.log" + ) + testcase_summary["log"] = logs_file_abs_path + + summary["details"].append(testcase_summary) + + return summary + + def run_tests(self, tests_mapping): + """ run testcase/testsuite data + """ + tests = TestsMapping.parse_obj(tests_mapping) + self.test_path = tests.project_mapping.test_path + + if self.save_tests: + utils.dump_json_file( + tests_mapping, + utils.prepare_log_file_abs_path(self.test_path, "loaded.json") + ) + + # prepare testcases + self.exception_stage = "prepare testcases" + prepared_testcases = self._prepare_tests(tests) + + # run prepared testcases + self.exception_stage = "run prepared testcases" + results = self._run_suite(prepared_testcases) + + # aggregate results + self.exception_stage = "aggregate results" + self._summary = self._aggregate(results) + + # generate html report + self.exception_stage = "generate html report" + report.stringify_summary(self._summary) + + if self.save_tests: + utils.dump_json_file( + self._summary, + utils.prepare_log_file_abs_path(self.test_path, "summary.json") + ) + # save variables and export data + vars_out = self.get_vars_out() # TODO + utils.dump_json_file( + vars_out, + utils.prepare_log_file_abs_path(self.test_path, "io.json") + ) + + return self._summary + + def run_path(self, path, dot_env_path=None, mapping=None): + """ run testcase/testsuite file or folder. + + Args: + path (str): testcase/testsuite file/foler path. + dot_env_path (str): specified .env file path. + mapping (dict): if mapping is specified, it will override variables in config block. + + Returns: + dict: result summary + + """ + # load tests + self.exception_stage = "load tests" + tests_mapping = loader.load_cases(path, dot_env_path) + + if mapping: + tests_mapping["project_mapping"]["variables"] = mapping + + return self.run_tests(tests_mapping) + + def run(self, path_or_tests, dot_env_path=None, mapping=None): + """ main interface. + + Args: + path_or_tests: + str: testcase/testsuite file/foler path + dict: valid testcase/testsuite data + dot_env_path (str): specified .env file path. + mapping (dict): if mapping is specified, it will override variables in config block. + + Returns: + dict: result summary + + """ + logger.info(f"HttpRunner version: {__version__}") + if loader.is_test_path(path_or_tests): + return self.run_path(path_or_tests, dot_env_path, mapping) + elif loader.is_test_content(path_or_tests): + project_working_directory = path_or_tests.get("project_mapping", {}).get("PWD", os.getcwd()) + loader.init_pwd(project_working_directory) + return self.run_tests(path_or_tests) + else: + raise exceptions.ParamsError(f"Invalid testcase path or testcases: {path_or_tests}") diff --git a/httprunner/v3/api_test.py b/httprunner/v3/api_test.py new file mode 100644 index 00000000..418c6f15 --- /dev/null +++ b/httprunner/v3/api_test.py @@ -0,0 +1,17 @@ +import unittest + +from httprunner.v3.api import HttpRunner + + +class TestHttpRunner(unittest.TestCase): + + def setUp(self): + self.runner = HttpRunner(failfast=True) + + def test_run_testcase_by_path(self): + summary = self.runner.run_path("examples/postman_echo/request_methods/request_with_variables.yml") + self.assertTrue(summary["success"]) + self.assertEqual(summary["details"][0]["name"], "request methods testcase with variables") + self.assertEqual(summary["details"][0]["records"][0]["name"], "request methods testcase with variables") + self.assertEqual(summary["stat"]["testcases"]["total"], 1) + # self.assertEqual(summary["stat"]["teststeps"]["total"], 2)