Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/log/host_log_collect.py: 100%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ---------------------------------------------------------------------------- 

4# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

5# 

6# Licensed under the Apache License, Version 2.0 (the "License"); 

7# you may not use this file except in compliance with the License. 

8# You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, software 

13# distributed under the License is distributed on an "AS IS" BASIS, 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

15# See the License for the specific language governing permissions and 

16# limitations under the License. 

17# ---------------------------------------------------------------------------- 

18 

19import os 

20 

21from common import FileOperate as f 

22from params import ParamDict 

23from common import consts, get_cann_log_path 

24from common.const import CANN_LOG_NAME, ATRACE_LOG_NAME 

25from common import log_warning 

26from common.file_operate import COPY_MODE, MOVE_MODE 

27 

28__all__ = ["collect_host_logs"] 

29 

30 

31def collect_messages(output_root_dir): 

32 root_path = os.sep 

33 message_root_dir = os.path.join(root_path, "var", "log") 

34 message_file_name = "syslog" if f.check_file(os.path.join(message_root_dir, "syslog")) else "messages" 

35 message_source = os.path.join(message_root_dir, message_file_name) 

36 message_target = os.path.join(output_root_dir, "dfx", "log", "host", "message") 

37 return f.collect_file_to_dir(message_source, message_target, COPY_MODE) 

38 

39 

40def collect_atrace_logs(output_root_dir): 

41 atrace_log_path, _ = get_cann_log_path(ATRACE_LOG_NAME) 

42 if f.check_dir(atrace_log_path): 

43 atrace_target = os.path.join(output_root_dir, "dfx", "atrace") 

44 return f.collect_dir(atrace_log_path, atrace_target, COPY_MODE) 

45 return False 

46 

47 

48def collect_cann_logs(output_root_dir): 

49 cann_log_path, env_path_name = get_cann_log_path(CANN_LOG_NAME) 

50 log_types = ["debug", "run", "security"] 

51 ret = True 

52 for log_type in log_types: 

53 cann_source = os.path.join(cann_log_path, log_type) 

54 cann_target = os.path.join(output_root_dir, "dfx", "log", "host", "cann", log_type) 

55 if not os.path.exists(cann_source): 

56 log_warning("No file exists in %s/%s" % (env_path_name, log_type)) 

57 if ParamDict().get_command() == consts.launch_cmd: 

58 ret = ret and f.collect_dir(cann_source, cann_target, MOVE_MODE) 

59 else: 

60 ret = ret and f.collect_dir(cann_source, cann_target, COPY_MODE) 

61 return ret 

62 

63 

64def collect_install_logs(output_root_dir): 

65 root_path = os.sep 

66 install_source = os.path.join(root_path, "var", "log", "ascend_seclog", "ascend_install.log") 

67 install_target = os.path.join(output_root_dir, "dfx", "log", "host", "install") 

68 return f.collect_file_to_dir(install_source, install_target, COPY_MODE) 

69 

70 

71def collect_host_logs(output_root_dir): 

72 err = 0 # err equals to zero denotes no errors 

73 message_ret = collect_messages(output_root_dir) 

74 if not message_ret: 

75 log_warning("Collect host messages failed.") 

76 err += 1 

77 cann_ret = collect_cann_logs(output_root_dir) 

78 if not cann_ret: 

79 log_warning("Collect host cann logs failed.") 

80 err += 1 

81 atrace_ret = collect_atrace_logs(output_root_dir) 

82 if not atrace_ret: 

83 log_warning("Collect host atrace failed.") 

84 err += 1 

85 install_ret = collect_install_logs(output_root_dir) 

86 if not install_ret: 

87 log_warning("Collect install logs failed.") 

88 err += 1 

89 return err == 0