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

1""" 

2Tracing utils 

3""" 

4 

5from __future__ import annotations 

6 

7from typing import Any 

8from typing import Callable 

9from typing import Sequence 

10from typing import Tuple 

11 

12 

13_Writer = Callable[[str], object] 

14_Processor = Callable[[Tuple[str, ...], Tuple[Any, ...]], object] 

15 

16 

17class TagTracer: 

18 def __init__(self) -> None: 

19 self._tags2proc: dict[tuple[str, ...], _Processor] = {} 

20 self._writer: _Writer | None = None 

21 self.indent = 0 

22 

23 def get(self, name: str) -> TagTracerSub: 

24 return TagTracerSub(self, (name,)) 

25 

26 def _format_message(self, tags: Sequence[str], args: Sequence[object]) -> str: 

27 if isinstance(args[-1], dict): 

28 extra = args[-1] 

29 args = args[:-1] 

30 else: 

31 extra = {} 

32 

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

34 indent = " " * self.indent 

35 

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

37 

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

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

40 

41 return "".join(lines) 

42 

43 def _processmessage(self, tags: tuple[str, ...], args: tuple[object, ...]) -> None: 

44 if self._writer is not None and args: 

45 self._writer(self._format_message(tags, args)) 

46 try: 

47 processor = self._tags2proc[tags] 

48 except KeyError: 

49 pass 

50 else: 

51 processor(tags, args) 

52 

53 def setwriter(self, writer: _Writer | None) -> None: 

54 self._writer = writer 

55 

56 def setprocessor(self, tags: str | tuple[str, ...], processor: _Processor) -> None: 

57 if isinstance(tags, str): 

58 tags = tuple(tags.split(":")) 

59 else: 

60 assert isinstance(tags, tuple) 

61 self._tags2proc[tags] = processor 

62 

63 

64class TagTracerSub: 

65 def __init__(self, root: TagTracer, tags: tuple[str, ...]) -> None: 

66 self.root = root 

67 self.tags = tags 

68 

69 def __call__(self, *args: object) -> None: 

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

71 

72 def get(self, name: str) -> TagTracerSub: 

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