Coverage for src/unit_cooler/actuator/work_log.py: 95%

1#!/usr/bin/env python3 

2""" 

3作動ログを記録します。主にテストで使用します。 

4 

5Usage: 

6 work_log.py [-c CONFIG] [-D] 

7 

8Options: 

9 -c CONFIG : CONFIG を設定ファイルとして読み込んで実行します。[default: config.yaml] 

10 -D : デバッグモードで動作します。 

11""" 

12 

13import logging 

14 

15import my_lib.webapp.event 

16import my_lib.webapp.log 

17 

18import unit_cooler.const 

19import unit_cooler.util 

20 

21config = None 

22enven_queue = None 

23 

24log_hist = [] 

25 

26 

27def init(config_, event_queue_): 

28 global config # noqa: PLW0603 

29 global event_queue # noqa: PLW0603 

30 

31 config = config_ 

32 event_queue = event_queue_ 

33 

34 

35def term(): 

36 global event_queue 

37 my_lib.webapp.log.term() 

38 

39 

40# NOTE: テスト用 

41def hist_clear(): 

42 global log_hist # noqa: PLW0603 

43 

44 log_hist = [] 

45 

46 

47# NOTE: テスト用 

48def hist_get(): 

49 global log_hist 

50 

51 return log_hist 

52 

53 

54def add(message, level=unit_cooler.const.LOG_LEVEL.INFO): 

55 global log_hist 

56 global config 

57 global event_queue 

58 

59 event_queue.put(my_lib.webapp.event.EVENT_TYPE.LOG) 

60 my_lib.webapp.log.add(message, level) 

61 

62 log_hist.append(message) 

63 

64 if level == unit_cooler.const.LOG_LEVEL.ERROR: 

65 unit_cooler.util.notify_error(config, message) 

66 # エラーメトリクス記録 

67 try: 

68 from unit_cooler.actuator.webapi.metrics import record_error 

69 

70 record_error("work_log_error", message) 

71 except ImportError: 

72 pass 

73 elif level == unit_cooler.const.LOG_LEVEL.WARN: 

74 # 警告メトリクス記録 

75 try: 

76 from unit_cooler.actuator.webapi.metrics import record_warning 

77 

78 record_warning("work_log_warning", message) 

79 except ImportError: 

80 pass 

81 

82 

83if __name__ == "__main__": 

84 # TEST Code 

85 import multiprocessing 

86 

87 import docopt 

88 import my_lib.config 

89 import my_lib.logger 

90 import my_lib.pretty 

91 import my_lib.webapp.config 

92 

93 args = docopt.docopt(__doc__) 

94 

95 config_file = args["-c"] 

96 debug_mode = args["-D"] 

97 

98 my_lib.logger.init("test", level=logging.DEBUG if debug_mode else logging.INFO) 

99 

100 config = my_lib.config.load(config_file) 

101 event_queue = multiprocessing.Queue() 

102 

103 my_lib.webapp.config.init(config["actuator"]) 

104 my_lib.webapp.log.init(config) 

105 init(config, event_queue) 

106 

107 add("Test", unit_cooler.const.LOG_LEVEL.INFO) 

108 add("Test", unit_cooler.const.LOG_LEVEL.WARN) 

109 add("Test", unit_cooler.const.LOG_LEVEL.ERROR) 

110 

111 logging.info(my_lib.pretty.format(hist_get())) 

112 

113 term()