Add experimental _to_json and _from_json to TestReport and CollectReport

This methods were moved from xdist (ca03269).

Our intention is to keep this code closer to the core, given that it
might break easily due to refactorings.

Having it in the core might also allow to improve the code by moving
some responsibility to the "code" objects (ReprEntry, etc) which
are often found in the reports.

Finally pytest-xdist and pytest-subtests can use those functions
instead of coding it themselves.
This commit is contained in:
Bruno Oliveira 2019-03-20 20:24:51 -03:00
parent 2df9d05981
commit 0c63f99016
2 changed files with 315 additions and 0 deletions

View File

@ -1,6 +1,14 @@
import py
import six
from _pytest._code.code import ExceptionInfo
from _pytest._code.code import ReprEntry
from _pytest._code.code import ReprEntryNative
from _pytest._code.code import ReprExceptionInfo
from _pytest._code.code import ReprFileLocation
from _pytest._code.code import ReprFuncArgs
from _pytest._code.code import ReprLocals
from _pytest._code.code import ReprTraceback
from _pytest._code.code import TerminalRepr
from _pytest.outcomes import skip
@ -137,6 +145,130 @@ class BaseReport(object):
fspath, lineno, domain = self.location
return domain
def _to_json(self):
"""
This was originally the serialize_report() function from xdist (ca03269).
Returns the contents of this report as a dict of builtin entries, suitable for
serialization.
Experimental method.
"""
def disassembled_report(rep):
reprtraceback = rep.longrepr.reprtraceback.__dict__.copy()
reprcrash = rep.longrepr.reprcrash.__dict__.copy()
new_entries = []
for entry in reprtraceback["reprentries"]:
entry_data = {
"type": type(entry).__name__,
"data": entry.__dict__.copy(),
}
for key, value in entry_data["data"].items():
if hasattr(value, "__dict__"):
entry_data["data"][key] = value.__dict__.copy()
new_entries.append(entry_data)
reprtraceback["reprentries"] = new_entries
return {
"reprcrash": reprcrash,
"reprtraceback": reprtraceback,
"sections": rep.longrepr.sections,
}
d = self.__dict__.copy()
if hasattr(self.longrepr, "toterminal"):
if hasattr(self.longrepr, "reprtraceback") and hasattr(
self.longrepr, "reprcrash"
):
d["longrepr"] = disassembled_report(self)
else:
d["longrepr"] = six.text_type(self.longrepr)
else:
d["longrepr"] = self.longrepr
for name in d:
if isinstance(d[name], py.path.local):
d[name] = str(d[name])
elif name == "result":
d[name] = None # for now
return d
@classmethod
def _from_json(cls, reportdict):
"""
This was originally the serialize_report() function from xdist (ca03269).
Factory method that returns either a TestReport or CollectReport, depending on the calling
class. It's the callers responsibility to know which class to pass here.
Experimental method.
"""
if reportdict["longrepr"]:
if (
"reprcrash" in reportdict["longrepr"]
and "reprtraceback" in reportdict["longrepr"]
):
reprtraceback = reportdict["longrepr"]["reprtraceback"]
reprcrash = reportdict["longrepr"]["reprcrash"]
unserialized_entries = []
reprentry = None
for entry_data in reprtraceback["reprentries"]:
data = entry_data["data"]
entry_type = entry_data["type"]
if entry_type == "ReprEntry":
reprfuncargs = None
reprfileloc = None
reprlocals = None
if data["reprfuncargs"]:
reprfuncargs = ReprFuncArgs(**data["reprfuncargs"])
if data["reprfileloc"]:
reprfileloc = ReprFileLocation(**data["reprfileloc"])
if data["reprlocals"]:
reprlocals = ReprLocals(data["reprlocals"]["lines"])
reprentry = ReprEntry(
lines=data["lines"],
reprfuncargs=reprfuncargs,
reprlocals=reprlocals,
filelocrepr=reprfileloc,
style=data["style"],
)
elif entry_type == "ReprEntryNative":
reprentry = ReprEntryNative(data["lines"])
else:
_report_unserialization_failure(entry_type, cls, reportdict)
unserialized_entries.append(reprentry)
reprtraceback["reprentries"] = unserialized_entries
exception_info = ReprExceptionInfo(
reprtraceback=ReprTraceback(**reprtraceback),
reprcrash=ReprFileLocation(**reprcrash),
)
for section in reportdict["longrepr"]["sections"]:
exception_info.addsection(*section)
reportdict["longrepr"] = exception_info
return cls(**reportdict)
def _report_unserialization_failure(type_name, report_class, reportdict):
from pprint import pprint
url = "https://github.com/pytest-dev/pytest/issues"
stream = py.io.TextIO()
pprint("-" * 100, stream=stream)
pprint("INTERNALERROR: Unknown entry type returned: %s" % type_name, stream=stream)
pprint("report_name: %s" % report_class, stream=stream)
pprint(reportdict, stream=stream)
pprint("Please report this bug at %s" % url, stream=stream)
pprint("-" * 100, stream=stream)
assert 0, stream.getvalue()
class TestReport(BaseReport):
""" Basic test report object (also used for setup and teardown calls if

183
testing/test_reports.py Normal file
View File

@ -0,0 +1,183 @@
from _pytest.reports import CollectReport
from _pytest.reports import TestReport
class TestReportSerialization(object):
"""
All the tests in this class came originally from test_remote.py in xdist (ca03269).
"""
def test_xdist_longrepr_to_str_issue_241(self, testdir):
testdir.makepyfile(
"""
import os
def test_a(): assert False
def test_b(): pass
"""
)
reprec = testdir.inline_run()
reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 6
test_a_call = reports[1]
assert test_a_call.when == "call"
assert test_a_call.outcome == "failed"
assert test_a_call._to_json()["longrepr"]["reprtraceback"]["style"] == "long"
test_b_call = reports[4]
assert test_b_call.when == "call"
assert test_b_call.outcome == "passed"
assert test_b_call._to_json()["longrepr"] is None
def test_xdist_report_longrepr_reprcrash_130(self, testdir):
reprec = testdir.inline_runsource(
"""
import py
def test_fail(): assert False, 'Expected Message'
"""
)
reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 3
rep = reports[1]
added_section = ("Failure Metadata", str("metadata metadata"), "*")
rep.longrepr.sections.append(added_section)
d = rep._to_json()
a = TestReport._from_json(d)
# Check assembled == rep
assert a.__dict__.keys() == rep.__dict__.keys()
for key in rep.__dict__.keys():
if key != "longrepr":
assert getattr(a, key) == getattr(rep, key)
assert rep.longrepr.reprcrash.lineno == a.longrepr.reprcrash.lineno
assert rep.longrepr.reprcrash.message == a.longrepr.reprcrash.message
assert rep.longrepr.reprcrash.path == a.longrepr.reprcrash.path
assert rep.longrepr.reprtraceback.entrysep == a.longrepr.reprtraceback.entrysep
assert (
rep.longrepr.reprtraceback.extraline == a.longrepr.reprtraceback.extraline
)
assert rep.longrepr.reprtraceback.style == a.longrepr.reprtraceback.style
assert rep.longrepr.sections == a.longrepr.sections
# Missing section attribute PR171
assert added_section in a.longrepr.sections
def test_reprentries_serialization_170(self, testdir):
from _pytest._code.code import ReprEntry
reprec = testdir.inline_runsource(
"""
def test_repr_entry():
x = 0
assert x
""",
"--showlocals",
)
reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 3
rep = reports[1]
d = rep._to_json()
a = TestReport._from_json(d)
rep_entries = rep.longrepr.reprtraceback.reprentries
a_entries = a.longrepr.reprtraceback.reprentries
for i in range(len(a_entries)):
assert isinstance(rep_entries[i], ReprEntry)
assert rep_entries[i].lines == a_entries[i].lines
assert rep_entries[i].reprfileloc.lineno == a_entries[i].reprfileloc.lineno
assert (
rep_entries[i].reprfileloc.message == a_entries[i].reprfileloc.message
)
assert rep_entries[i].reprfileloc.path == a_entries[i].reprfileloc.path
assert rep_entries[i].reprfuncargs.args == a_entries[i].reprfuncargs.args
assert rep_entries[i].reprlocals.lines == a_entries[i].reprlocals.lines
assert rep_entries[i].style == a_entries[i].style
def test_reprentries_serialization_196(self, testdir):
from _pytest._code.code import ReprEntryNative
reprec = testdir.inline_runsource(
"""
def test_repr_entry_native():
x = 0
assert x
""",
"--tb=native",
)
reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 3
rep = reports[1]
d = rep._to_json()
a = TestReport._from_json(d)
rep_entries = rep.longrepr.reprtraceback.reprentries
a_entries = a.longrepr.reprtraceback.reprentries
for i in range(len(a_entries)):
assert isinstance(rep_entries[i], ReprEntryNative)
assert rep_entries[i].lines == a_entries[i].lines
def test_itemreport_outcomes(self, testdir):
reprec = testdir.inline_runsource(
"""
import py
def test_pass(): pass
def test_fail(): 0/0
@py.test.mark.skipif("True")
def test_skip(): pass
def test_skip_imperative():
py.test.skip("hello")
@py.test.mark.xfail("True")
def test_xfail(): 0/0
def test_xfail_imperative():
py.test.xfail("hello")
"""
)
reports = reprec.getreports("pytest_runtest_logreport")
assert len(reports) == 17 # with setup/teardown "passed" reports
for rep in reports:
d = rep._to_json()
newrep = TestReport._from_json(d)
assert newrep.passed == rep.passed
assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped
if newrep.skipped and not hasattr(newrep, "wasxfail"):
assert len(newrep.longrepr) == 3
assert newrep.outcome == rep.outcome
assert newrep.when == rep.when
assert newrep.keywords == rep.keywords
if rep.failed:
assert newrep.longreprtext == rep.longreprtext
def test_collectreport_passed(self, testdir):
reprec = testdir.inline_runsource("def test_func(): pass")
reports = reprec.getreports("pytest_collectreport")
for rep in reports:
d = rep._to_json()
newrep = CollectReport._from_json(d)
assert newrep.passed == rep.passed
assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped
def test_collectreport_fail(self, testdir):
reprec = testdir.inline_runsource("qwe abc")
reports = reprec.getreports("pytest_collectreport")
assert reports
for rep in reports:
d = rep._to_json()
newrep = CollectReport._from_json(d)
assert newrep.passed == rep.passed
assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped
if rep.failed:
assert newrep.longrepr == str(rep.longrepr)
def test_extended_report_deserialization(self, testdir):
reprec = testdir.inline_runsource("qwe abc")
reports = reprec.getreports("pytest_collectreport")
assert reports
for rep in reports:
rep.extra = True
d = rep._to_json()
newrep = CollectReport._from_json(d)
assert newrep.extra
assert newrep.passed == rep.passed
assert newrep.failed == rep.failed
assert newrep.skipped == rep.skipped
if rep.failed:
assert newrep.longrepr == str(rep.longrepr)