# encoding: utf-8
|
|
import io
|
import logging
|
import os
|
import platform
|
import time
|
import unittest
|
from base64 import b64encode
|
try:
|
from collections.abc import Iterable
|
except ImportError:
|
from collections import Iterable
|
from datetime import datetime
|
|
from jinja2 import Template
|
from markupsafe import escape
|
|
from httprunner.__about__ import __version__
|
from httprunner.compat import basestring, bytes, json, numeric_types
|
|
logger = logging.getLogger(__name__)
|
|
|
def get_platform():
|
return {
|
"httprunner_version": __version__,
|
"python_version": "{} {}".format(
|
platform.python_implementation(), platform.python_version()
|
),
|
"platform": platform.platform(),
|
}
|
|
|
def get_summary(result):
|
"""get summary from test result"""
|
summary = {
|
"success": result.wasSuccessful(),
|
"stat": {
|
"testsRun": result.testsRun,
|
"failures": len(result.failures),
|
"errors": len(result.errors),
|
"skipped": len(result.skipped),
|
"expectedFailures": len(result.expectedFailures),
|
"unexpectedSuccesses": len(result.unexpectedSuccesses),
|
},
|
}
|
summary["stat"]["successes"] = (
|
summary["stat"]["testsRun"]
|
- summary["stat"]["failures"]
|
- summary["stat"]["errors"]
|
- summary["stat"]["skipped"]
|
- summary["stat"]["expectedFailures"]
|
- summary["stat"]["unexpectedSuccesses"]
|
)
|
|
if getattr(result, "records", None):
|
summary["time"] = {
|
"start_at": result.start_at,
|
"setup_hooks_duration": result.setup_hooks_duration,
|
"teardown_hooks_duration": result.teardown_hooks_duration,
|
"duration": result.duration,
|
}
|
summary["records"] = result.records
|
else:
|
summary["records"] = []
|
|
return summary
|
|
|
def aggregate_stat(origin_stat, new_stat):
|
"""aggregate new_stat to origin_stat.
|
|
Args:
|
origin_stat (dict): origin stat dict, will be updated with new_stat dict.
|
new_stat (dict): new stat dict.
|
|
"""
|
for key in new_stat:
|
if key not in origin_stat:
|
origin_stat[key] = new_stat[key]
|
elif key == "start_at":
|
# start datetime
|
origin_stat[key] = min(origin_stat[key], new_stat[key])
|
else:
|
origin_stat[key] += new_stat[key]
|
|
|
def render_html_report(summary, html_report_name=None, html_report_template=None):
|
"""render html report with specified report name and template
|
if html_report_name is not specified, use current datetime
|
if html_report_template is not specified, use default report template
|
"""
|
if not html_report_template:
|
html_report_template = os.path.join(
|
os.path.abspath(os.path.dirname(__file__)),
|
"templates",
|
"report_template.html",
|
)
|
logger.debug("No html report template specified, use default.")
|
else:
|
logger.info("render with html report template: {}".format(html_report_template))
|
|
logger.info("Start to render Html report ...")
|
logger.debug("render data: {}".format(summary))
|
|
report_dir_path = os.path.join(os.getcwd(), "reports")
|
start_at_timestamp = int(summary["time"]["start_at"])
|
summary["time"]["start_datetime"] = datetime.fromtimestamp(
|
start_at_timestamp
|
).strftime("%Y-%m-%d %H:%M:%S")
|
if html_report_name:
|
summary["html_report_name"] = html_report_name
|
report_dir_path = os.path.join(report_dir_path, html_report_name)
|
html_report_name += "-{}.html".format(start_at_timestamp)
|
else:
|
summary["html_report_name"] = ""
|
html_report_name = "{}.html".format(start_at_timestamp)
|
|
if not os.path.isdir(report_dir_path):
|
os.makedirs(report_dir_path)
|
|
for index, suite_summary in enumerate(summary["details"]):
|
if not suite_summary.get("name"):
|
suite_summary["name"] = "test suite {}".format(index)
|
for record in suite_summary.get("records"):
|
meta_data = record["meta_data"]
|
stringify_data(meta_data, "request")
|
stringify_data(meta_data, "response")
|
|
with io.open(html_report_template, "r", encoding="utf-8") as fp_r:
|
template_content = fp_r.read()
|
report_path = os.path.join(report_dir_path, html_report_name)
|
with io.open(report_path, "w", encoding="utf-8") as fp_w:
|
rendered_content = Template(
|
template_content, extensions=["jinja2.ext.loopcontrols"]
|
).render(summary)
|
fp_w.write(rendered_content)
|
|
logger.info("Generated Html report: {}".format(report_path))
|
|
return report_path
|
|
|
def stringify_data(meta_data, request_or_response):
|
"""
|
meta_data = {
|
"request": {},
|
"response": {}
|
}
|
"""
|
headers = meta_data[request_or_response]["headers"]
|
request_or_response_dict = meta_data[request_or_response]
|
|
for key, value in request_or_response_dict.items():
|
|
if isinstance(value, list):
|
value = json.dumps(value, indent=2, ensure_ascii=False)
|
|
elif isinstance(value, bytes):
|
try:
|
encoding = meta_data["response"].get("encoding")
|
if not encoding or encoding == "None":
|
encoding = "utf-8"
|
|
if (
|
request_or_response == "response"
|
and key == "content"
|
and "image" in meta_data["response"]["content_type"]
|
):
|
# display image
|
value = "data:{};base64,{}".format(
|
meta_data["response"]["content_type"],
|
b64encode(value).decode(encoding),
|
)
|
else:
|
value = escape(value.decode(encoding))
|
except UnicodeDecodeError:
|
pass
|
|
elif not isinstance(value, (basestring, numeric_types, Iterable)):
|
# class instance, e.g. MultipartEncoder()
|
value = repr(value)
|
|
meta_data[request_or_response][key] = value
|
|
|
class HtmlTestResult(unittest.TextTestResult):
|
"""A html result class that can generate formatted html results.
|
|
Used by TextTestRunner.
|
"""
|
|
def __init__(self, stream, descriptions, verbosity):
|
super(HtmlTestResult, self).__init__(stream, descriptions, verbosity)
|
self.records = []
|
|
def _record_test(self, test, status, attachment=""):
|
data = {
|
"name": test.shortDescription(),
|
"status": status,
|
"attachment": attachment,
|
"meta_data": {},
|
}
|
if hasattr(test, "meta_data"):
|
data["meta_data"] = test.meta_data
|
|
self.records.append(data)
|
|
def startTestRun(self):
|
self.start_at = time.time()
|
|
def startTest(self, test):
|
"""add start test time"""
|
super(HtmlTestResult, self).startTest(test)
|
logger.info(test.shortDescription())
|
|
def addSuccess(self, test):
|
super(HtmlTestResult, self).addSuccess(test)
|
self._record_test(test, "success")
|
print("")
|
|
def addError(self, test, err):
|
super(HtmlTestResult, self).addError(test, err)
|
self._record_test(test, "error", self._exc_info_to_string(err, test))
|
print("")
|
|
def addFailure(self, test, err):
|
super(HtmlTestResult, self).addFailure(test, err)
|
self._record_test(test, "failure", self._exc_info_to_string(err, test))
|
print("")
|
|
def addSkip(self, test, reason):
|
super(HtmlTestResult, self).addSkip(test, reason)
|
self._record_test(test, "skipped", reason)
|
print("")
|
|
def addExpectedFailure(self, test, err):
|
super(HtmlTestResult, self).addExpectedFailure(test, err)
|
self._record_test(test, "ExpectedFailure", self._exc_info_to_string(err, test))
|
print("")
|
|
def addUnexpectedSuccess(self, test):
|
super(HtmlTestResult, self).addUnexpectedSuccess(test)
|
self._record_test(test, "UnexpectedSuccess")
|
print("")
|
|
@property
|
def duration(self):
|
case_elapsed = 0
|
for record in self.records:
|
elapsed_ms = record["meta_data"]["response"]["elapsed_ms"]
|
if isinstance(elapsed_ms, (int, float)):
|
case_elapsed += record["meta_data"]["response"]["elapsed_ms"]
|
# 毫秒转秒级，保留三位
|
total_duration = (
|
case_elapsed / 1000
|
+ self.setup_hooks_duration
|
+ self.teardown_hooks_duration
|
)
|
return round(total_duration, 3)
|
|
@property
|
def setup_hooks_duration(self):
|
# 整个case的前置函数消耗时间
|
res = 0
|
for record in self.records:
|
setup_hooks_duration = record["meta_data"]["request"].get(
|
"setup_hooks_duration"
|
)
|
if setup_hooks_duration:
|
res += setup_hooks_duration
|
return res
|
|
@property
|
def teardown_hooks_duration(self):
|
# 整个case的后置函数消耗时间
|
res = 0
|
for record in self.records:
|
teardown_hooks_duration = record["meta_data"]["response"].get(
|
"teardown_hooks_duration"
|
)
|
if teardown_hooks_duration:
|
res += teardown_hooks_duration
|
return res
|
