move tracer to own file

Author: RonnyPfannschmidt

pluggy/__init__.py

@@ -104,68 +104,6 @@ def normalize_hookimpl_opts(opts):
     opts.setdefault("optionalhook", False)
 
 
-class _TagTracer(object):
-    def __init__(self):
-        self._tag2proc = {}
-        self.writer = None
-        self.indent = 0
-
-    def get(self, name):
-        return _TagTracerSub(self, (name,))
-
-    def format_message(self, tags, args):
-        if isinstance(args[-1], dict):
-            extra = args[-1]
-            args = args[:-1]
-        else:
-            extra = {}
-
-        content = " ".join(map(str, args))
-        indent = "  " * self.indent
-
-        lines = [
-            "%s%s [%s]\n" % (indent, content, ":".join(tags))
-        ]
-
-        for name, value in extra.items():
-            lines.append("%s    %s: %s\n" % (indent, name, value))
-        return lines
-
-    def processmessage(self, tags, args):
-        if self.writer is not None and args:
-            lines = self.format_message(tags, args)
-            self.writer(''.join(lines))
-        try:
-            self._tag2proc[tags](tags, args)
-        except KeyError:
-            pass
-
-    def setwriter(self, writer):
-        self.writer = writer
-
-    def setprocessor(self, tags, processor):
-        if isinstance(tags, str):
-            tags = tuple(tags.split(":"))
-        else:
-            assert isinstance(tags, tuple)
-        self._tag2proc[tags] = processor
-
-
-class _TagTracerSub(object):
-    def __init__(self, root, tags):
-        self.root = root
-        self.tags = tags
-
-    def __call__(self, *args):
-        self.root.processmessage(self.tags, args)
-
-    def setmyprocessor(self, processor):
-        self.root.setprocessor(self.tags, processor)
-
-    def get(self, name):
-        return self.__class__(self.root, self.tags + (name,))
-
-
 class _TracedHookExecution(object):
     def __init__(self, pluginmanager, before, after):
         self.pluginmanager = pluginmanager
@@ -206,7 +144,7 @@ def __init__(self, project_name, implprefix=None):
         self._name2plugin = {}
         self._plugin2hookcallers = {}
         self._plugin_distinfo = []
-        self.trace = _TagTracer().get("pluginmanage")
+        self.trace = _tracing.TagTracer().get("pluginmanage")
         self.hook = _HookRelay(self.trace.root.get("hook"))
         self._implprefix = implprefix
         self._inner_hookexec = lambda hook, methods, kwargs: \

pluggy/_tracing.py

@@ -0,0 +1,62 @@
+
+
+class TagTracer(object):
+    def __init__(self):
+        self._tag2proc = {}
+        self.writer = None
+        self.indent = 0
+
+    def get(self, name):
+        return TagTracerSub(self, (name,))
+
+    def format_message(self, tags, args):
+        if isinstance(args[-1], dict):
+            extra = args[-1]
+            args = args[:-1]
+        else:
+            extra = {}
+
+        content = " ".join(map(str, args))
+        indent = "  " * self.indent
+
+        lines = [
+            "%s%s [%s]\n" % (indent, content, ":".join(tags))
+        ]
+
+        for name, value in extra.items():
+            lines.append("%s    %s: %s\n" % (indent, name, value))
+        return lines
+
+    def processmessage(self, tags, args):
+        if self.writer is not None and args:
+            lines = self.format_message(tags, args)
+            self.writer(''.join(lines))
+        try:
+            self._tag2proc[tags](tags, args)
+        except KeyError:
+            pass
+
+    def setwriter(self, writer):
+        self.writer = writer
+
+    def setprocessor(self, tags, processor):
+        if isinstance(tags, str):
+            tags = tuple(tags.split(":"))
+        else:
+            assert isinstance(tags, tuple)
+        self._tag2proc[tags] = processor
+
+
+class TagTracerSub(object):
+    def __init__(self, root, tags):
+        self.root = root
+        self.tags = tags
+
+    def __call__(self, *args):
+        self.root.processmessage(self.tags, args)
+
+    def setmyprocessor(self, processor):
+        self.root.setprocessor(self.tags, processor)
+
+    def get(self, name):
+        return self.__class__(self.root, self.tags + (name,))

testing/test_tracer.py

@@ -1,9 +1,15 @@
 
-from pluggy import _TagTracer
+from pluggy._tracing import TagTracer
 
+import pytest
 
-def test_simple():
-    rootlogger = _TagTracer()
+
+@pytest.fixture
+def rootlogger():
+    return TagTracer()
+
+
+def test_simple(rootlogger):
     log = rootlogger.get("pytest")
     log("hello")
     out = []
@@ -16,8 +22,7 @@ def test_simple():
     assert out[1] == "hello [pytest:collection]\n"
 
 
-def test_indent():
-    rootlogger = _TagTracer()
+def test_indent(rootlogger):
     log = rootlogger.get("1")
     out = []
     log.root.setwriter(lambda arg: out.append(arg))
@@ -39,8 +44,7 @@ def test_indent():
         '    line3', '    line4', '  line5', 'last']
 
 
-def test_readable_output_dictargs():
-    rootlogger = _TagTracer()
+def test_readable_output_dictargs(rootlogger):
 
     out = rootlogger.format_message(['test'], [1])
     assert out == ['1 [test]\n']
@@ -52,8 +56,7 @@ def test_readable_output_dictargs():
     ]
 
 
-def test_setprocessor():
-    rootlogger = _TagTracer()
+def test_setprocessor(rootlogger):
     log = rootlogger.get("1")
     log2 = log.get("2")
     assert log2.tags == tuple("12")
@@ -73,8 +76,7 @@ def test_setprocessor():
     assert args == ("seen",)
 
 
-def test_setmyprocessor():
-    rootlogger = _TagTracer()
+def test_setmyprocessor(rootlogger):
     log = rootlogger.get("1")
     log2 = log.get("2")
     out = []