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.

Author: nicoddemus

src/_pytest/reports.py

@@ -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 @@ def head_line(self):
             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

testing/test_reports.py

@@ -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)