- try:
- self.db.get_one.side_effect = [
- {"_id": did, "_admin": db_nsd_content["_admin"]},
- None,
- ]
- self.db.get_list.return_value = [db_vnfd_content]
- self.topic.upload_content(
- fake_session, did, test_nsd, {}, {"Content-Type": []}
- )
- msg_args = self.msg.write.call_args[0]
- test_nsd["_id"] = did
- self.assertEqual(
- msg_args[0], self.topic.topic_msg, "Wrong message topic"
- )
- self.assertEqual(msg_args[1], "edited", "Wrong message action")
- self.assertEqual(msg_args[2], test_nsd, "Wrong message content")
- db_args = self.db.get_one.mock_calls[0][1]
- self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
- self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id")
- db_args = self.db.replace.call_args[0]
- self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic")
- self.assertEqual(db_args[1], did, "Wrong DB NSD id")
- admin = db_args[2]["_admin"]
- db_admin = db_nsd_content["_admin"]
- self.assertEqual(
- admin["created"], db_admin["created"], "Wrong creation time"
- )
- self.assertGreater(
- admin["modified"], db_admin["created"], "Wrong modification time"
- )
- self.assertEqual(
- admin["projects_read"],
- db_admin["projects_read"],
- "Wrong read-only project list",
- )
- self.assertEqual(
- admin["projects_write"],
- db_admin["projects_write"],
- "Wrong read-write project list",
- )
- self.assertEqual(
- admin["onboardingState"], "ONBOARDED", "Wrong onboarding state"
- )
- self.assertEqual(
- admin["operationalState"], "ENABLED", "Wrong operational state"
- )
- self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state")
- storage = admin["storage"]
- self.assertEqual(storage["folder"], did, "Wrong storage folder")
- self.assertEqual(
- storage["descriptor"], "package", "Wrong storage descriptor"
- )
- compare_desc(self, test_nsd, db_args[2], "NSD")
- finally:
- pass
- self.db.get_one.side_effect = (
- lambda table, filter, fail_on_empty=None, fail_on_more=None: {
- "_id": did,
- "_admin": db_nsd_content["_admin"],
- }
- )
- with self.subTest(i=2, t="Check Pyangbind Validation: required properties"):
- tmp = test_nsd["id"]
- del test_nsd["id"]
- try:
- with self.assertRaises(
- EngineException, msg="Accepted NSD with a missing required property"
- ) as e:
- self.topic.upload_content(
- fake_session, did, test_nsd, {}, {"Content-Type": []}
- )
- self.assertEqual(
- e.exception.http_code,
- HTTPStatus.UNPROCESSABLE_ENTITY,
- "Wrong HTTP status code",
- )
- self.assertIn(
- norm("Error in pyangbind validation: '{}'".format("id")),
- norm(str(e.exception)),
- "Wrong exception text",
- )
- finally:
- test_nsd["id"] = tmp
- with self.subTest(i=3, t="Check Pyangbind Validation: additional properties"):
- test_nsd["extra-property"] = 0
- try:
- with self.assertRaises(
- EngineException, msg="Accepted NSD with an additional property"
- ) as e:
- self.topic.upload_content(
- fake_session, did, test_nsd, {}, {"Content-Type": []}
- )
- self.assertEqual(
- e.exception.http_code,
- HTTPStatus.UNPROCESSABLE_ENTITY,
- "Wrong HTTP status code",
- )
- self.assertIn(
- norm(
- "Error in pyangbind validation: {} ({})".format(
- "json object contained a key that did not exist",
- "extra-property",
- )
- ),
- norm(str(e.exception)),
- "Wrong exception text",
- )
- finally:
- del test_nsd["extra-property"]
- with self.subTest(i=4, t="Check Pyangbind Validation: property types"):
- tmp = test_nsd["designer"]
- test_nsd["designer"] = {"key": 0}
- try:
- with self.assertRaises(
- EngineException, msg="Accepted NSD with a wrongly typed property"
- ) as e:
- self.topic.upload_content(
- fake_session, did, test_nsd, {}, {"Content-Type": []}
- )
- self.assertEqual(
- e.exception.http_code,
- HTTPStatus.UNPROCESSABLE_ENTITY,
- "Wrong HTTP status code",
- )
- self.assertIn(
- norm(
- "Error in pyangbind validation: {} ({})".format(
- "json object contained a key that did not exist", "key"
- )
- ),
- norm(str(e.exception)),
- "Wrong exception text",
- )
- finally:
- test_nsd["designer"] = tmp
- with self.subTest(
- i=5, t="Check Input Validation: mgmt-network+virtual-link-protocol-data"
- ):
- df = test_nsd["df"][0]
- mgmt_profile = {
- "id": "id",
- "virtual-link-desc-id": "mgmt",
- "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"},
- }
- df["virtual-link-profile"] = [mgmt_profile]
- try:
- with self.assertRaises(
- EngineException, msg="Accepted VLD with mgmt-network+ip-profile"
- ) as e:
- self.topic.upload_content(
- fake_session, did, test_nsd, {}, {"Content-Type": []}
- )
- self.assertEqual(
- e.exception.http_code,
- HTTPStatus.UNPROCESSABLE_ENTITY,
- "Wrong HTTP status code",
- )
- self.assertIn(
- norm(
- "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data"
- " You cannot set a virtual-link-protocol-data when mgmt-network is True".format(
- df["id"], mgmt_profile["id"]
- )
- ),
- norm(str(e.exception)),
- "Wrong exception text",