1# -*- coding: utf-8 -*-
2from pprint import pprint
3
4import py
5import six
6
7from _pytest._code.code import ExceptionInfo
8from _pytest._code.code import ReprEntry
9from _pytest._code.code import ReprEntryNative
10from _pytest._code.code import ReprExceptionInfo
11from _pytest._code.code import ReprFileLocation
12from _pytest._code.code import ReprFuncArgs
13from _pytest._code.code import ReprLocals
14from _pytest._code.code import ReprTraceback
15from _pytest._code.code import TerminalRepr
16from _pytest.outcomes import skip
17from _pytest.pathlib import Path
18
19
20def getslaveinfoline(node):
21    try:
22        return node._slaveinfocache
23    except AttributeError:
24        d = node.slaveinfo
25        ver = "%s.%s.%s" % d["version_info"][:3]
26        node._slaveinfocache = s = "[%s] %s -- Python %s %s" % (
27            d["id"],
28            d["sysplatform"],
29            ver,
30            d["executable"],
31        )
32        return s
33
34
35class BaseReport(object):
36    when = None
37    location = None
38
39    def __init__(self, **kw):
40        self.__dict__.update(kw)
41
42    def toterminal(self, out):
43        if hasattr(self, "node"):
44            out.line(getslaveinfoline(self.node))
45
46        longrepr = self.longrepr
47        if longrepr is None:
48            return
49
50        if hasattr(longrepr, "toterminal"):
51            longrepr.toterminal(out)
52        else:
53            try:
54                out.line(longrepr)
55            except UnicodeEncodeError:
56                out.line("<unprintable longrepr>")
57
58    def get_sections(self, prefix):
59        for name, content in self.sections:
60            if name.startswith(prefix):
61                yield prefix, content
62
63    @property
64    def longreprtext(self):
65        """
66        Read-only property that returns the full string representation
67        of ``longrepr``.
68
69        .. versionadded:: 3.0
70        """
71        tw = py.io.TerminalWriter(stringio=True)
72        tw.hasmarkup = False
73        self.toterminal(tw)
74        exc = tw.stringio.getvalue()
75        return exc.strip()
76
77    @property
78    def caplog(self):
79        """Return captured log lines, if log capturing is enabled
80
81        .. versionadded:: 3.5
82        """
83        return "\n".join(
84            content for (prefix, content) in self.get_sections("Captured log")
85        )
86
87    @property
88    def capstdout(self):
89        """Return captured text from stdout, if capturing is enabled
90
91        .. versionadded:: 3.0
92        """
93        return "".join(
94            content for (prefix, content) in self.get_sections("Captured stdout")
95        )
96
97    @property
98    def capstderr(self):
99        """Return captured text from stderr, if capturing is enabled
100
101        .. versionadded:: 3.0
102        """
103        return "".join(
104            content for (prefix, content) in self.get_sections("Captured stderr")
105        )
106
107    passed = property(lambda x: x.outcome == "passed")
108    failed = property(lambda x: x.outcome == "failed")
109    skipped = property(lambda x: x.outcome == "skipped")
110
111    @property
112    def fspath(self):
113        return self.nodeid.split("::")[0]
114
115    @property
116    def count_towards_summary(self):
117        """
118        **Experimental**
119
120        Returns True if this report should be counted towards the totals shown at the end of the
121        test session: "1 passed, 1 failure, etc".
122
123        .. note::
124
125            This function is considered **experimental**, so beware that it is subject to changes
126            even in patch releases.
127        """
128        return True
129
130    @property
131    def head_line(self):
132        """
133        **Experimental**
134
135        Returns the head line shown with longrepr output for this report, more commonly during
136        traceback representation during failures::
137
138            ________ Test.foo ________
139
140
141        In the example above, the head_line is "Test.foo".
142
143        .. note::
144
145            This function is considered **experimental**, so beware that it is subject to changes
146            even in patch releases.
147        """
148        if self.location is not None:
149            fspath, lineno, domain = self.location
150            return domain
151
152    def _get_verbose_word(self, config):
153        _category, _short, verbose = config.hook.pytest_report_teststatus(
154            report=self, config=config
155        )
156        return verbose
157
158    def _to_json(self):
159        """
160        This was originally the serialize_report() function from xdist (ca03269).
161
162        Returns the contents of this report as a dict of builtin entries, suitable for
163        serialization.
164
165        Experimental method.
166        """
167
168        def disassembled_report(rep):
169            reprtraceback = rep.longrepr.reprtraceback.__dict__.copy()
170            reprcrash = rep.longrepr.reprcrash.__dict__.copy()
171
172            new_entries = []
173            for entry in reprtraceback["reprentries"]:
174                entry_data = {
175                    "type": type(entry).__name__,
176                    "data": entry.__dict__.copy(),
177                }
178                for key, value in entry_data["data"].items():
179                    if hasattr(value, "__dict__"):
180                        entry_data["data"][key] = value.__dict__.copy()
181                new_entries.append(entry_data)
182
183            reprtraceback["reprentries"] = new_entries
184
185            return {
186                "reprcrash": reprcrash,
187                "reprtraceback": reprtraceback,
188                "sections": rep.longrepr.sections,
189            }
190
191        d = self.__dict__.copy()
192        if hasattr(self.longrepr, "toterminal"):
193            if hasattr(self.longrepr, "reprtraceback") and hasattr(
194                self.longrepr, "reprcrash"
195            ):
196                d["longrepr"] = disassembled_report(self)
197            else:
198                d["longrepr"] = six.text_type(self.longrepr)
199        else:
200            d["longrepr"] = self.longrepr
201        for name in d:
202            if isinstance(d[name], (py.path.local, Path)):
203                d[name] = str(d[name])
204            elif name == "result":
205                d[name] = None  # for now
206        return d
207
208    @classmethod
209    def _from_json(cls, reportdict):
210        """
211        This was originally the serialize_report() function from xdist (ca03269).
212
213        Factory method that returns either a TestReport or CollectReport, depending on the calling
214        class. It's the callers responsibility to know which class to pass here.
215
216        Experimental method.
217        """
218        if reportdict["longrepr"]:
219            if (
220                "reprcrash" in reportdict["longrepr"]
221                and "reprtraceback" in reportdict["longrepr"]
222            ):
223
224                reprtraceback = reportdict["longrepr"]["reprtraceback"]
225                reprcrash = reportdict["longrepr"]["reprcrash"]
226
227                unserialized_entries = []
228                reprentry = None
229                for entry_data in reprtraceback["reprentries"]:
230                    data = entry_data["data"]
231                    entry_type = entry_data["type"]
232                    if entry_type == "ReprEntry":
233                        reprfuncargs = None
234                        reprfileloc = None
235                        reprlocals = None
236                        if data["reprfuncargs"]:
237                            reprfuncargs = ReprFuncArgs(**data["reprfuncargs"])
238                        if data["reprfileloc"]:
239                            reprfileloc = ReprFileLocation(**data["reprfileloc"])
240                        if data["reprlocals"]:
241                            reprlocals = ReprLocals(data["reprlocals"]["lines"])
242
243                        reprentry = ReprEntry(
244                            lines=data["lines"],
245                            reprfuncargs=reprfuncargs,
246                            reprlocals=reprlocals,
247                            filelocrepr=reprfileloc,
248                            style=data["style"],
249                        )
250                    elif entry_type == "ReprEntryNative":
251                        reprentry = ReprEntryNative(data["lines"])
252                    else:
253                        _report_unserialization_failure(entry_type, cls, reportdict)
254                    unserialized_entries.append(reprentry)
255                reprtraceback["reprentries"] = unserialized_entries
256
257                exception_info = ReprExceptionInfo(
258                    reprtraceback=ReprTraceback(**reprtraceback),
259                    reprcrash=ReprFileLocation(**reprcrash),
260                )
261
262                for section in reportdict["longrepr"]["sections"]:
263                    exception_info.addsection(*section)
264                reportdict["longrepr"] = exception_info
265
266        return cls(**reportdict)
267
268
269def _report_unserialization_failure(type_name, report_class, reportdict):
270    url = "https://github.com/pytest-dev/pytest/issues"
271    stream = py.io.TextIO()
272    pprint("-" * 100, stream=stream)
273    pprint("INTERNALERROR: Unknown entry type returned: %s" % type_name, stream=stream)
274    pprint("report_name: %s" % report_class, stream=stream)
275    pprint(reportdict, stream=stream)
276    pprint("Please report this bug at %s" % url, stream=stream)
277    pprint("-" * 100, stream=stream)
278    raise RuntimeError(stream.getvalue())
279
280
281class TestReport(BaseReport):
282    """ Basic test report object (also used for setup and teardown calls if
283    they fail).
284    """
285
286    __test__ = False
287
288    def __init__(
289        self,
290        nodeid,
291        location,
292        keywords,
293        outcome,
294        longrepr,
295        when,
296        sections=(),
297        duration=0,
298        user_properties=None,
299        **extra
300    ):
301        #: normalized collection node id
302        self.nodeid = nodeid
303
304        #: a (filesystempath, lineno, domaininfo) tuple indicating the
305        #: actual location of a test item - it might be different from the
306        #: collected one e.g. if a method is inherited from a different module.
307        self.location = location
308
309        #: a name -> value dictionary containing all keywords and
310        #: markers associated with a test invocation.
311        self.keywords = keywords
312
313        #: test outcome, always one of "passed", "failed", "skipped".
314        self.outcome = outcome
315
316        #: None or a failure representation.
317        self.longrepr = longrepr
318
319        #: one of 'setup', 'call', 'teardown' to indicate runtest phase.
320        self.when = when
321
322        #: user properties is a list of tuples (name, value) that holds user
323        #: defined properties of the test
324        self.user_properties = list(user_properties or [])
325
326        #: list of pairs ``(str, str)`` of extra information which needs to
327        #: marshallable. Used by pytest to add captured text
328        #: from ``stdout`` and ``stderr``, but may be used by other plugins
329        #: to add arbitrary information to reports.
330        self.sections = list(sections)
331
332        #: time it took to run just the test
333        self.duration = duration
334
335        self.__dict__.update(extra)
336
337    def __repr__(self):
338        return "<%s %r when=%r outcome=%r>" % (
339            self.__class__.__name__,
340            self.nodeid,
341            self.when,
342            self.outcome,
343        )
344
345    @classmethod
346    def from_item_and_call(cls, item, call):
347        """
348        Factory method to create and fill a TestReport with standard item and call info.
349        """
350        when = call.when
351        duration = call.stop - call.start
352        keywords = {x: 1 for x in item.keywords}
353        excinfo = call.excinfo
354        sections = []
355        if not call.excinfo:
356            outcome = "passed"
357            longrepr = None
358        else:
359            if not isinstance(excinfo, ExceptionInfo):
360                outcome = "failed"
361                longrepr = excinfo
362            elif excinfo.errisinstance(skip.Exception):
363                outcome = "skipped"
364                r = excinfo._getreprcrash()
365                longrepr = (str(r.path), r.lineno, r.message)
366            else:
367                outcome = "failed"
368                if call.when == "call":
369                    longrepr = item.repr_failure(excinfo)
370                else:  # exception in setup or teardown
371                    longrepr = item._repr_failure_py(
372                        excinfo, style=item.config.getoption("tbstyle", "auto")
373                    )
374        for rwhen, key, content in item._report_sections:
375            sections.append(("Captured %s %s" % (key, rwhen), content))
376        return cls(
377            item.nodeid,
378            item.location,
379            keywords,
380            outcome,
381            longrepr,
382            when,
383            sections,
384            duration,
385            user_properties=item.user_properties,
386        )
387
388
389class CollectReport(BaseReport):
390    when = "collect"
391
392    def __init__(self, nodeid, outcome, longrepr, result, sections=(), **extra):
393        self.nodeid = nodeid
394        self.outcome = outcome
395        self.longrepr = longrepr
396        self.result = result or []
397        self.sections = list(sections)
398        self.__dict__.update(extra)
399
400    @property
401    def location(self):
402        return (self.fspath, None, self.fspath)
403
404    def __repr__(self):
405        return "<CollectReport %r lenresult=%s outcome=%r>" % (
406            self.nodeid,
407            len(self.result),
408            self.outcome,
409        )
410
411
412class CollectErrorRepr(TerminalRepr):
413    def __init__(self, msg):
414        self.longrepr = msg
415
416    def toterminal(self, out):
417        out.line(self.longrepr, red=True)
418
419
420def pytest_report_to_serializable(report):
421    if isinstance(report, (TestReport, CollectReport)):
422        data = report._to_json()
423        data["_report_type"] = report.__class__.__name__
424        return data
425
426
427def pytest_report_from_serializable(data):
428    if "_report_type" in data:
429        if data["_report_type"] == "TestReport":
430            return TestReport._from_json(data)
431        elif data["_report_type"] == "CollectReport":
432            return CollectReport._from_json(data)
433        assert False, "Unknown report_type unserialize data: {}".format(
434            data["_report_type"]
435        )
436