Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pluggy/_tracing.py: 41%
41 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
1"""
2Tracing utils
3"""
6class TagTracer:
7 def __init__(self):
8 self._tags2proc = {}
9 self._writer = None
10 self.indent = 0
12 def get(self, name):
13 return TagTracerSub(self, (name,))
15 def _format_message(self, tags, args):
16 if isinstance(args[-1], dict):
17 extra = args[-1]
18 args = args[:-1]
19 else:
20 extra = {}
22 content = " ".join(map(str, args))
23 indent = " " * self.indent
25 lines = ["{}{} [{}]\n".format(indent, content, ":".join(tags))]
27 for name, value in extra.items():
28 lines.append(f"{indent} {name}: {value}\n")
30 return "".join(lines)
32 def _processmessage(self, tags, args):
33 if self._writer is not None and args:
34 self._writer(self._format_message(tags, args))
35 try:
36 processor = self._tags2proc[tags]
37 except KeyError:
38 pass
39 else:
40 processor(tags, args)
42 def setwriter(self, writer):
43 self._writer = writer
45 def setprocessor(self, tags, processor):
46 if isinstance(tags, str):
47 tags = tuple(tags.split(":"))
48 else:
49 assert isinstance(tags, tuple)
50 self._tags2proc[tags] = processor
53class TagTracerSub:
54 def __init__(self, root, tags):
55 self.root = root
56 self.tags = tags
58 def __call__(self, *args):
59 self.root._processmessage(self.tags, args)
61 def get(self, name):
62 return self.__class__(self.root, self.tags + (name,))