Coverage for tests / setup_logging.py: 62%

1import logging 

2import os 

3import inspect 

4import time 

5 

6 

7class CallerFormatter(logging.Formatter): 

8 """ 

9 Formatter that adds the external caller (e.g., test function) to each log record. 

10 """ 

11 

12 _last_time = None 

13 _start_time = time.time() 

14 

15 def format(self, record): 

16 # --- Find external caller --- 

17 stack = inspect.stack() 

18 for frame_info in stack[2:]: 

19 filename = frame_info.filename 

20 if "tests" in filename: 

21 record.external_caller = f"{os.path.relpath(filename)}:{frame_info.lineno} in {frame_info.function}()" 

22 break 

23 else: 

24 record.external_caller = "unknown" 

25 

26 # --- Timing info --- 

27 now = time.time() 

28 record.total_elapsed = now - self._start_time 

29 if self._last_time is None: 

30 record.delta = 0.0 

31 else: 

32 record.delta = now - self._last_time 

33 self._last_time = now 

34 

35 return super().format(record) 

36 

37 # def format(self, record): 

38 # # Walk up the stack to find the first frame outside the logging module 

39 # stack = inspect.stack() 

40 # for frame_info in stack[2:]: 

41 # filename = frame_info.filename 

42 # # Heuristic: find a file in your tests folder or not in rivapy 

43 # if "tests" in filename: 

44 # record.external_caller = f"{os.path.relpath(filename)}:{frame_info.lineno} in {frame_info.function}()" 

45 # break 

46 # else: 

47 # record.external_caller = "unknown" 

48 # return super().format(record) 

49 

50 

51def setup_logging_for_tests(log_file="rivapy_test.log"): 

52 """ 

53 Configure logging for unit tests. 

54 

55 This sets up a global logger for all rivapy.* modules so that 

56 logs are written both to console (INFO+) and to a test log file (DEBUG+). 

57 Safe to call multiple times — will not add duplicate handlers. 

58 """ 

59 logger = logging.getLogger("rivapy") 

60 logger.setLevel(logging.DEBUG) 

61 

62 # Prevent re-adding handlers if already configured 

63 if logger.handlers: 

64 return logger 

65 

66 # Ensure log directory exists 

67 log_path = os.path.abspath(log_file) 

68 os.makedirs(os.path.dirname(log_path), exist_ok=True) 

69 

70 # File handler — all logs (DEBUG and above) 

71 fh = logging.FileHandler(log_path, mode="w", encoding="utf-8") 

72 fh.setLevel(logging.DEBUG) 

73 

74 # Console handler — only INFO and above 

75 ch = logging.StreamHandler() 

76 ch.setLevel(logging.INFO) 

77 

78 # Common format 

79 formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") # does not include callback trace 

80 # formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(funcName)s() - %(message)s") 

81 

82 # Formatter includes function and external caller 

83 # formatter = CallerFormatter( 

84 # "%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(funcName)s() - [called by %(external_caller)s] - %(message)s" 

85 # ) 

86 # formatter = CallerFormatter( 

87 # "%(asctime)s - %(name)s - %(levelname)s - " 

88 # "%(filename)s:%(lineno)d - %(funcName)s() - " 

89 # "[called by %(external_caller)s] - %(message)s " 

90 # "(Δ +%(delta).3fs, total +%(total_elapsed).3fs)" 

91 # ) 

92 

93 # For callbacks and intense debugging 

94 # formatter = CallerFormatter( 

95 # "%(asctime)s (+%(delta).3fs, total +%(total_elapsed).3fs) - %(name)s - %(levelname)s - " 

96 # "%(filename)s:%(lineno)d - %(funcName)s() - " 

97 # "[called by %(external_caller)s] - %(message)s" 

98 # ) 

99 

100 fh.setFormatter(formatter) 

101 ch.setFormatter(formatter) 

102 

103 # Add both handlers 

104 logger.addHandler(fh) 

105 logger.addHandler(ch) 

106 

107 # Optional: fine-tune per module 

108 # logging.getLogger('BIG.marketdata').setLevel(logging.DEBUG) 

109 # logging.getLogger('BIG.pricing').setLevel(logging.DEBUG) 

110 # logging.getLogger('BIG.instruments').setLevel(logging.DEBUG) 

111 

112 return logger