Switch to using Pytest's terminal reporter

Signed-off-by: Pedro Algarvio <[email protected]>

Author: s0undt3ch

pytest_timeout.py

@@ -8,7 +8,6 @@
 """
 import inspect
 import os
-import shutil
 import signal
 import sys
 import threading
@@ -445,11 +444,12 @@ def timeout_sigalrm(item, settings):
         return
     __tracebackhide__ = True
     nthreads = len(threading.enumerate())
+    terminal = item.config.get_terminal_writer()
     if nthreads > 1:
-        write_title("Timeout", sep="+")
-    dump_stacks()
+        terminal.sep("+", title="Timeout")
+    dump_stacks(terminal)
     if nthreads > 1:
-        write_title("Timeout", sep="+")
+        terminal.sep("+", title="Timeout")
     pytest.fail("Timeout >%ss" % settings.timeout)
 
 
@@ -461,28 +461,29 @@ def timeout_timer(item, settings):
     """
     if not settings.disable_debugger_detection and is_debugging():
         return
+    terminal = item.config.get_terminal_writer()
     try:
         capman = item.config.pluginmanager.getplugin("capturemanager")
         if capman:
             capman.suspend_global_capture(item)
             stdout, stderr = capman.read_global_capture()
         else:
             stdout, stderr = None, None
-        write_title("Timeout", sep="+")
+        terminal.sep("+", title="Timeout")
         caplog = item.config.pluginmanager.getplugin("_capturelog")
         if caplog and hasattr(item, "capturelog_handler"):
             log = item.capturelog_handler.stream.getvalue()
             if log:
-                write_title("Captured log")
-                write(log)
+                terminal.sep("~", title="Captured log")
+                terminal.write(log)
         if stdout:
-            write_title("Captured stdout")
-            write(stdout)
+            terminal.sep("~", title="Captured stdout")
+            terminal.write(stdout)
         if stderr:
-            write_title("Captured stderr")
-            write(stderr)
-        dump_stacks()
-        write_title("Timeout", sep="+")
+            terminal.sep("~", title="Captured stderr")
+            terminal.write(stderr)
+        dump_stacks(terminal)
+        terminal.sep("+", title="Timeout")
     except Exception:
         traceback.print_exc()
     finally:
@@ -491,7 +492,7 @@ def timeout_timer(item, settings):
         os._exit(1)
 
 
-def dump_stacks():
+def dump_stacks(terminal):
     """Dump the stacks of all threads except the current thread."""
     current_ident = threading.current_thread().ident
     for thread_ident, frame in sys._current_frames().items():
@@ -503,31 +504,5 @@ def dump_stacks():
                 break
         else:
             thread_name = "<unknown>"
-        write_title("Stack of %s (%s)" % (thread_name, thread_ident))
-        write("".join(traceback.format_stack(frame)))
-
-
-def write_title(title, stream=None, sep="~"):
-    """Write a section title.
-
-    If *stream* is None sys.stderr will be used, *sep* is used to
-    draw the line.
-    """
-    if stream is None:
-        stream = sys.stderr
-    width, height = shutil.get_terminal_size()
-    fill = int((width - len(title) - 2) / 2)
-    line = " ".join([sep * fill, title, sep * fill])
-    if len(line) < width:
-        line += sep * (width - len(line))
-    stream.write("\n" + line + "\n")
-
-
-def write(text, stream=None):
-    """Write text to stream.
-
-    Pretty stupid really, only here for symmetry with .write_title().
-    """
-    if stream is None:
-        stream = sys.stderr
-    stream.write(text)
+        terminal.sep("~", title="Stack of %s (%s)" % (thread_name, thread_ident))
+        terminal.write("".join(traceback.format_stack(frame)))