Use serialization hooks from pytest 4.4
This commit is contained in:
Bruno Oliveira
@@ -3,8 +3,7 @@ import pprint
|
||||
import pytest
|
||||
import sys
|
||||
|
||||
from xdist.workermanage import WorkerController, unserialize_report
|
||||
from xdist.remote import serialize_report
|
||||
from xdist.workermanage import WorkerController
|
||||
import execnet
|
||||
import marshal
|
||||
|
||||
@@ -81,187 +80,17 @@ def test_remoteinitconfig(testdir):
|
||||
assert config2.pluginmanager.getplugin("terminal") in (-1, None)
|
||||
|
||||
|
||||
class TestReportSerialization:
|
||||
def test_xdist_longrepr_to_str_issue_241(self, testdir):
|
||||
testdir.makepyfile(
|
||||
"""
|
||||
import os
|
||||
def test_a(): assert False
|
||||
def test_b(): pass
|
||||
"""
|
||||
)
|
||||
testdir.makeconftest(
|
||||
"""
|
||||
def pytest_runtest_logreport(report):
|
||||
print(report.longrepr)
|
||||
"""
|
||||
)
|
||||
res = testdir.runpytest("-n1", "-s")
|
||||
res.stdout.fnmatch_lines(["*1 failed, 1 passed *"])
|
||||
|
||||
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 = serialize_report(rep)
|
||||
check_marshallable(d)
|
||||
a = unserialize_report("testreport", 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 = serialize_report(rep)
|
||||
a = unserialize_report("testreport", 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 = serialize_report(rep)
|
||||
a = unserialize_report("testreport", 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 = serialize_report(rep)
|
||||
check_marshallable(d)
|
||||
newrep = unserialize_report("testreport", 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 = serialize_report(rep)
|
||||
check_marshallable(d)
|
||||
newrep = unserialize_report("collectreport", 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 = serialize_report(rep)
|
||||
check_marshallable(d)
|
||||
newrep = unserialize_report("collectreport", 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 = serialize_report(rep)
|
||||
check_marshallable(d)
|
||||
newrep = unserialize_report("collectreport", 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)
|
||||
|
||||
|
||||
class TestWorkerInteractor:
|
||||
def test_basic_collect_and_runtests(self, worker):
|
||||
@pytest.fixture
|
||||
def unserialize_report(self, pytestconfig):
|
||||
def unserialize(data):
|
||||
return pytestconfig.hook.pytest_report_from_serializable(
|
||||
config=pytestconfig, data=data
|
||||
)
|
||||
|
||||
return unserialize
|
||||
|
||||
def test_basic_collect_and_runtests(self, worker, unserialize_report):
|
||||
worker.testdir.makepyfile(
|
||||
"""
|
||||
def test_func():
|
||||
@@ -286,14 +115,14 @@ class TestWorkerInteractor:
|
||||
ev = worker.popevent("testreport") # setup
|
||||
ev = worker.popevent("testreport")
|
||||
assert ev.name == "testreport"
|
||||
rep = unserialize_report(ev.name, ev.kwargs["data"])
|
||||
rep = unserialize_report(ev.kwargs["data"])
|
||||
assert rep.nodeid.endswith("::test_func")
|
||||
assert rep.passed
|
||||
assert rep.when == "call"
|
||||
ev = worker.popevent("workerfinished")
|
||||
assert "workeroutput" in ev.kwargs
|
||||
|
||||
def test_remote_collect_skip(self, worker):
|
||||
def test_remote_collect_skip(self, worker, unserialize_report):
|
||||
worker.testdir.makepyfile(
|
||||
"""
|
||||
import pytest
|
||||
@@ -305,25 +134,25 @@ class TestWorkerInteractor:
|
||||
assert not ev.kwargs
|
||||
ev = worker.popevent()
|
||||
assert ev.name == "collectreport"
|
||||
rep = unserialize_report(ev.name, ev.kwargs["data"])
|
||||
rep = unserialize_report(ev.kwargs["data"])
|
||||
assert rep.skipped
|
||||
assert rep.longrepr[2] == "Skipped: hello"
|
||||
ev = worker.popevent("collectionfinish")
|
||||
assert not ev.kwargs["ids"]
|
||||
|
||||
def test_remote_collect_fail(self, worker):
|
||||
def test_remote_collect_fail(self, worker, unserialize_report):
|
||||
worker.testdir.makepyfile("""aasd qwe""")
|
||||
worker.setup()
|
||||
ev = worker.popevent("collectionstart")
|
||||
assert not ev.kwargs
|
||||
ev = worker.popevent()
|
||||
assert ev.name == "collectreport"
|
||||
rep = unserialize_report(ev.name, ev.kwargs["data"])
|
||||
rep = unserialize_report(ev.kwargs["data"])
|
||||
assert rep.failed
|
||||
ev = worker.popevent("collectionfinish")
|
||||
assert not ev.kwargs["ids"]
|
||||
|
||||
def test_runtests_all(self, worker):
|
||||
def test_runtests_all(self, worker, unserialize_report):
|
||||
worker.testdir.makepyfile(
|
||||
"""
|
||||
def test_func(): pass
|
||||
@@ -345,7 +174,7 @@ class TestWorkerInteractor:
|
||||
for i in range(3): # setup/call/teardown
|
||||
ev = worker.popevent("testreport")
|
||||
assert ev.name == "testreport"
|
||||
rep = unserialize_report(ev.name, ev.kwargs["data"])
|
||||
rep = unserialize_report(ev.kwargs["data"])
|
||||
assert rep.nodeid.endswith(func)
|
||||
ev = worker.popevent("workerfinished")
|
||||
assert "workeroutput" in ev.kwargs
|
||||
|
||||
Reference in New Issue
Block a user