Coverage for osm_nbi/tests/test_instance_topics.py: 99%

1# 

2# Licensed under the Apache License, Version 2.0 (the "License"); you may 

3# not use this file except in compliance with the License. You may obtain 

4# a copy of the License at 

5# 

6# http://www.apache.org/licenses/LICENSE-2.0 

7# 

8# Unless required by applicable law or agreed to in writing, software 

9# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

10# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

11# License for the specific language governing permissions and limitations 

12# under the License. 

13# 

14# For those usages not covered by the Apache License, Version 2.0 please 

15# contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com 

16## 

17 

18from contextlib import contextmanager 

19import unittest 

20from time import time 

21from unittest.mock import Mock, mock_open # patch, MagicMock 

22from osm_common.dbbase import DbException 

23from osm_nbi.engine import EngineException 

24from osm_common.dbmemory import DbMemory 

25from osm_common.fsbase import FsBase 

26from osm_common.msgbase import MsgBase 

27from osm_common import dbbase 

28from http import HTTPStatus 

29from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic 

30from osm_nbi.tests.test_db_descriptors import ( 

31 db_vim_accounts_text, 

32 db_nsds_text, 

33 db_vnfds_text, 

34 db_nsrs_text, 

35 db_vnfrs_text, 

36) 

37from copy import deepcopy 

38import yaml 

39 

40 

41class TestNsLcmOpTopic(unittest.TestCase): 

42 def setUp(self): 

43 self.db = DbMemory() 

44 self.fs = Mock(FsBase()) 

45 self.fs.get_params.return_value = {"./fake/folder"} 

46 self.fs.file_open = mock_open() 

47 self.msg = Mock(MsgBase()) 

48 # create class 

49 self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) 

50 self.nslcmop_topic.check_quota = Mock(return_value=None) # skip quota 

51 

52 self.db.create_list("vim_accounts", yaml.safe_load(db_vim_accounts_text)) 

53 self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) 

54 self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) 

55 self.db.create_list("vnfrs", yaml.safe_load(db_vnfrs_text)) 

56 self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) 

57 self.db.create = Mock(return_value="created_id") 

58 self.nsd = self.db.get_list("nsds")[0] 

59 self.nsd_id = self.nsd["_id"] 

60 self.nsr = self.db.get_list("nsrs")[0] 

61 self.nsr_id = self.nsr["_id"] 

62 self.nsr_project = self.nsr["_admin"]["projects_read"][0] 

63 

64 self.vim = self.db.get_list("vim_accounts")[0] 

65 self.vim_id = self.vim["_id"] 

66 

67 def test_create_instantiate(self): 

68 self.db.set_one = Mock(return_value={"updated": 1}) 

69 session = { 

70 "force": False, 

71 "admin": False, 

72 "public": False, 

73 "project_id": [self.nsr_project], 

74 "method": "write", 

75 } 

76 indata = { 

77 "nsdId": self.nsd_id, 

78 "nsInstanceId": self.nsr_id, 

79 "nsName": "name", 

80 "vimAccountId": self.vim_id, 

81 "additionalParamsForVnf": [ 

82 { 

83 "member-vnf-index": "1", 

84 "additionalParams": {"touch_filename": "file"}, 

85 }, 

86 { 

87 "member-vnf-index": "2", 

88 "additionalParams": {"touch_filename": "file"}, 

89 }, 

90 ], 

91 "vnf": [ 

92 { 

93 "member-vnf-index": "1", 

94 "vdu": [ 

95 { 

96 "id": "dataVM", 

97 "interface": [ 

98 { 

99 "name": "dataVM-eth0", 

100 "ip-address": "10.11.12.13", 

101 "floating-ip-required": True, 

102 } 

103 ], 

104 } 

105 ], 

106 "internal-vld": [ 

107 {"name": "internal", "vim-network-id": "vim-net-id"} 

108 ], 

109 } 

110 ], 

111 "lcmOperationType": "instantiate", 

112 } 

113 rollback = [] 

114 headers = {} 

115 

116 nslcmop_id, _ = self.nslcmop_topic.new( 

117 rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers 

118 ) 

119 

120 # check nslcmop is created at database 

121 self.assertEqual( 

122 self.db.create.call_count, 

123 1, 

124 "database create not called, or called more than once", 

125 ) 

126 _call = self.db.create.call_args_list[0] 

127 self.assertEqual( 

128 _call[0][0], "nslcmops", "must be create a nslcmops entry at database" 

129 ) 

130 

131 created_nslcmop = _call[0][1] 

132 self.assertEqual( 

133 nslcmop_id, 

134 created_nslcmop["_id"], 

135 "mismatch between return id and database '_id'", 

136 ) 

137 self.assertEqual( 

138 self.nsr_id, 

139 created_nslcmop["nsInstanceId"], 

140 "bad reference id from nslcmop to nsr", 

141 ) 

142 self.assertTrue( 

143 created_nslcmop["_admin"].get("projects_read"), 

144 "Database record must contain '_amdin.projects_read'", 

145 ) 

146 self.assertIn( 

147 "created", 

148 created_nslcmop["_admin"], 

149 "Database record must contain '_admin.created'", 

150 ) 

151 self.assertTrue( 

152 created_nslcmop["lcmOperationType"] == "instantiate", 

153 "Database record must contain 'lcmOperationType=instantiate'", 

154 ) 

155 

156 self.assertEqual( 

157 len(rollback), 

158 len(self.db.set_one.call_args_list) + 1, 

159 "rollback mismatch with created/set items at database", 

160 ) 

161 

162 # test parameters with error 

163 bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa" 

164 test_set = ( 

165 ( 

166 "nsr not found", 

167 {"nsInstanceId": bad_id}, 

168 DbException, 

169 HTTPStatus.NOT_FOUND, 

170 ("not found", bad_id), 

171 ), 

172 # TODO add "nsd" 

173 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim" 

174 ( 

175 "bad member-vnf-index", 

176 {"vnf.0.member-vnf-index": "k"}, 

177 EngineException, 

178 HTTPStatus.BAD_REQUEST, 

179 ("k",), 

180 ), 

181 ) 

182 for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set: 

183 with self.assertRaises(expect_exc, msg=message) as e: 

184 self.nslcmop_topic.new( 

185 rollback, 

186 session, 

187 indata=deepcopy(indata), 

188 kwargs=kwargs_, 

189 headers=headers, 

190 ) 

191 if expect_code: 

192 self.assertTrue(e.exception.http_code == expect_code) 

193 if expect_text_list: 

194 for expect_text in expect_text_list: 

195 self.assertIn( 

196 expect_text, 

197 str(e.exception).lower(), 

198 "Expected '{}' at exception text".format(expect_text), 

199 ) 

200 

201 def test_check_ns_operation_action(self): 

202 nsrs = self.db.get_list("nsrs")[0] 

203 session = {} 

204 

205 indata = { 

206 "member_vnf_index": "1", 

207 "vdu_id": None, 

208 "primitive": "touch", 

209 "primitive_params": {"filename": "file"}, 

210 } 

211 

212 self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata) 

213 for k in indata: 

214 indata_copy = indata.copy() 

215 if k == "primitive_params": 

216 continue 

217 indata_copy[k] = "non_existing" 

218 with self.assertRaises(EngineException) as exc_manager: 

219 self.nslcmop_topic._check_ns_operation( 

220 session, nsrs, "action", indata_copy 

221 ) 

222 exc = exc_manager.exception 

223 self.assertEqual( 

224 exc.http_code, 

225 HTTPStatus.BAD_REQUEST, 

226 "Engine exception bad http_code with {}".format(indata_copy), 

227 ) 

228 

229 def test_update_remove_vnf(self): 

230 vnfr_id = self.db.get_list("vnfrs")[0]["_id"] 

231 session = {} 

232 self.db.set_one( 

233 "nsrs", 

234 {"_id": self.nsr_id}, 

235 {"_admin.nsState": "INSTANTIATED"}, 

236 ) 

237 indata = { 

238 "lcmOperationType": "update", 

239 "updateType": "REMOVE_VNF", 

240 "nsInstanceId": self.nsr_id, 

241 "removeVnfInstanceId": vnfr_id, 

242 } 

243 

244 session = { 

245 "force": False, 

246 "admin": False, 

247 "public": False, 

248 "project_id": [self.nsr_project], 

249 "method": "write", 

250 } 

251 rollback = [] 

252 headers = {} 

253 

254 nslcmop_id, _ = self.nslcmop_topic.new( 

255 rollback, session, indata, kwargs=None, headers=headers 

256 ) 

257 

258 self.assertEqual( 

259 self.db.create.call_count, 

260 1, 

261 "database create not called, or called more than once", 

262 ) 

263 _call = self.db.create.call_args_list[0] 

264 self.assertEqual( 

265 _call[0][0], "nslcmops", "nslcmops entry must be created at database" 

266 ) 

267 created_nslcmop = _call[0][1] 

268 self.assertEqual( 

269 self.nsr_id, 

270 created_nslcmop["nsInstanceId"], 

271 "mismatch between nsId '_id' in created nslcmop and database nsr", 

272 ) 

273 self.assertTrue( 

274 created_nslcmop["lcmOperationType"] == "update", 

275 "Database record must contain 'lcmOperationType=update'", 

276 ) 

277 self.assertTrue( 

278 created_nslcmop["operationParams"]["updateType"] == "REMOVE_VNF", 

279 "Database record must contain 'updateType=REMOVE_VNF'", 

280 ) 

281 

282 def test_migrate(self): 

283 _ = self.db.get_list("vnfrs")[0]["_id"] 

284 session = {} 

285 self.db.set_one( 

286 "nsrs", 

287 {"_id": self.nsr_id}, 

288 {"_admin.nsState": "INSTANTIATED"}, 

289 ) 

290 session = { 

291 "force": False, 

292 "admin": False, 

293 "public": False, 

294 "project_id": [self.nsr_project], 

295 "method": "write", 

296 } 

297 rollback = [] 

298 headers = {} 

299 

300 with self.subTest(i=1, t="Migration for Specific VM"): 

301 indata = { 

302 "lcmOperationType": "migrate", 

303 "nsInstanceId": self.nsr_id, 

304 "migrateToHost": "sample02", 

305 "vdu": {"vduCountIndex": 0, "vduId": "mgmtVM"}, 

306 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f", 

307 } 

308 nslcmop_id, _ = self.nslcmop_topic.new( 

309 rollback, session, indata, kwargs=None, headers=headers 

310 ) 

311 

312 self.assertEqual( 

313 self.db.create.call_count, 

314 1, 

315 "database create not called, or called more than once", 

316 ) 

317 _call = self.db.create.call_args_list[0] 

318 self.assertEqual( 

319 _call[0][0], "nslcmops", "nslcmops entry must be created at database" 

320 ) 

321 created_nslcmop = _call[0][1] 

322 self.assertEqual( 

323 self.nsr_id, 

324 created_nslcmop["nsInstanceId"], 

325 "mismatch between nsId '_id' in created nslcmop and database nsr", 

326 ) 

327 self.assertTrue( 

328 created_nslcmop["lcmOperationType"] == "migrate", 

329 "Database record must contain 'lcmOperationType=migrate'", 

330 ) 

331 with self.subTest(i=2, t="Migration of all VDUs in a VNF"): 

332 indata = { 

333 "lcmOperationType": "migrate", 

334 "nsInstanceId": self.nsr_id, 

335 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f", 

336 } 

337 nslcmop_id, _ = self.nslcmop_topic.new( 

338 rollback, session, indata, kwargs=None, headers=headers 

339 ) 

340 

341 self.assertEqual( 

342 self.db.create.call_count, 

343 2, 

344 "database create not called, or called more than once", 

345 ) 

346 _call = self.db.create.call_args_list[0] 

347 self.assertEqual( 

348 _call[0][0], "nslcmops", "nslcmops entry must be created at database" 

349 ) 

350 created_nslcmop = _call[0][1] 

351 self.assertEqual( 

352 self.nsr_id, 

353 created_nslcmop["nsInstanceId"], 

354 "mismatch between nsId '_id' in created nslcmop and database nsr", 

355 ) 

356 self.assertTrue( 

357 created_nslcmop["lcmOperationType"] == "migrate", 

358 "Database record must contain 'lcmOperationType=migrate'", 

359 ) 

360 with self.subTest(i=3, t="Migration failure - vduId not provided in vdu "): 

361 indata = { 

362 "lcmOperationType": "migrate", 

363 "nsInstanceId": self.nsr_id, 

364 "migrateToHost": "sample02", 

365 "vdu": {"vduCountIndex": 0}, 

366 "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f", 

367 } 

368 

369 with self.assertRaises(Exception) as e: 

370 nslcmop_id, _ = self.nslcmop_topic.new( 

371 rollback, session, indata, kwargs=None, headers=headers 

372 ) 

373 self.assertTrue( 

374 "Format error at 'vdu' ''vduId' is a required property'" 

375 in str(e.exception) 

376 ) 

377 

378 

379class TestNsLcmOpTopicWithMock(unittest.TestCase): 

380 def setUp(self): 

381 self.db = Mock(dbbase.DbBase()) 

382 self.fs = Mock(FsBase()) 

383 self.fs.get_params.return_value = {"./fake/folder"} 

384 self.fs.file_open = mock_open() 

385 self.msg = Mock(MsgBase()) 

386 # create class 

387 self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) 

388 

389 def test_get_vnfd_from_vnf_member_revision(self): 

390 test_vnfr = yaml.safe_load(db_vnfrs_text)[0] 

391 test_vnfd = yaml.safe_load(db_vnfds_text) 

392 self.db.get_one.side_effect = [test_vnfr, test_vnfd] 

393 _ = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"]) 

394 self.assertEqual( 

395 self.db.get_one.call_args_list[0][0][0], 

396 "vnfrs", 

397 "Incorrect first DB lookup", 

398 ) 

399 self.assertEqual( 

400 self.db.get_one.call_args_list[1][0][0], 

401 "vnfds", 

402 "Incorrect second DB lookup", 

403 ) 

404 

405 def test_get_vnfd_from_vnf_member_no_revision(self): 

406 test_vnfr = yaml.safe_load(db_vnfrs_text)[0] 

407 test_vnfr["revision"] = 3 

408 test_vnfd = yaml.safe_load(db_vnfds_text) 

409 self.db.get_one.side_effect = [test_vnfr, test_vnfd] 

410 _ = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"]) 

411 self.assertEqual( 

412 self.db.get_one.call_args_list[0][0][0], 

413 "vnfrs", 

414 "Incorrect first DB lookup", 

415 ) 

416 self.assertEqual( 

417 self.db.get_one.call_args_list[1][0][0], 

418 "vnfds_revisions", 

419 "Incorrect second DB lookup", 

420 ) 

421 

422 @contextmanager 

423 def assertNotRaises(self, exception_type): 

424 try: 

425 yield None 

426 except exception_type: 

427 raise self.failureException("{} raised".format(exception_type.__name__)) 

428 

429 def test_check_ns_update_operation(self): 

430 self.db = DbMemory() 

431 self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) 

432 session = {} 

433 

434 with self.subTest(i=1, t="VNF instance does not belong to NS"): 

435 test_vnfr = yaml.safe_load(db_vnfrs_text) 

436 test_vnfr[0]["revision"] = 2 

437 test_nsr = yaml.safe_load(db_nsrs_text) 

438 test_nsr[0]["constituent-vnfr-ref"][ 

439 0 

440 ] = "99d90b0c-faff-4b9f-bccd-017f33985984" 

441 self.db.create_list("vnfrs", test_vnfr) 

442 self.db.create_list("nsrs", test_nsr) 

443 nsrs = self.db.get_list("nsrs")[0] 

444 indata = { 

445 "updateType": "CHANGE_VNFPKG", 

446 "changeVnfPackageData": { 

447 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", 

448 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", 

449 }, 

450 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", 

451 } 

452 with self.assertRaises(EngineException) as expected_exception: 

453 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) 

454 self.assertEqual( 

455 str(expected_exception.exception), 

456 "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984" 

457 " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85", 

458 ) 

459 

460 with self.subTest(i=2, t="Ns update request validated with no exception"): 

461 test_vnfr = yaml.safe_load(db_vnfrs_text) 

462 test_vnfr[0]["revision"] = 2 

463 test_nsr = yaml.safe_load(db_nsrs_text) 

464 self.db.create_list("vnfrs", test_vnfr) 

465 self.db.create_list("nsrs", test_nsr) 

466 nsrs = self.db.get_list("nsrs")[1] 

467 indata = { 

468 "updateType": "CHANGE_VNFPKG", 

469 "changeVnfPackageData": { 

470 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", 

471 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", 

472 }, 

473 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", 

474 } 

475 with self.assertNotRaises(EngineException): 

476 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) 

477 

478 with self.subTest( 

479 i=3, t="Ns update request rejected because of too small timeout" 

480 ): 

481 indata = { 

482 "updateType": "CHANGE_VNFPKG", 

483 "changeVnfPackageData": { 

484 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", 

485 "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", 

486 }, 

487 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", 

488 "timeout_ns_update": 50, 

489 } 

490 with self.assertRaises(EngineException) as expected_exception: 

491 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) 

492 self.assertEqual( 

493 str(expected_exception.exception), 

494 "Error in validating ns-update request: 50 second is not enough " 

495 "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984", 

496 ) 

497 

498 with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"): 

499 test_vnfr = yaml.safe_load(db_vnfrs_text) 

500 test_vnfr[0]["revision"] = 2 

501 test_nsr = yaml.safe_load(db_nsrs_text) 

502 self.db.create_list("vnfrs", test_vnfr) 

503 self.db.create_list("nsrs", test_nsr) 

504 nsrs = self.db.get_list("nsrs")[2] 

505 indata = { 

506 "updateType": "CHANGE_VNFPKG", 

507 "changeVnfPackageData": { 

508 "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", 

509 "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77", 

510 }, 

511 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", 

512 } 

513 with self.assertRaises(EngineException) as expected_exception: 

514 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) 

515 self.assertEqual( 

516 str(expected_exception.exception), 

517 "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not " 

518 "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of " 

519 "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984", 

520 ) 

521 

522 with self.subTest( 

523 i=5, t="Ns update REMOVE_VNF request validated with no exception" 

524 ): 

525 test_vnfr = yaml.safe_load(db_vnfrs_text) 

526 test_vnfr[0]["revision"] = 2 

527 test_nsr = yaml.safe_load(db_nsrs_text) 

528 self.db.create_list("vnfrs", test_vnfr) 

529 self.db.create_list("nsrs", test_nsr) 

530 nsrs = self.db.get_list("nsrs")[1] 

531 indata = { 

532 "updateType": "REMOVE_VNF", 

533 "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", 

534 "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", 

535 } 

536 with self.assertNotRaises(EngineException): 

537 self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) 

538 

539 

540class TestNsrTopic(unittest.TestCase): 

541 def setUp(self): 

542 self.db = DbMemory() 

543 self.fs = Mock(FsBase()) 

544 self.fs.get_params.return_value = {"./fake/folder"} 

545 self.fs.file_open = mock_open() 

546 self.msg = Mock(MsgBase()) 

547 # create class 

548 self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None) 

549 self.nsr_topic.check_quota = Mock(return_value=None) # skip quota 

550 

551 self.db.create_list("vim_accounts", yaml.safe_load(db_vim_accounts_text)) 

552 self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) 

553 self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) 

554 self.db.create = Mock(return_value="created_id") 

555 self.nsd = self.db.get_list("nsds")[0] 

556 self.nsd_id = self.nsd["_id"] 

557 self.nsd_project = self.nsd["_admin"]["projects_read"][0] 

558 

559 self.vim = self.db.get_list("vim_accounts")[0] 

560 self.vim_id = self.vim["_id"] 

561 

562 def test_create(self): 

563 session = { 

564 "force": False, 

565 "admin": False, 

566 "public": False, 

567 "project_id": [self.nsd_project], 

568 "method": "write", 

569 } 

570 indata = { 

571 "nsdId": self.nsd_id, 

572 "nsName": "name", 

573 "vimAccountId": self.vim_id, 

574 "additionalParamsForVnf": [ 

575 { 

576 "member-vnf-index": "hackfest_vnf1", 

577 "additionalParams": {"touch_filename": "file"}, 

578 }, 

579 { 

580 "member-vnf-index": "hackfest_vnf2", 

581 "additionalParams": {"touch_filename": "file"}, 

582 }, 

583 ], 

584 } 

585 rollback = [] 

586 headers = {} 

587 

588 self.nsr_topic.new( 

589 rollback, session, indata=indata, kwargs=None, headers=headers 

590 ) 

591 

592 # check vnfrs and nsrs created in whatever order 

593 created_vnfrs = [] 

594 created_nsrs = [] 

595 nsr_id = None 

596 for _call in self.db.create.call_args_list: 

597 assert len(_call[0]) >= 2, "called db.create with few parameters" 

598 created_item = _call[0][1] 

599 if _call[0][0] == "vnfrs": 

600 created_vnfrs.append(created_item) 

601 self.assertIn( 

602 "member-vnf-index-ref", 

603 created_item, 

604 "Created item must contain member-vnf-index-ref section", 

605 ) 

606 if nsr_id: 

607 self.assertEqual( 

608 nsr_id, 

609 created_item["nsr-id-ref"], 

610 "bad reference id from vnfr to nsr", 

611 ) 

612 else: 

613 nsr_id = created_item["nsr-id-ref"] 

614 

615 elif _call[0][0] == "nsrs": 

616 created_nsrs.append(created_item) 

617 if nsr_id: 

618 self.assertEqual( 

619 nsr_id, created_item["_id"], "bad reference id from vnfr to nsr" 

620 ) 

621 else: 

622 nsr_id = created_item["_id"] 

623 else: 

624 assert True, "created an unknown record {} at database".format( 

625 _call[0][0] 

626 ) 

627 

628 self.assertTrue( 

629 created_item["_admin"].get("projects_read"), 

630 "Database record must contain '_amdin.projects_read'", 

631 ) 

632 self.assertIn( 

633 "created", 

634 created_item["_admin"], 

635 "Database record must contain '_admin.created'", 

636 ) 

637 self.assertTrue( 

638 created_item["_admin"]["nsState"] == "NOT_INSTANTIATED", 

639 "Database record must contain '_admin.nstate=NOT INSTANTIATE'", 

640 ) 

641 

642 self.assertEqual( 

643 len(created_vnfrs), 2, "created a mismatch number of vnfr at database" 

644 ) 

645 

646 self.assertEqual( 

647 created_vnfrs[0]["vdur"][0]["interfaces"][0]["position"], 

648 1, 

649 "vdur first interface position does not match", 

650 ) 

651 

652 self.assertEqual( 

653 created_vnfrs[0]["vdur"][0]["interfaces"][1]["position"], 

654 2, 

655 "vdur second interface position does not match", 

656 ) 

657 

658 self.assertEqual( 

659 len(created_nsrs), 1, "Only one nsrs must be created at database" 

660 ) 

661 self.assertEqual( 

662 len(rollback), 

663 len(created_vnfrs) + 1, 

664 "rollback mismatch with created items at database", 

665 ) 

666 

667 # test parameters with error 

668 bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa" 

669 test_set = ( 

670 # TODO add "nsd" 

671 ( 

672 "nsd not found", 

673 {"nsdId": bad_id}, 

674 DbException, 

675 HTTPStatus.NOT_FOUND, 

676 ("not found", bad_id), 

677 ), 

678 # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)), # TODO add "vim" 

679 ( 

680 "additional params not supply", 

681 {"additionalParamsForVnf.0.member-vnf-index": "k"}, 

682 EngineException, 

683 HTTPStatus.BAD_REQUEST, 

684 None, 

685 ), 

686 ) 

687 for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set: 

688 with self.assertRaises(expect_exc, msg=message) as e: 

689 self.nsr_topic.new( 

690 rollback, 

691 session, 

692 indata=deepcopy(indata), 

693 kwargs=kwargs_, 

694 headers=headers, 

695 ) 

696 if expect_code: 

697 self.assertTrue(e.exception.http_code == expect_code) 

698 if expect_text_list: 

699 for expect_text in expect_text_list: 

700 self.assertIn( 

701 expect_text, 

702 str(e.exception).lower(), 

703 "Expected '{}' at exception text".format(expect_text), 

704 ) 

705 

706 def test_show_instance(self): 

707 session = { 

708 "force": False, 

709 "admin": False, 

710 "public": False, 

711 "project_id": [self.nsd_project], 

712 "method": "write", 

713 } 

714 filter_q = {} 

715 for refresh_status in ("true", "false"): 

716 self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) 

717 actual_nsr = self.db.get_list("nsrs")[0] 

718 nsr_id = actual_nsr["_id"] 

719 filter_q["vcaStatus-refresh"] = refresh_status 

720 expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q) 

721 self.nsr_topic.delete(session, nsr_id) 

722 actual_nsr.pop("_admin") 

723 expected_nsr.pop("_admin") 

724 self.assertEqual( 

725 expected_nsr, actual_nsr, "Database nsr and show() nsr do not match." 

726 ) 

727 

728 def test_vca_status_refresh(self): 

729 session = { 

730 "force": False, 

731 "admin": False, 

732 "public": False, 

733 "project_id": [self.nsd_project], 

734 "method": "write", 

735 } 

736 filter_q = {"vcaStatus-refresh": "true"} 

737 time_delta = 120 

738 self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) 

739 nsr = self.db.get_list("nsrs")[0] 

740 

741 # When vcaStatus-refresh is true 

742 filter_q["vcaStatus-refresh"] = "true" 

743 self.nsr_topic.vca_status_refresh(session, nsr, filter_q) 

744 msg_args = self.msg.write.call_args[0] 

745 self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action") 

746 self.assertGreater(nsr["_admin"]["modified"], time() - time_delta) 

747 

748 # When vcaStatus-refresh is false but modified time is within threshold 

749 filter_q["vcaStatus-refresh"] = "false" 

750 time_now = time() 

751 nsr["_admin"]["modified"] = time_now 

752 self.nsr_topic.vca_status_refresh(session, nsr, filter_q) 

753 msg_args = self.msg.write.call_args[1] 

754 self.assertEqual(msg_args, {}, "Message should not be sent.") 

755 self.assertEqual( 

756 nsr["_admin"]["modified"], time_now, "Modified time should not be changed." 

757 ) 

758 

759 # When vcaStatus-refresh is false but modified time is less than threshold 

760 filter_q["vcaStatus-refresh"] = "false" 

761 nsr["_admin"]["modified"] = time() - (2 * time_delta) 

762 self.nsr_topic.vca_status_refresh(session, nsr, filter_q) 

763 msg_args = self.msg.write.call_args[0] 

764 self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action") 

765 self.nsr_topic.delete(session, nsr["_id"]) 

766 self.assertGreater( 

767 nsr["_admin"]["modified"], 

768 time() - time_delta, 

769 "Modified time is not changed.", 

770 ) 

771 

772 def test_delete_ns(self): 

773 self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) 

774 self.nsr = self.db.get_list("nsrs")[0] 

775 self.nsr_id = self.nsr["_id"] 

776 self.db_set_one = self.db.set_one 

777 p_id = self.nsd_project 

778 p_other = "other_p" 

779 

780 session = { 

781 "force": False, 

782 "admin": False, 

783 "public": None, 

784 "project_id": [p_id], 

785 "method": "delete", 

786 } 

787 session2 = { 

788 "force": False, 

789 "admin": False, 

790 "public": None, 

791 "project_id": [p_other],