self.validate_internal_virtual_links(indata)
self.validate_monitoring_params(indata)
self.validate_scaling_group_descriptor(indata)
+ self.validate_healing_group_descriptor(indata)
+ self.validate_alarm_group_descriptor(indata)
+ self.validate_storage_compute_descriptor(indata)
self.validate_helm_chart(indata)
return indata
@staticmethod
def validate_scaling_group_descriptor(indata):
all_monitoring_params = set()
+ all_vdu_ids = set()
+ for df in get_iterable(indata.get("df")):
+ for il in get_iterable(df.get("instantiation-level")):
+ for vl in get_iterable(il.get("vdu-level")):
+ all_vdu_ids.add(vl.get("vdu-id"))
+
for ivld in get_iterable(indata.get("int-virtual-link-desc")):
for mp in get_iterable(ivld.get("monitoring-parameters")):
all_monitoring_params.add(mp.get("id"))
for mp in get_iterable(df.get("monitoring-parameter")):
all_monitoring_params.add(mp.get("id"))
+ for df in get_iterable(indata.get("df")):
+ for sa in get_iterable(df.get("scaling-aspect")):
+ for deltas in get_iterable(
+ sa.get("aspect-delta-details").get("deltas")
+ ):
+ for vds in get_iterable(deltas.get("vdu-delta")):
+ sa_vdu_id = vds.get("id")
+ if sa_vdu_id and sa_vdu_id not in all_vdu_ids:
+ raise EngineException(
+ "df[id='{}']:scaling-aspect[id='{}']:aspect-delta-details"
+ "[delta='{}']: "
+ "vdu-id='{}' not defined in vdu".format(
+ df["id"],
+ sa["id"],
+ deltas["id"],
+ sa_vdu_id,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
for df in get_iterable(indata.get("df")):
for sa in get_iterable(df.get("scaling-aspect")):
for sp in get_iterable(sa.get("scaling-policy")):
http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
)
+ @staticmethod
+ def validate_healing_group_descriptor(indata):
+ all_vdu_ids = set()
+ for df in get_iterable(indata.get("df")):
+ for il in get_iterable(df.get("instantiation-level")):
+ for vl in get_iterable(il.get("vdu-level")):
+ all_vdu_ids.add(vl.get("vdu-id"))
+
+ for df in get_iterable(indata.get("df")):
+ for ha in get_iterable(df.get("healing-aspect")):
+ for hp in get_iterable(ha.get("healing-policy")):
+ hp_monitoring_param = hp.get("vdu-id")
+ if hp_monitoring_param and hp_monitoring_param not in all_vdu_ids:
+ raise EngineException(
+ "df[id='{}']:healing-aspect[id='{}']:healing-policy"
+ "[name='{}']: "
+ "vdu-id='{}' not defined in vdu".format(
+ df["id"],
+ ha["id"],
+ hp["event-name"],
+ hp_monitoring_param,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_alarm_group_descriptor(indata):
+ all_monitoring_params = set()
+ for ivld in get_iterable(indata.get("int-virtual-link-desc")):
+ for mp in get_iterable(ivld.get("monitoring-parameters")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for vdu in get_iterable(indata.get("vdu")):
+ for mp in get_iterable(vdu.get("monitoring-parameter")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for df in get_iterable(indata.get("df")):
+ for mp in get_iterable(df.get("monitoring-parameter")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for vdus in get_iterable(indata.get("vdu")):
+ for alarms in get_iterable(vdus.get("alarm")):
+ alarm_monitoring_param = alarms.get("vnf-monitoring-param-ref")
+ if (
+ alarm_monitoring_param
+ and alarm_monitoring_param not in all_monitoring_params
+ ):
+ raise EngineException(
+ "vdu[id='{}']:alarm[id='{}']:"
+ "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
+ vdus["id"],
+ alarms["alarm-id"],
+ alarm_monitoring_param,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_storage_compute_descriptor(indata):
+ all_vsd_ids = set()
+ for vsd in get_iterable(indata.get("virtual-storage-desc")):
+ all_vsd_ids.add(vsd.get("id"))
+
+ all_vcd_ids = set()
+ for vcd in get_iterable(indata.get("virtual-compute-desc")):
+ all_vcd_ids.add(vcd.get("id"))
+
+ for vdus in get_iterable(indata.get("vdu")):
+ for vsd_ref in vdus.get("virtual-storage-desc"):
+ if vsd_ref and vsd_ref not in all_vsd_ids:
+ raise EngineException(
+ "vdu[virtual-storage-desc='{}']"
+ "not defined in vnfd".format(
+ vsd_ref,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ for vdus in get_iterable(indata.get("vdu")):
+ vcd_ref = vdus.get("virtual-compute-desc")
+ if vcd_ref and vcd_ref not in all_vcd_ids:
+ raise EngineException(
+ "vdu[virtual-compute-desc='{}']"
+ "not defined in vnfd".format(
+ vdus["virtual-compute-desc"],
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
def delete_extra(self, session, _id, db_content, not_send_msg=None):
"""
Deletes associate file system storage (via super)
"Wrong exception text",
)
+ @patch("osm_nbi.descriptor_topics.shutil")
+ @patch("osm_nbi.descriptor_topics.os.rename")
+ def test_new_vnfd_check_input_validation_scaling_criteria_vdu_id(
+ self, mock_rename, mock_shutil
+ ):
+ """Testing input validation during new vnfd creation
+ for scaling criteria with invalid vdu-id"""
+ did, test_vnfd = self.prepare_vnfd_creation()
+ test_vnfd = self.prepare_test_vnfd(test_vnfd)
+ test_vnfd["df"][0]["scaling-aspect"][0]["aspect-delta-details"]["deltas"][0][
+ "vdu-delta"
+ ][0]["id"] = "vdudelta1"
+ affected_df = test_vnfd["df"][0]
+ sa = affected_df["scaling-aspect"][0]
+ delta = sa["aspect-delta-details"]["deltas"][0]
+ vdu_delta = delta["vdu-delta"][0]
+
+ with self.assertRaises(
+ EngineException, msg="Accepted invalid Scaling Group Policy Criteria"
+ ) as e:
+ self.topic.upload_content(
+ fake_session, did, test_vnfd, {}, {"Content-Type": []}
+ )
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "df[id='{}']:scaling-aspect[id='{}']:aspect-delta-details"
+ "[delta='{}']: "
+ "vdu-id='{}' not defined in vdu".format(
+ affected_df["id"],
+ sa["id"],
+ delta["id"],
+ vdu_delta["id"],
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
@patch("osm_nbi.descriptor_topics.shutil")
@patch("osm_nbi.descriptor_topics.os.rename")
def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref(
"Wrong exception text",
)
+ @patch("osm_nbi.descriptor_topics.shutil")
+ @patch("osm_nbi.descriptor_topics.os.rename")
+ def test_new_vnfd_check_input_validation_healing_criteria_vdu_id(
+ self, mock_rename, mock_shutil
+ ):
+ """Testing input validation during new vnfd creation
+ for healing criteria with invalid vdu-id"""
+ did, test_vnfd = self.prepare_vnfd_creation()
+ test_vnfd = self.prepare_test_vnfd(test_vnfd)
+ test_vnfd["df"][0]["healing-aspect"][0]["healing-policy"][0][
+ "vdu-id"
+ ] = "vduid1"
+ affected_df = test_vnfd["df"][0]
+ ha = affected_df["healing-aspect"][0]
+ hp = ha["healing-policy"][0]
+ hp_vdu_id = hp["vdu-id"]
+
+ with self.assertRaises(
+ EngineException, msg="Accepted invalid Healing Group Policy Criteria"
+ ) as e:
+ self.topic.upload_content(
+ fake_session, did, test_vnfd, {}, {"Content-Type": []}
+ )
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "df[id='{}']:healing-aspect[id='{}']:healing-policy"
+ "[name='{}']: "
+ "vdu-id='{}' not defined in vdu".format(
+ affected_df["id"],
+ ha["id"],
+ hp["event-name"],
+ hp_vdu_id,
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ @patch("osm_nbi.descriptor_topics.shutil")
+ @patch("osm_nbi.descriptor_topics.os.rename")
+ def test_new_vnfd_check_input_validation_alarm_criteria_monitoring_param_ref(
+ self, mock_rename, mock_shutil
+ ):
+ """Testing input validation during new vnfd creation
+ for alarm with invalid monitoring parameter reference"""
+ did, test_vnfd = self.prepare_vnfd_creation()
+ test_vnfd = self.prepare_test_vnfd(test_vnfd)
+ test_vnfd["vdu"][1]["alarm"][0]["vnf-monitoring-param-ref"] = "unit_test_alarm"
+ vdu = test_vnfd["vdu"][1]
+ alarm = vdu["alarm"][0]
+ alarm_monitoring_param = alarm["vnf-monitoring-param-ref"]
+
+ with self.assertRaises(
+ EngineException, msg="Accepted invalid Alarm Criteria"
+ ) as e:
+ self.topic.upload_content(
+ fake_session, did, test_vnfd, {}, {"Content-Type": []}
+ )
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "vdu[id='{}']:alarm[id='{}']:"
+ "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
+ vdu["id"],
+ alarm["alarm-id"],
+ alarm_monitoring_param,
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ @patch("osm_nbi.descriptor_topics.shutil")
+ @patch("osm_nbi.descriptor_topics.os.rename")
+ def test_new_vnfd_check_input_validation_storage_reference_criteria(
+ self, mock_rename, mock_shutil
+ ):
+ """Testing input validation during new vnfd creation
+ for invalid virtual-storge-desc reference"""
+ did, test_vnfd = self.prepare_vnfd_creation()
+ test_vnfd = self.prepare_test_vnfd(test_vnfd)
+ test_vnfd["vdu"][1]["virtual-storage-desc"] = "unit_test_storage"
+ vdu = test_vnfd["vdu"][1]
+ vsd_ref = vdu["virtual-storage-desc"]
+
+ with self.assertRaises(
+ EngineException, msg="Accepted invalid virtual-storage-desc"
+ ) as e:
+ self.topic.upload_content(
+ fake_session, did, test_vnfd, {}, {"Content-Type": []}
+ )
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "vdu[virtual-storage-desc='{}']"
+ "not defined in vnfd".format(
+ vsd_ref,
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ @patch("osm_nbi.descriptor_topics.shutil")
+ @patch("osm_nbi.descriptor_topics.os.rename")
+ def test_new_vnfd_check_input_validation_compute_reference_criteria(
+ self, mock_rename, mock_shutil
+ ):
+ """Testing input validation during new vnfd creation
+ for invalid virtual-compute-desc reference"""
+ did, test_vnfd = self.prepare_vnfd_creation()
+ test_vnfd = self.prepare_test_vnfd(test_vnfd)
+ test_vnfd["vdu"][1]["virtual-compute-desc"] = "unit_test_compute"
+ vdu = test_vnfd["vdu"][1]
+ vcd_ref = vdu["virtual-compute-desc"]
+
+ with self.assertRaises(
+ EngineException, msg="Accepted invalid virtual-compute-desc"
+ ) as e:
+ self.topic.upload_content(
+ fake_session, did, test_vnfd, {}, {"Content-Type": []}
+ )
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "vdu[virtual-compute-desc='{}']"
+ "not defined in vnfd".format(
+ vcd_ref,
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
@patch("osm_nbi.descriptor_topics.shutil")
@patch("osm_nbi.descriptor_topics.os.rename")
def test_new_vnfd_check_input_validation_everything_right(
projects / osm / NBI.git / commitdiff