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

1""" 

2Tracing utils 

3""" 

4 

5from __future__ import annotations 

6 

7from collections.abc import Sequence 

8from typing import Any 

9from typing import Callable 

10 

11 

12_Writer = Callable[[str], object] 

13_Processor = Callable[[tuple[str, ...], tuple[Any, ...]], object] 

14 

15 

16class TagTracer: 

17 def __init__(self) -> None: 

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

19 self._writer: _Writer | None = None 

20 self.indent = 0 

21 

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

23 return TagTracerSub(self, (name,)) 

24 

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

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

27 extra = args[-1] 

28 args = args[:-1] 

29 else: 

30 extra = {} 

31 

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

33 indent = " " * self.indent 

34 

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

36 

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

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

39 

40 return "".join(lines) 

41 

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

43 if self._writer is not None and args: 

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

45 try: 

46 processor = self._tags2proc[tags] 

47 except KeyError: 

48 pass 

49 else: 

50 processor(tags, args) 

51 

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

53 self._writer = writer 

54 

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

56 if isinstance(tags, str): 

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

58 else: 

59 assert isinstance(tags, tuple) 

60 self._tags2proc[tags] = processor 

61 

62 

63class TagTracerSub: 

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

65 self.root = root 

66 self.tags = tags 

67 

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

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

70 

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

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