Coverage for src / server_list / spec / webapi / vm.py: 88%

1#!/usr/bin/env python3 

2""" 

3Web API for VM information. 

4Endpoint: /server-list/api/vm 

5""" 

6 

7import dataclasses 

8 

9import flask 

10 

11import server_list.spec.data_collector as data_collector 

12import server_list.spec.models as models 

13import server_list.spec.webapi as webapi 

14 

15vm_api = flask.Blueprint("vm_api", __name__) 

16 

17 

18def _vm_to_response(vm: models.VMInfo, host_reachable: bool) -> dict: 

19 """VMInfo をAPIレスポンス用dictに変換. 

20 

21 Args: 

22 vm: VMInfo dataclass 

23 host_reachable: ESXi ホストへの到達可否 

24 

25 Returns: 

26 API レスポンス用 dict (cached_power_state を含む) 

27 """ 

28 result = dataclasses.asdict(vm) 

29 result["cached_power_state"] = result.get("power_state") 

30 if not host_reachable: 

31 result["power_state"] = "unknown" 

32 return result 

33 

34 

35def apply_unknown_power_state_if_unreachable(vm_info: models.VMInfo) -> dict: 

36 """Apply 'unknown' power_state if the ESXi host is unreachable. 

37 

38 When ESXi host cannot be contacted, we use cached data but 

39 set power_state to 'unknown' to indicate the current state 

40 cannot be determined. The original cached value is preserved 

41 in cached_power_state for resource calculations. 

42 

43 Converts VMInfo to dict for JSON serialization. 

44 

45 Note: 呼び出し元で vm_info の None チェックを行うこと。 

46 """ 

47 host_reachable = data_collector.is_host_reachable(vm_info.esxi_host) 

48 return _vm_to_response(vm_info, host_reachable) 

49 

50 

51@vm_api.route("/vm/info", methods=["GET"]) 

52def get_vm_info_api(): 

53 """ 

54 Get VM information. 

55 

56 Query parameters: 

57 vm_name: VM name to look up (required) 

58 esxi_host: ESXi host (optional, for disambiguation) 

59 

60 Returns: 

61 JSON with VM info (cpu_count, ram_mb, storage_gb, power_state, collected_at) 

62 """ 

63 vm_name = flask.request.args.get("vm_name") 

64 

65 if not vm_name: 

66 return webapi.error_response("vm_name is required", 400) 

67 

68 esxi_host = flask.request.args.get("esxi_host") 

69 

70 result = data_collector.get_vm_info(vm_name, esxi_host) 

71 

72 if result: 

73 # Apply unknown power_state if host is unreachable 

74 result_dict = apply_unknown_power_state_if_unreachable(result) 

75 return webapi.success_response(result_dict) 

76 

77 return webapi.error_response(f"VM not found: {vm_name}") 

78 

79 

80@vm_api.route("/vm/info/batch", methods=["POST"]) 

81def get_vm_info_batch(): 

82 """ 

83 Get VM information for multiple VMs. 

84 

85 Request body (JSON): 

86 vms: List of VM names to look up 

87 esxi_host: ESXi host (optional) 

88 

89 Returns: 

90 JSON with results for each VM 

91 """ 

92 data = flask.request.get_json() 

93 

94 if not data or "vms" not in data: 

95 return webapi.error_response("VM list is required", 400) 

96 

97 vm_list = data["vms"] 

98 esxi_host = data.get("esxi_host") 

99 results = {} 

100 

101 for vm_name in vm_list: 

102 result = data_collector.get_vm_info(vm_name, esxi_host) 

103 if result: 103 ↛ 111line 103 didn't jump to line 111 because the condition on line 103 was always true

104 # Apply unknown power_state if host is unreachable 

105 result_dict = apply_unknown_power_state_if_unreachable(result) 

106 results[vm_name] = { 

107 "success": True, 

108 "data": result_dict 

109 } 

110 else: 

111 results[vm_name] = { 

112 "success": False, 

113 "data": None 

114 } 

115 

116 return flask.jsonify({ 

117 "success": True, 

118 "results": results 

119 }) 

120 

121 

122@vm_api.route("/vm/host/<esxi_host>", methods=["GET"]) 

123def get_vms_for_host(esxi_host: str): 

124 """ 

125 Get all VMs for a specific ESXi host. 

126 

127 Returns: 

128 JSON with list of VMs and their info 

129 """ 

130 vms = data_collector.get_all_vm_info_for_host(esxi_host) 

131 host_reachable = data_collector.is_host_reachable(esxi_host) 

132 

133 return webapi.success_response({ 

134 "esxi_host": esxi_host, 

135 "vms": [_vm_to_response(vm, host_reachable) for vm in vms], 

136 }) 

137 

138 

139@vm_api.route("/vm/refresh/<esxi_host>", methods=["POST"]) 

140def refresh_host_data(esxi_host: str): 

141 """ 

142 Trigger immediate data collection from an ESXi host. 

143 

144 This endpoint triggers the data collector to fetch fresh data 

145 from the specified ESXi host. After collection is complete, 

146 an SSE event is sent to notify connected clients. 

147 

148 Returns: 

149 JSON with success status 

150 """ 

151 success = data_collector.collect_host_data(esxi_host) 

152 

153 if success: 

154 return flask.jsonify({ 

155 "success": True, 

156 "message": f"Data collection completed for {esxi_host}", 

157 }) 

158 

159 return webapi.error_response(f"Failed to collect data from {esxi_host}", 500)