Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/health/asys_health.py: 72%

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 sys 

21 

22from common import DeviceInfo 

23from common import log_error 

24from common import consts 

25from common.const import UNKNOWN, NONE 

26from params import ParamDict 

27from view import generate_report 

28 

29 

30class AsysHealth: 

31 """device health check""" 

32 def __init__(self): 

33 self.device = DeviceInfo() 

34 

35 @staticmethod 

36 def _get_highest_status(ret): 

37 critical = "Critical" 

38 alarm = "Alarm" 

39 warn = "Warning" 

40 healthy = "Healthy" 

41 all_device_status = [info[0] for info in ret.values()] 

42 

43 if UNKNOWN in all_device_status: 

44 return UNKNOWN 

45 if critical in all_device_status: 

46 return critical 

47 if alarm in all_device_status: 

48 return alarm 

49 if warn in all_device_status: 

50 return warn 

51 if healthy in all_device_status: 

52 return healthy 

53 return UNKNOWN 

54 

55 @staticmethod 

56 def _save_file(ret): 

57 """save file""" 

58 save_str = "" 

59 # Example: err_info = [health, [[error_code, error_msg], [error_code, error_msg] ...]] 

60 for device_id in sorted(ret.keys()): 

61 err_info = ret[device_id] 

62 table_header = [ 

63 [f"Device ID: {device_id}", f"Overall Health: {err_info[0]}"], 

64 ["", f"ErrorCode Num: {len(err_info[1])}"] 

65 ] 

66 table_data = {NONE: err_info[1]} 

67 save_str += generate_report(table_header, table_data, split_line=True) 

68 try: 

69 output_file = os.path.join(ParamDict().asys_output_timestamp_dir, "health_result.txt") 

70 with open(output_file, "w", encoding="utf8") as file: 

71 file.write(save_str) 

72 except Exception as e: 

73 log_error(f"Failed to save result: {e}.") 

74 

75 def run_health_check(self, diagnose_devices): 

76 """Multi-thread parallel execution""" 

77 ret = {} 

78 for device_id in diagnose_devices: 

79 # status -> "Healthy"/"Warning"/"Alarm"/"Critical"/"Unknown" 

80 status = self.device.get_device_health(device_id) 

81 # Example: err_info = [[error_code, error_msg], [error_code, error_msg] ...] 

82 err_info = self.device.get_device_errorcode(device_id) 

83 ret[device_id] = [status, err_info] 

84 return ret 

85 

86 def _print_screen(self, device_id, ret): 

87 """print screen""" 

88 if device_id is False: 

89 # without '-d', displays brief information about all devices. 

90 highest_status = self._get_highest_status(ret) 

91 table_header = [[f"Group of {len(ret)} Device", f"Overall Health: {highest_status}"]] 

92 table_data = {NONE: [[f"Device ID: {idx}", ret[idx][0]] for idx in sorted(ret.keys())]} 

93 else: 

94 # with '-d', displays detailed information about the input device. 

95 table_header = [ 

96 [f"Device ID: {device_id}", f"Overall Health: {ret[device_id][0]}"], 

97 ["", f"ErrorCode Num: {len(ret[device_id][1])}"] 

98 ] 

99 table_data = {NONE: ret[device_id][1]} 

100 # Only the first five records are displayed on the screen. 

101 if len(table_data[NONE]) > 5: 

102 table_data[NONE] = table_data[NONE][:5] 

103 table_data[NONE].append(["......", "......"]) 

104 print_str = generate_report(table_header, table_data, split_line=True) 

105 sys.stdout.write(print_str) 

106 

107 def run(self): 

108 """health check main""" 

109 devices_num = self.device.get_device_count() 

110 if devices_num is None: 

111 return False 

112 

113 device_id = ParamDict().get_arg("device_id") 

114 if device_id is False: 

115 diagnose_devices = [i for i in range(devices_num)] 

116 else: 

117 diagnose_devices = [device_id] 

118 

119 # Example: ret = {device_id: [health, [[error_code, error_msg], [error_code, error_msg] ...]]} 

120 ret = self.run_health_check(diagnose_devices) 

121 

122 if ParamDict().get_command() == consts.health_cmd: 

123 self._print_screen(device_id, ret) 

124 else: 

125 # only collect or launch save file 

126 self._save_file(ret) 

127 

128 return True