Coverage for tests/routers/test_template_router.py: 97%

1import json 

2import pytest 

3from zenossapi.apiclient import ZenossAPIClientError 

4from zenossapi.routers.template import TemplateRouter, ZenossTemplate, ZenossDataSource, ZenossDataPoint, ZenossThreshold, ZenossGraph 

5import template_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 getDeviceClassTemplates(rdata): 

19 return template_resp.list_templates 

20 

21 def getInfo(rdata): 

22 t = rdata['uid'].split('/')[-1] 

23 return template_resp.templates[t] 

24 

25 def getObjTemplates(rdata): 

26 return template_resp.obj_templates 

27 

28 def addTemplate(rdata): 

29 return dict( 

30 result=dict( 

31 nodeConfig=dict( 

32 uid='/zport/dmd/Devices/Server/TEST/rrdTemplates/TestAdd' 

33 ) 

34 ) 

35 ) 

36 

37 def addLocalRRDTemplate(rdata): 

38 return dict( 

39 result=dict( 

40 nodeConfig=dict( 

41 uid='/zport/dmd/Devices/Server/TEST/test.example.com/DnsMonitor' 

42 ) 

43 ) 

44 ) 

45 

46 def getDataSourceTypes(rdata): 

47 return template_resp.ds_types 

48 

49 def getThresholdTypes(rdata): 

50 return template_resp.threshold_types 

51 

52 def copyTemplate(rdata): 

53 return template_resp.templates['LogicalVolume'] 

54 

55 def getDataSources(rdata): 

56 return template_resp.lv_data_sources 

57 

58 def getDataSourceDetails(rdata): 

59 if rdata['uid'].endswith('status'): 

60 return template_resp.lv_ds_details 

61 elif rdata['uid'].endswith('disk'): 

62 return template_resp.fs_ds_details 

63 else: 

64 return template_resp.fail 

65 

66 def getDataPoints(rdata): 

67 return template_resp.lv_data_points 

68 

69 def getThresholds(rdata): 

70 return template_resp.fs_thresholds 

71 

72 def getThresholdDetails(rdata): 

73 return template_resp.threshold_details 

74 

75 def getGraphs(rdata): 

76 return template_resp.graphs_list 

77 

78 def getGraphDefinition(rdata): 

79 return template_resp.graph_def 

80 

81 def getDataPointDetails(rdata): 

82 if rdata['uid'].endswith('state'): 

83 return template_resp.lv_dp_details 

84 elif rdata['uid'].endswith('disk_percentUsed'): 

85 return template_resp.fs_dp_details 

86 elif rdata['uid'].endswith('disk_usedBlocks'): 

87 return template_resp.fs_dp_details_2 

88 else: 

89 return template_resp.no_points 

90 

91 def getGraphPoints(rdata): 

92 return template_resp.graph_points 

93 

94 if rdata['method'] in ['deleteTemplate', 'removeLocalRRDTemplate', 

95 'addDataPointToGraph', 'addDataSource', 

96 'deleteDataSource', 'addThreshold', 

97 'removeThreshold', 'addGraphDefinition', 

98 'deleteGraphDefinition', 'addDataPoint', 

99 'deleteDataPoint', 'addThreshold', 

100 'removeThreshold', 'addThresholdToGraph', 'setInfo', 

101 'setGraphDefinition', 'setGraphPointSequence', 

102 'deleteGraphPoint']: 

103 resp_body = template_resp.success 

104 else: 

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

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

107 

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

109 

110 

111def responses_callback(responses): 

112 responses.add_callback( 

113 responses.POST, 

114 '{0}/template_router'.format(url), 

115 callback=request_callback, 

116 content_type='application/json', 

117 ) 

118 

119 

120class TestTemplateRouter(object): 

121 

122 def test_template_router_init(self): 

123 tr = TemplateRouter(url, headers, True) 

124 assert tr.uid is None 

125 assert tr.properties is None 

126 

127 def test_template_router_list_templates(self, responses): 

128 responses_callback(responses) 

129 

130 tr = TemplateRouter(url, headers, True) 

131 resp = tr.list_device_class_templates('Server/TEST') 

132 assert len(resp) == 12 

133 assert resp == template_resp.list_templates_match 

134 

135 def test_template_router_get_templates(self, responses): 

136 responses_callback(responses) 

137 

138 tr = TemplateRouter(url, headers, True) 

139 resp = tr.get_device_class_templates('Server/TEST') 

140 assert len(resp) == 12 

141 assert isinstance(resp[0], ZenossTemplate) 

142 

143 def test_template_router_get_object_templates(self, responses): 

144 responses_callback(responses) 

145 

146 tr = TemplateRouter(url, headers, True) 

147 resp = tr.get_object_templates('Server/TEST/test.example.com') 

148 assert len(resp) == 2 

149 assert isinstance(resp[0], ZenossTemplate) 

150 assert isinstance(resp[1], ZenossTemplate) 

151 

152 def test_template_router_get_template(self, responses): 

153 responses_callback(responses) 

154 

155 tr = TemplateRouter(url, headers, True) 

156 resp = tr.get_template('Devices', 'Device') 

157 assert isinstance(resp, ZenossTemplate) 

158 assert resp.uid == "Devices/Server/TEST/rrdTemplates/Device" 

159 

160 def test_template_router_add_template(self, responses): 

161 responses_callback(responses) 

162 

163 tr = TemplateRouter(url, headers, True) 

164 resp = tr.add_template('Servers/TEST', 'TestAdd') 

165 assert isinstance(resp, ZenossTemplate) 

166 assert resp.uid == "Devices/Server/TEST/rrdTemplates/TestAdd" 

167 

168 def test_template_router_add_local_template(self, responses): 

169 responses_callback(responses) 

170 

171 tr = TemplateRouter(url, headers, True) 

172 resp = tr.add_local_template('Servers/TEST/devices/test.example.com', 'DnsMonitor') 

173 assert isinstance(resp, ZenossTemplate) 

174 assert resp.uid == "Devices/Server/TEST/devices/test.example.com/DnsMonitor" 

175 

176 def test_template_router_delete_template(self, responses): 

177 responses_callback(responses) 

178 

179 tr = TemplateRouter(url, headers, True) 

180 resp = tr.delete_template('Servers/TEST', 'TestAdd') 

181 assert resp['success'] 

182 

183 def test_template_router_delete_local_template(self, responses): 

184 responses_callback(responses) 

185 

186 tr = TemplateRouter(url, headers, True) 

187 resp = tr.delete_local_template('Server/TEST/devices/test.example.com', 'DnsMonitor') 

188 assert resp['success'] 

189 

190 def test_template_router_get_data_source_types(self, responses): 

191 responses_callback(responses) 

192 

193 tr = TemplateRouter(url, headers, True) 

194 resp = tr.get_data_source_types() 

195 assert len(resp) == 18 

196 assert "HttpMonitor" in resp 

197 

198 def test_template_router_get_threshold_types(self, responses): 

199 responses_callback(responses) 

200 

201 tr = TemplateRouter(url, headers, True) 

202 resp = tr.get_threshold_types() 

203 assert len(resp) == 4 

204 assert "MinMaxThreshold" in resp 

205 

206 def test_template_router_add_data_point_to_graph(self, responses): 

207 responses_callback(responses) 

208 

209 tr = TemplateRouter(url, headers, True) 

210 resp = tr.add_data_point_to_graph('Data/Point', 'Graph/UID', True) 

211 assert resp['success'] 

212 

213 def test_template_router_zenosstemplate_copy(self, responses): 

214 responses_callback(responses) 

215 

216 tr = TemplateRouter(url, headers, True) 

217 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

218 resp = zt.copy('Server') 

219 assert isinstance(resp, ZenossTemplate) 

220 

221 def test_template_router_zenosstemplate_list_data_sources(self, responses): 

222 responses_callback(responses) 

223 

224 tr = TemplateRouter(url, headers, True) 

225 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

226 resp = zt.list_data_sources() 

227 assert len(resp) == 1 

228 assert resp[0] == "Devices/Server/TEST/rrdTemplates/LogicalVolume/datasources/status" 

229 

230 def test_template_router_zenosstemplate_get_data_sources(self, responses): 

231 responses_callback(responses) 

232 

233 tr = TemplateRouter(url, headers, True) 

234 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

235 resp = zt.get_data_sources() 

236 assert len(resp) == 1 

237 assert isinstance(resp[0], ZenossDataSource) 

238 

239 def test_template_router_zenosstemplate_get_data_source(self, responses): 

240 responses_callback(responses) 

241 

242 tr = TemplateRouter(url, headers, True) 

243 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

244 resp = zt.get_data_source('status') 

245 assert isinstance(resp, ZenossDataSource) 

246 assert resp.uid == "Devices/Server/TEST/rrdTemplates/LogicalVolume/datasources/status" 

247 

248 def test_template_router_zenosstemplate_add_data_source(self, responses): 

249 responses_callback(responses) 

250 

251 tr = TemplateRouter(url, headers, True) 

252 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

253 resp = zt.add_data_source('status', 'COMMAND') 

254 assert isinstance(resp, ZenossDataSource) 

255 assert resp.uid == "Devices/Server/TEST/rrdTemplates/LogicalVolume/datasources/status" 

256 

257 def test_template_router_zenosstemplate_delete_data_source(self, responses): 

258 responses_callback(responses) 

259 

260 tr = TemplateRouter(url, headers, True) 

261 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

262 resp = zt.delete_data_source('status') 

263 assert resp['success'] 

264 

265 def test_temmplate_router_zenosstemplate_get_data_source_not_found(self, responses): 

266 responses_callback(responses) 

267 

268 tr = TemplateRouter(url, headers, True) 

269 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

270 with pytest.raises(ZenossAPIClientError, match="Request failed: ObjectNotFoundException"): 

271 resp = zt.get_data_source('bad') 

272 

273 def test_template_router_zenosstemplate_list_data_points(self, responses): 

274 responses_callback(responses) 

275 

276 tr = TemplateRouter(url, headers, True) 

277 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

278 resp = zt.list_data_points() 

279 assert len(resp) == 2 

280 assert resp[0] == "Devices/Server/TEST/rrdTemplates/LogicalVolume/datasources/status/datapoints/state" 

281 

282 def test_template_router_zenosstemplate_get_data_points(self, responses): 

283 responses_callback(responses) 

284 

285 tr = TemplateRouter(url, headers, True) 

286 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

287 resp = zt.get_data_points() 

288 assert len(resp) == 2 

289 assert isinstance(resp[0], ZenossDataPoint) 

290 

291 def test_template_router_zenosstemplate_list_thresholds(self, responses): 

292 responses_callback(responses) 

293 

294 tr = TemplateRouter(url, headers, True) 

295 zt = tr.get_template('Server/TEST', 'FileSystem') 

296 resp = zt.list_thresholds() 

297 assert len(resp) == 2 

298 assert resp[0] == "Devices/Server/TEST/rrdTemplates/FileSystem/thresholds/90 percent used" 

299 

300 def test_template_router_zenosstemplate_get_thresholds(self, responses): 

301 responses_callback(responses) 

302 

303 tr = TemplateRouter(url, headers, True) 

304 zt = tr.get_template('Server/TEST', 'FileSystem') 

305 resp = zt.get_thresholds() 

306 assert len(resp) == 2 

307 assert isinstance(resp[0], ZenossThreshold) 

308 

309 def test_template_router_zenosstemplate_get_threshold(self, responses): 

310 responses_callback(responses) 

311 

312 tr = TemplateRouter(url, headers, True) 

313 zt = tr.get_template('Server/TEST', 'FileSystem') 

314 resp = zt.get_threshold('90 percent used') 

315 assert isinstance(resp, ZenossThreshold) 

316 assert resp.uid == "Devices/Server/TEST/rrdTemplates/FileSystem/thresholds/90 percent used" 

317 

318 def test_template_router_zenosstemplate_add_threshold(self, responses): 

319 responses_callback(responses) 

320 

321 tr = TemplateRouter(url, headers, True) 

322 zt = tr.get_template('Server/TEST', 'FileSystem') 

323 resp = zt.add_threshold('90 percent used', 'MinMaxThreshold', ['dp']) 

324 assert isinstance(resp, ZenossThreshold) 

325 assert resp.uid == "Devices/Server/TEST/rrdTemplates/FileSystem/thresholds/90 percent used" 

326 

327 def test_template_router_zenosstemplate_delete_threshold(self, responses): 

328 responses_callback(responses) 

329 

330 tr = TemplateRouter(url, headers, True) 

331 zt = tr.get_template('Server/TEST', 'FileSystem') 

332 resp = zt.delete_threshold('90 percent used') 

333 assert resp['success'] 

334 

335 def test_template_router_zenosstemplate_list_graphs(self, responses): 

336 responses_callback(responses) 

337 

338 tr = TemplateRouter(url, headers, True) 

339 zt = tr.get_template('Server/TEST', 'FileSystem') 

340 resp = zt.list_graphs() 

341 assert len(resp) == 1 

342 assert resp[0] == "Devices/Server/TEST/rrdTemplates/FileSystem/graphDefs/Usage" 

343 

344 def test_template_router_zenosstemplate_get_graphs(self, responses): 

345 responses_callback(responses) 

346 

347 tr = TemplateRouter(url, headers, True) 

348 zt = tr.get_template('Server/TEST', 'FileSystem') 

349 resp = zt.get_graphs() 

350 assert len(resp) == 1 

351 assert isinstance(resp[0], ZenossGraph) 

352 

353 def test_template_router_zenosstemplate_get_graph(self, responses): 

354 responses_callback(responses) 

355 

356 tr = TemplateRouter(url, headers, True) 

357 zt = tr.get_template('Server/TEST', 'FileSystem') 

358 resp = zt.get_graph('Usage') 

359 assert isinstance(resp, ZenossGraph) 

360 assert resp.uid == "Devices/Server/TEST/rrdTemplates/FileSystem/graphDefs/Usage" 

361 

362 def test_template_router_zenosstemplate_add_graph(self, responses): 

363 responses_callback(responses) 

364 

365 tr = TemplateRouter(url, headers, True) 

366 zt = tr.get_template('Server/TEST', 'FileSystem') 

367 resp = zt.add_graph('Usage') 

368 assert isinstance(resp, ZenossGraph) 

369 assert resp.uid == "Devices/Server/TEST/rrdTemplates/FileSystem/graphDefs/Usage" 

370 

371 def test_template_router_zenosstemplate_delete_graph(self, responses): 

372 responses_callback(responses) 

373 

374 tr = TemplateRouter(url, headers, True) 

375 zt = tr.get_template('Server/TEST', 'FileSystem') 

376 resp = zt.delete_graph('Usage') 

377 assert resp['success'] 

378 

379 def test_template_router_zenossdatasource_list_data_points(self, responses): 

380 responses_callback(responses) 

381 

382 tr = TemplateRouter(url, headers, True) 

383 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

384 ds = zt.get_data_source('status') 

385 resp = ds.list_data_points() 

386 assert len(resp) == 2 

387 assert resp[0] == "Devices/Server/TEST/rrdTemplates/LogicalVolume/datasources/status/datapoints/state" 

388 

389 def test_template_router_zenossdatasource_get_data_points(self, responses): 

390 responses_callback(responses) 

391 

392 tr = TemplateRouter(url, headers, True) 

393 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

394 ds = zt.get_data_source('status') 

395 resp = ds.get_data_points() 

396 assert len(resp) == 2 

397 assert isinstance(resp[0], ZenossDataPoint) 

398 

399 def test_template_router_zenossdatasource_get_data_point(self, responses): 

400 responses_callback(responses) 

401 

402 tr = TemplateRouter(url, headers, True) 

403 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

404 ds = zt.get_data_source('status') 

405 resp = ds.get_data_point('state') 

406 assert isinstance(resp, ZenossDataPoint) 

407 assert resp.uid == "Devices/Server/TEST/rrdTemplates/LogicalVolume/datasources/status/datapoints/state" 

408 

409 def test_template_router_zenossdatasource_add_data_point(self, responses): 

410 responses_callback(responses) 

411 

412 tr = TemplateRouter(url, headers, True) 

413 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

414 ds = zt.get_data_source('status') 

415 resp = ds.get_data_point('state') 

416 assert isinstance(resp, ZenossDataPoint) 

417 assert resp.uid == "Devices/Server/TEST/rrdTemplates/LogicalVolume/datasources/status/datapoints/state" 

418 

419 def test_template_router_zenossdatasource_delete_data_point(self, responses): 

420 responses_callback(responses) 

421 

422 tr = TemplateRouter(url, headers, True) 

423 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

424 ds = zt.get_data_source('status') 

425 resp = ds.delete_data_point('state') 

426 assert resp['success'] 

427 

428 def test_template_router_zenossdatasource_delete(self, responses): 

429 responses_callback(responses) 

430 

431 tr = TemplateRouter(url, headers, True) 

432 zt = tr.get_template('Server/TEST', 'LogicalVolume') 

433 ds = zt.get_data_source('status') 

434 resp = ds.delete() 

435 assert resp['success'] 

436 

437 def test_template_router_zenossdatapoint_set_threshold(self, responses): 

438 responses_callback(responses) 

439 

440 tr = TemplateRouter(url, headers, True) 

441 zt = tr.get_template('Server/TEST', 'FileSystem') 

442 ds = zt.get_data_source('disk') 

443 dp = ds.get_data_point('disk_percentUsed') 

444 resp = dp.set_threshold('90 percent used', 'MinMaxThreshold') 

445 assert isinstance(resp, ZenossThreshold) 

446 assert resp.uid == "Devices/Server/TEST/rrdTemplates/FileSystem/thresholds/90 percent used" 

447 

448 def tet_template_router_zenossdatapoint_make_counter(self, responses): 

449 responses_callback(responses) 

450 

451 tr = TemplateRouter(url, headers, True) 

452 zt = tr.get_template('Server/TEST', 'FileSystem') 

453 ds = zt.get_data_source('disk') 

454 dp = ds.get_data_point('disk_percentUsed') 

455 dp.make_counter() 

456 

457 def tet_template_router_zenossdatapoint_make_gauge(self, responses): 

458 responses_callback(responses) 

459 

460 tr = TemplateRouter(url, headers, True) 

461 zt = tr.get_template('Server/TEST', 'FileSystem') 

462 ds = zt.get_data_source('disk') 

463 dp = ds.get_data_point('disk_percentUsed') 

464 dp.make_gauge() 

465 

466 def test_template_router_zenossdatapoint_delete(self, responses): 

467 responses_callback(responses) 

468 

469 tr = TemplateRouter(url, headers, True) 

470 zt = tr.get_template('Server/TEST', 'FileSystem') 

471 ds = zt.get_data_source('disk') 

472 dp = ds.get_data_point('disk_percentUsed') 

473 resp = dp.delete() 

474 assert resp['success'] 

475 

476 def test_template_router_zenossthreshold_set_max(self, responses): 

477 responses_callback(responses) 

478 

479 tr = TemplateRouter(url, headers, True) 

480 zt = tr.get_template('Server/TEST', 'FileSystem') 

481 threshold = zt.get_threshold('90 percent used') 

482 threshold.set_max(100) 

483 

484 def test_template_router_zenossthreshold_set_min(self, responses): 

485 responses_callback(responses) 

486 

487 tr = TemplateRouter(url, headers, True) 

488 zt = tr.get_template('Server/TEST', 'FileSystem') 

489 threshold = zt.get_threshold('90 percent used') 

490 threshold.set_min(1) 

491 

492 def test_template_router_zenossthreshold_delete(self, responses): 

493 responses_callback(responses) 

494 

495 tr = TemplateRouter(url, headers, True) 

496 zt = tr.get_template('Server/TEST', 'FileSystem') 

497 threshold = zt.get_threshold('90 percent used') 

498 resp = threshold.delete() 

499 assert resp['success'] 

500 

501 def test_template_router_zenossgraph_delete(self, responses): 

502 responses_callback(responses) 

503 

504 tr = TemplateRouter(url, headers, True) 

505 zt = tr.get_template('Server/TEST', 'FileSystem') 

506 graph = zt.get_graph('Usage') 

507 resp = graph.delete() 

508 assert resp['success'] 

509 

510 def test_template_router_zenossgraph_list_points(self, responses): 

511 responses_callback(responses) 

512 

513 tr = TemplateRouter(url, headers, True) 

514 zt = tr.get_template('Server/TEST', 'FileSystem') 

515 graph = zt.get_graph('Usage') 

516 resp = graph.list_points() 

517 assert len(resp) == 1 

518 assert resp[0] == "Devices/Server/TEST/rrdTemplates/FileSystem/graphDefs/Usage/graphPoints/Used" 

519 

520 def test_template_router_zenossgraph_get_points(self, responses): 

521 responses_callback(responses) 

522 

523 tr = TemplateRouter(url, headers, True) 

524 zt = tr.get_template('Server/TEST', 'FileSystem') 

525 graph = zt.get_graph('Usage') 

526 resp = graph.get_points() 

527 assert len(resp) == 1 

528 assert isinstance(resp[0], ZenossDataPoint) 

529 

530 def test_template_router_zenossgraph_add_point(self, responses): 

531 responses_callback(responses) 

532 

533 tr = TemplateRouter(url, headers, True) 

534 zt = tr.get_template('Server/TEST', 'FileSystem') 

535 graph = zt.get_graph('Usage') 

536 resp = graph.add_point('disk', 'usedBlocks') 

537 assert resp['success'] 

538 

539 def test_template_router_zenossgraph_set_graph_properties(self, responses): 

540 responses_callback(responses) 

541 

542 tr = TemplateRouter(url, headers, True) 

543 zt = tr.get_template('Server/TEST', 'FileSystem') 

544 graph = zt.get_graph('Usage') 

545 resp = graph.set_graph_properties(dict(foo='bar')) 

546 assert resp['success'] 

547 

548 def test_template_router_zenossgraph_set_point_sequence(self, responses): 

549 responses_callback(responses) 

550 

551 tr = TemplateRouter(url, headers, True) 

552 zt = tr.get_template('Server/TEST', 'FileSystem') 

553 graph = zt.get_graph('Usage') 

554 resp = graph.set_point_sequence(['Used']) 

555 assert resp['success'] 

556 

557 def test_template_router_zenossgraph_delete_point(self, responses): 

558 responses_callback(responses) 

559 

560 tr = TemplateRouter(url, headers, True) 

561 zt = tr.get_template('Server/TEST', 'FileSystem') 

562 graph = zt.get_graph('Usage') 

563 resp = graph.delete_point('Used') 

564 assert resp['success'] 

565 

566 def test_template_router_zenossgraph_add_graph_threshold(self, responses): 

567 responses_callback(responses) 

568 

569 tr = TemplateRouter(url, headers, True) 

570 zt = tr.get_template('Server/TEST', 'FileSystem') 

571 graph = zt.get_graph('Usage') 

572 resp = graph.add_graph_threshold('Devices/Server/TEST/rrdTemplates/FileSystem/thresholds/90 percent used') 

573 assert resp['success']