Coverage for /usr/local/lib/python3.7/site-packages/pluggy/_tracing.py : 17%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Tracing utils
3"""
4from .callers import _Result
7class TagTracer(object):
8 def __init__(self):
9 self._tag2proc = {}
10 self.writer = None
11 self.indent = 0
13 def get(self, name):
14 return TagTracerSub(self, (name,))
16 def format_message(self, tags, args):
17 if isinstance(args[-1], dict):
18 extra = args[-1]
19 args = args[:-1]
20 else:
21 extra = {}
23 content = " ".join(map(str, args))
24 indent = " " * self.indent
26 lines = ["%s%s [%s]\n" % (indent, content, ":".join(tags))]
28 for name, value in extra.items():
29 lines.append("%s %s: %s\n" % (indent, name, value))
30 return lines
32 def processmessage(self, tags, args):
33 if self.writer is not None and args:
34 lines = self.format_message(tags, args)
35 self.writer("".join(lines))
36 try:
37 self._tag2proc[tags](tags, args)
38 except KeyError:
39 pass
41 def setwriter(self, writer):
42 self.writer = writer
44 def setprocessor(self, tags, processor):
45 if isinstance(tags, str):
46 tags = tuple(tags.split(":"))
47 else:
48 assert isinstance(tags, tuple)
49 self._tag2proc[tags] = processor
52class TagTracerSub(object):
53 def __init__(self, root, tags):
54 self.root = root
55 self.tags = tags
57 def __call__(self, *args):
58 self.root.processmessage(self.tags, args)
60 def setmyprocessor(self, processor):
61 self.root.setprocessor(self.tags, processor)
63 def get(self, name):
64 return self.__class__(self.root, self.tags + (name,))
67class _TracedHookExecution(object):
68 def __init__(self, pluginmanager, before, after):
69 self.pluginmanager = pluginmanager
70 self.before = before
71 self.after = after
72 self.oldcall = pluginmanager._inner_hookexec
73 assert not isinstance(self.oldcall, _TracedHookExecution)
74 self.pluginmanager._inner_hookexec = self
76 def __call__(self, hook, hook_impls, kwargs):
77 self.before(hook.name, hook_impls, kwargs)
78 outcome = _Result.from_call(lambda: self.oldcall(hook, hook_impls, kwargs))
79 self.after(outcome, hook.name, hook_impls, kwargs)
80 return outcome.get_result()
82 def undo(self):
83 self.pluginmanager._inner_hookexec = self.oldcall