Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pluggy/_tracing.py: 27%

41 statements  

« prev     ^ index     » next       coverage.py v7.0.1, created at 2022-12-25 06:11 +0000

1""" 

2Tracing utils 

3""" 

4 

5 

6class TagTracer: 

7 def __init__(self): 

8 self._tags2proc = {} 

9 self._writer = None 

10 self.indent = 0 

11 

12 def get(self, name): 

13 return TagTracerSub(self, (name,)) 

14 

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 = {} 

21 

22 content = " ".join(map(str, args)) 

23 indent = " " * self.indent 

24 

25 lines = ["{}{} [{}]\n".format(indent, content, ":".join(tags))] 

26 

27 for name, value in extra.items(): 

28 lines.append(f"{indent} {name}: {value}\n") 

29 

30 return "".join(lines) 

31 

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) 

41 

42 def setwriter(self, writer): 

43 self._writer = writer 

44 

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 

51 

52 

53class TagTracerSub: 

54 def __init__(self, root, tags): 

55 self.root = root 

56 self.tags = tags 

57 

58 def __call__(self, *args): 

59 self.root._processmessage(self.tags, args) 

60 

61 def get(self, name): 

62 return self.__class__(self.root, self.tags + (name,))