Coverage for tests/routers/test_events_router.py: 100%

1import json 

2import pytest 

3from zenossapi.apiclient import ZenossAPIClientError 

4from zenossapi.routers.events import EventsRouter, ZenossEvent 

5import events_resp 

6 

7pytest_plugins = "pytest-responses" 

8url = 'https://zenoss/zport/dmd' 

9headers = dict( 

10 ContentType='application/json' 

11) 

12 

13 

14def request_callback(request): 

15 rdata = json.loads(request.body) 

16 resp_headers = dict(ContentType='application/json') 

17 

18 def query(rdata): 

19 if rdata['keys'] and rdata['keys'][0] == "evid": 

20 if 'params' in rdata and 'summary' in rdata['params'] and rdata['params']['summary'] == "Out of Tea": 

21 return events_resp.add_event_evid_query 

22 else: 

23 return events_resp.events_query_evid 

24 else: 

25 return events_resp.events_query 

26 

27 def detail(rdata): 

28 if rdata['evid'] == "02420a11-0015-98b9-11e7-9d96ae351999": 

29 return events_resp.event_detail 

30 elif rdata['evid'] == "02420a11-000c-a561-11e7-ba9b510182b3": 

31 return events_resp.add_event_detail 

32 else: 

33 return events_resp.fail 

34 

35 def getConfig(rdata): 

36 return events_resp.events_config 

37 

38 if rdata['method'] in ['close', 'acknowledge', 'reopen', 

39 'setConfigValues', 'add_event', 

40 'clear_heartbeats', 'clear_heartbeat', 

41 'write_log']: 

42 resp_body = events_resp.success 

43 else: 

44 method = locals()[rdata['method']] 

45 resp_body = method(rdata['data'][0]) 

46 

47 return (200, resp_headers, json.dumps(resp_body)) 

48 

49 

50def responses_callback(responses): 

51 responses.add_callback( 

52 responses.POST, 

53 '{0}/evconsole_router'.format(url), 

54 callback=request_callback, 

55 content_type='application/json', 

56 ) 

57 

58 

59class TestEventsRouter(object): 

60 

61 def test_events_router_init(self): 

62 er = EventsRouter(url, headers, True) 

63 assert er.evid is None 

64 assert isinstance(er.event_states_map, dict) 

65 assert er.event_states_map['Acknowledged'] == 1 

66 assert isinstance(er.event_severity_map, dict) 

67 assert er.event_severity_map['Critical'] == 5 

68 

69 def test_events_router_query_events(self, responses): 

70 responses_callback(responses) 

71 

72 er = EventsRouter(url, headers, True) 

73 resp = er.query_events(limit=2) 

74 assert resp['total'] == 50 

75 assert resp['ts'] == 1508797504.409547 

76 assert len(resp['events']) == 1 

77 assert resp['events'][0]['eventState'] == "Acknowledged" 

78 

79 def test_events_router_get_details_by_evid(self, responses): 

80 responses_callback(responses) 

81 

82 er = EventsRouter(url, headers, True) 

83 resp = er.get_details_by_evid("02420a11-0015-98b9-11e7-9d96ae351999") 

84 assert resp['agent'] == "zenpython" 

85 assert resp['device_uuid'] == "02e21618-b30a-47bf-8591-471c70570932" 

86 

87 def test_events_router_get_details_by_evid_not_found(self, responses): 

88 responses_callback(responses) 

89 

90 er = EventsRouter(url, headers, True) 

91 with pytest.raises(ZenossAPIClientError, match="Request failed: ServiceResponseError: Not Found"): 

92 resp = er.get_details_by_evid("02420a11-0015-98b9-11e7-9d96ae35199f") 

93 

94 def test_events_router_event_actions(self, responses): 

95 responses_callback(responses) 

96 

97 er = EventsRouter(url, headers, True) 

98 resp = er.event_actions('acknowledge', evids=["02420a11-0015-98b9-11e7-9d96ae351999"]) 

99 

100 def test_events_router_event_actions_invalid(self): 

101 er = EventsRouter(url, headers, True) 

102 with pytest.raises(ZenossAPIClientError, match="Unknown event action: terminate"): 

103 er.event_actions('terminate') 

104 

105 def test_events_router_get_config(self, responses): 

106 responses_callback(responses) 

107 

108 er = EventsRouter(url, headers, True) 

109 resp = er.get_config() 

110 assert len(resp) == 2 

111 assert resp[0]['id'] == "event_age_disable_severity" 

112 

113 def test_events_router_update_config(self, responses): 

114 responses_callback(responses) 

115 

116 er = EventsRouter(url, headers, True) 

117 resp = er.update_config(dict(event_age_disable_severity=3)) 

118 assert resp 

119 

120 def test_events_router_get_event_by_evid(self, responses): 

121 responses_callback(responses) 

122 

123 er = EventsRouter(url, headers, True) 

124 resp = er.get_event_by_evid("02420a11-0015-98b9-11e7-9d96ae351999") 

125 assert isinstance(resp, ZenossEvent) 

126 assert resp.evid == "02420a11-0015-98b9-11e7-9d96ae351999" 

127 

128 def test_events_router_list_open_events(self, responses): 

129 responses_callback(responses) 

130 

131 er = EventsRouter(url, headers, True) 

132 resp = er.list_open_events() 

133 assert len(resp['events']) == 1 

134 assert resp['total'] == 50 

135 assert resp['events'][0]['evid'] == "02420a11-0015-98b9-11e7-9d96ae351999" 

136 

137 def test_events_router_get_open_events(self, responses): 

138 responses_callback(responses) 

139 

140 er = EventsRouter(url, headers, True) 

141 resp = er.get_open_events() 

142 assert isinstance(resp[0], ZenossEvent) 

143 

144 def test_events_router_list_open_production_events(self, responses): 

145 responses_callback(responses) 

146 

147 er = EventsRouter(url, headers, True) 

148 resp = er.list_open_production_events() 

149 assert len(resp['events']) == 1 

150 assert resp['total'] == 50 

151 assert resp['events'][0]['evid'] == "02420a11-0015-98b9-11e7-9d96ae351999" 

152 

153 def test_events_router_get_open_production_events(self, responses): 

154 responses_callback(responses) 

155 

156 er = EventsRouter(url, headers, True) 

157 resp = er.get_open_production_events() 

158 assert isinstance(resp[0], ZenossEvent) 

159 

160 def test_events_router_add_event(self, responses): 

161 responses_callback(responses) 

162 

163 er = EventsRouter(url, headers, True) 

164 resp = er.add_event( 

165 "Out of Tea", 

166 "Heart of Gold", 

167 3, 

168 component="Arthur Dent" 

169 ) 

170 assert isinstance(resp, ZenossEvent) 

171 assert resp.evid == "02420a11-000c-a561-11e7-ba9b510182b3" 

172 

173 def test_events_router_clear_heartbeats(self, responses): 

174 responses_callback(responses) 

175 

176 er = EventsRouter(url, headers, True) 

177 resp = er.clear_heartbeats() 

178 assert resp 

179 

180 def test_events_router_clear_heartbeat(self, responses): 

181 responses_callback(responses) 

182 

183 er = EventsRouter(url, headers, True) 

184 resp = er.clear_heartbeat( 

185 collector='localhost', 

186 daemon='zenpython' 

187 ) 

188 assert resp 

189 

190 def test_events_router_zenossevent_update_log(self, responses): 

191 responses_callback(responses) 

192 

193 er = EventsRouter(url, headers, True) 

194 ev = er.get_event_by_evid("02420a11-000c-a561-11e7-ba9b510182b3") 

195 ev.log = [] 

196 ev.update_log('Test log entry') 

197 assert len(ev.log) == 1 

198 assert ev.log[0][0] == "zenoss" 

199 

200 def test_events_router_zenossevent_close(self, responses): 

201 responses_callback(responses) 

202 

203 er = EventsRouter(url, headers, True) 

204 ev = er.get_event_by_evid("02420a11-000c-a561-11e7-ba9b510182b3") 

205 resp = ev.close() 

206 assert resp 

207 

208 def test_events_router_zenossevent_ack(self, responses): 

209 responses_callback(responses) 

210 

211 er = EventsRouter(url, headers, True) 

212 ev = er.get_event_by_evid("02420a11-000c-a561-11e7-ba9b510182b3") 

213 resp = ev.ack() 

214 assert resp 

215 

216 def test_events_router_zenossevent_reopen(self, responses): 

217 responses_callback(responses) 

218 

219 er = EventsRouter(url, headers, True) 

220 ev = er.get_event_by_evid("02420a11-000c-a561-11e7-ba9b510182b3") 

221 resp = ev.reopen() 

222 assert resp