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

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 

20import itertools 

21 

22from common import FileOperate as f 

23from common import log_warning, log_error 

24from common.file_operate import MOVE_MODE 

25 

26 

27__all__ = ["collect_device_logs"] 

28 

29 

30def collect_device(source_root_dir, output_root_dir, log_types, feature): 

31 slog_path = os.path.join(source_root_dir, "slog") 

32 dev_os_dirs = f.list_dir(slog_path) 

33 if not dev_os_dirs: 

34 return False 

35 target_dict = { 

36 "device-app": "application", 

37 "device-os": "system", 

38 "device-": "firmware", 

39 "event": "system" 

40 } 

41 ret = True 

42 for dev_os_id, log_type in itertools.product(dev_os_dirs, log_types): 

43 device_type_path = os.path.join(slog_path, dev_os_id, log_type) 

44 device_dirs = f.list_dir(device_type_path) 

45 if not device_dirs: 

46 continue 

47 for device_dir in device_dirs: 

48 if not device_dir.startswith(feature): 

49 continue 

50 device_source = os.path.join(slog_path, dev_os_id, log_type, device_dir) 

51 if log_type: 

52 device_target = os.path.join(output_root_dir, "dfx", "log", "device", 

53 dev_os_id, target_dict[feature], log_type, device_dir) 

54 else: 

55 device_target = os.path.join(output_root_dir, "dfx", "log", "device", 

56 dev_os_id, target_dict[feature], "debug", device_dir) 

57 ret = ret and f.collect_dir(device_source, device_target, MOVE_MODE) 

58 return ret 

59 

60 

61def collect_messages(source_root_dir, output_root_dir): 

62 message_dir = os.path.join(source_root_dir, "message") 

63 dev_os_dirs = f.list_dir(message_dir) 

64 if not dev_os_dirs: 

65 return False 

66 ret = True 

67 for dev_os_id in dev_os_dirs: 

68 message_target = os.path.join(output_root_dir, "dfx", "log", "device", dev_os_id, "system", "message") 

69 message_source = os.path.join(message_dir, dev_os_id) 

70 ret = ret and f.collect_dir(message_source, message_target, MOVE_MODE) 

71 return ret 

72 

73 

74def collect_stackcore(source_root_dir, output_root_dir): 

75 stackcore_source = os.path.join(source_root_dir, "stackcore") 

76 stackcore_target = os.path.join(output_root_dir, "dfx", "stackcore") 

77 return f.collect_dir(stackcore_source, stackcore_target, MOVE_MODE) 

78 

79 

80def collect_bbox(source_root_dir, output_root_dir): 

81 bbox_source = os.path.join(source_root_dir, "hisi_logs") 

82 bbox_target = os.path.join(output_root_dir, "dfx", "bbox") 

83 return f.collect_dir(bbox_source, bbox_target, MOVE_MODE) 

84 

85 

86def collect_host_driver(source_root_dir, output_root_dir): 

87 host_driver_path = os.path.join(source_root_dir, "slog", "host") 

88 host_driver_dirs = f.list_dir(host_driver_path) 

89 if not host_driver_dirs: 

90 return False 

91 ret = True 

92 host_driver_log_target = os.path.join(output_root_dir, "dfx", "log", "host", "driver") 

93 for host_driver_log in host_driver_dirs: 

94 host_driver_log_source = os.path.join(host_driver_path, host_driver_log) 

95 ret = ret and f.collect_file_to_dir(host_driver_log_source, host_driver_log_target, MOVE_MODE) 

96 f.remove_dir(host_driver_path) 

97 return ret 

98 

99 

100def collect_slogd(source_root_dir, output_root_dir): 

101 slog_path = os.path.join(source_root_dir, "slog") 

102 slog_dirs = f.list_dir(slog_path) 

103 if not slog_dirs: 

104 return False 

105 ret = True 

106 for slog_dir in slog_dirs: 

107 slogd_log_source = os.path.join(slog_path, slog_dir, "slogd") 

108 if f.check_dir(slogd_log_source): 

109 slogd_log_target = os.path.join(output_root_dir, "dfx", "log", "device", slog_dir, "slogd") 

110 ret = ret and f.collect_dir(slogd_log_source, slogd_log_target, MOVE_MODE) 

111 return ret 

112 

113 

114def collect_device_app(source_root_dir, output_root_dir): 

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

116 feature = "device-app" 

117 return collect_device(source_root_dir, output_root_dir, log_types, feature) 

118 

119 

120def collect_device_os(source_root_dir, output_root_dir): 

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

122 feature = "device-os" 

123 return collect_device(source_root_dir, output_root_dir, log_types, feature) 

124 

125 

126def collect_device_id(source_root_dir, output_root_dir): 

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

128 feature = "device-" 

129 return collect_device(source_root_dir, output_root_dir, log_types, feature) 

130 

131 

132def collect_event(source_root_dir, output_root_dir): 

133 log_types = ["run"] 

134 feature = "event" 

135 return collect_device(source_root_dir, output_root_dir, log_types, feature) 

136 

137 

138def collect_device_logs(source_root_dir, output_root_dir): 

139 if not source_root_dir: 

140 log_error("msnpureport output directory: {} is not exist".format(source_root_dir)) 

141 return False 

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

143 message_ret = collect_messages(source_root_dir, output_root_dir) 

144 if not message_ret: 

145 log_warning("Collect device messages failed.") 

146 err += 1 

147 stackcore_ret = collect_stackcore(source_root_dir, output_root_dir) 

148 if not stackcore_ret: 

149 log_warning("Collect device stackcore failed.") 

150 err += 1 

151 bbox_ret = collect_bbox(source_root_dir, output_root_dir) 

152 if not bbox_ret: 

153 log_warning("Collect device bbox failed.") 

154 err += 1 

155 host_driver_ret = collect_host_driver(source_root_dir, output_root_dir) 

156 if not host_driver_ret: 

157 log_warning("Collect host driver log failed.") 

158 err += 1 

159 slogd_ret = collect_slogd(source_root_dir, output_root_dir) 

160 if not slogd_ret: 

161 log_warning("Collect device slogd logs failed.") 

162 err += 1 

163 device_app_ret = collect_device_app(source_root_dir, output_root_dir) 

164 if not device_app_ret: 

165 log_warning("Collect device device_app logs failed.") 

166 err += 1 

167 device_os_ret = collect_device_os(source_root_dir, output_root_dir) 

168 if not device_os_ret: 

169 log_warning("Collect device device_os logs failed.") 

170 err += 1 

171 device_id_ret = collect_device_id(source_root_dir, output_root_dir) 

172 if not device_id_ret: 

173 log_warning("Collect device device_id logs failed.") 

174 err += 1 

175 event_ret = collect_event(source_root_dir, output_root_dir) 

176 if not event_ret: 

177 log_warning("Collect device event logs failed.") 

178 err += 1 

179 return err == 0