+ if session["force"]:
+ return
+ # check if used by VNF
+ filter_q = {"kdur.k8s-cluster.id": _id}
+ if session["project_id"]:
+ filter_q["_admin.projects_read.cont"] = session["project_id"]
+ if self.db.get_list("vnfrs", filter_q):
+ raise EngineException(
+ "There is at least one VNF using this k8scluster",
+ http_code=HTTPStatus.CONFLICT,
+ )
+ super().check_conflict_on_del(session, _id, db_content)
+
+
+class VcaTopic(CommonVimWimSdn):
+ topic = "vca"
+ topic_msg = "vca"
+ schema_new = vca_new_schema
+ schema_edit = vca_edit_schema
+ multiproject = True
+ password_to_encrypt = None
+
+ def format_on_new(self, content, project_id=None, make_public=False):
+ oid = super().format_on_new(content, project_id, make_public)
+ content["schema_version"] = schema_version = "1.11"
+ for key in ["secret", "cacert"]:
+ content[key] = self.db.encrypt(
+ content[key], schema_version=schema_version, salt=content["_id"]
+ )
+ return oid
+
+ def format_on_edit(self, final_content, edit_content):
+ oid = super().format_on_edit(final_content, edit_content)
+ schema_version = final_content.get("schema_version")
+ for key in ["secret", "cacert"]:
+ if key in edit_content:
+ final_content[key] = self.db.encrypt(
+ edit_content[key],
+ schema_version=schema_version,
+ salt=final_content["_id"],
+ )
+ return oid
+
+ def check_conflict_on_del(self, session, _id, db_content):
+ """
+ Check if deletion can be done because of dependencies if it is not force. To override
+ :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
+ :param _id: internal _id
+ :param db_content: The database content of this item _id
+ :return: None if ok or raises EngineException with the conflict
+ """
+ if session["force"]:
+ return
+ # check if used by VNF
+ filter_q = {"vca": _id}
+ if session["project_id"]:
+ filter_q["_admin.projects_read.cont"] = session["project_id"]
+ if self.db.get_list("vim_accounts", filter_q):
+ raise EngineException(
+ "There is at least one VIM account using this vca",
+ http_code=HTTPStatus.CONFLICT,
+ )
+ super().check_conflict_on_del(session, _id, db_content)
+
+
+class K8sRepoTopic(CommonVimWimSdn):
+ topic = "k8srepos"
+ topic_msg = "k8srepo"
+ schema_new = k8srepo_new_schema
+ schema_edit = k8srepo_edit_schema
+ multiproject = True
+ password_to_encrypt = None
+ config_to_encrypt = {}
+
+ def format_on_new(self, content, project_id=None, make_public=False):
+ oid = super().format_on_new(content, project_id, make_public)
+ # Update Helm/Juju Repo lists
+ repo_list = content["type"].replace("-", "_") + "_repos"
+ for proj in content["_admin"]["projects_read"]:
+ if proj != "ANY":
+ self.db.set_list(
+ "k8sclusters",
+ {
+ "_admin.projects_read": proj,
+ "_admin." + repo_list + ".ne": content["_id"],
+ },
+ {},
+ push={"_admin." + repo_list: content["_id"]},
+ )
+ return oid
+
+ def delete(self, session, _id, dry_run=False, not_send_msg=None):
+ type = self.db.get_one("k8srepos", {"_id": _id})["type"]
+ oid = super().delete(session, _id, dry_run, not_send_msg)
+ if oid:
+ # Remove from Helm/Juju Repo lists
+ repo_list = type.replace("-", "_") + "_repos"
+ self.db.set_list(
+ "k8sclusters",
+ {"_admin." + repo_list: _id},
+ {},
+ pull={"_admin." + repo_list: _id},
+ )
+ return oid
+
+
+class OsmRepoTopic(BaseTopic):
+ topic = "osmrepos"
+ topic_msg = "osmrepos"
+ schema_new = osmrepo_new_schema
+ schema_edit = osmrepo_edit_schema
+ multiproject = True
+ # TODO: Implement user/password