import yaml
from n2vc.loggable import Loggable
-from n2vc.utils import EntityType, JujuStatusToOSM, N2VCDeploymentStatus
+from n2vc.utils import JujuStatusToOSM, N2VCDeploymentStatus
class N2VCConnector(abc.ABC, Loggable):
# generate private/public key-pair
self.private_key_path = None
self.public_key_path = None
- self.get_public_key()
@abc.abstractmethod
async def get_status(self, namespace: str, yaml_format: bool = True):
# Find the path where we expect our key lives (~/.ssh)
homedir = os.environ.get("HOME")
if not homedir:
- self.warning("No HOME environment variable, using /tmp")
+ self.log.warning("No HOME environment variable, using /tmp")
homedir = "/tmp"
sshdir = "{}/.ssh".format(homedir)
if not os.path.exists(sshdir):
:param float total_timeout:
"""
+ @abc.abstractmethod
+ async def install_k8s_proxy_charm(
+ self,
+ charm_name: str,
+ namespace: str,
+ artifact_path: str,
+ db_dict: dict,
+ progress_timeout: float = None,
+ total_timeout: float = None,
+ config: dict = None,
+ ) -> str:
+ """
+ Install a k8s proxy charm
+
+ :param charm_name: Name of the charm being deployed
+ :param namespace: collection of all the uuids related to the charm.
+ :param str artifact_path: where to locate the artifacts (parent folder) using
+ the self.fs
+ the final artifact path will be a combination of this artifact_path and
+ additional string from the config_dict (e.g. charm name)
+ :param dict db_dict: where to write into database when the status changes.
+ It contains a dict with
+ {collection: <str>, filter: {}, path: <str>},
+ e.g. {collection: "nsrs", filter:
+ {_id: <nsd-id>, path: "_admin.deployed.VCA.3"}
+ :param float progress_timeout:
+ :param float total_timeout:
+ :param config: Dictionary with additional configuration
+
+ :returns ee_id: execution environment id.
+ """
+
@abc.abstractmethod
async def get_ee_ssh_public__key(
self,
else:
self.log.info("Exception writing status to database: {}".format(e))
- def osm_status(self, entity_type: EntityType, status: str) -> N2VCDeploymentStatus:
+ def osm_status(self, entity_type: str, status: str) -> N2VCDeploymentStatus:
if status not in JujuStatusToOSM[entity_type]:
- self.log.warning("Status {} not found in JujuStatusToOSM.")
+ self.log.warning("Status {} not found in JujuStatusToOSM.".format(status))
return N2VCDeploymentStatus.UNKNOWN
return JujuStatusToOSM[entity_type][status]
-# DEPRECATED
-def juju_status_2_osm_status(statustype: str, status: str) -> N2VCDeploymentStatus:
- if statustype == "application" or statustype == "unit":
- if status in ["waiting", "maintenance"]:
- return N2VCDeploymentStatus.RUNNING
- if status in ["error"]:
- return N2VCDeploymentStatus.FAILED
- elif status in ["active"]:
- return N2VCDeploymentStatus.COMPLETED
- elif status in ["blocked"]:
- return N2VCDeploymentStatus.RUNNING
- else:
- return N2VCDeploymentStatus.UNKNOWN
- elif statustype == "action":
- if status in ["running"]:
- return N2VCDeploymentStatus.RUNNING
- elif status in ["completed"]:
- return N2VCDeploymentStatus.COMPLETED
- else:
- return N2VCDeploymentStatus.UNKNOWN
- elif statustype == "machine":
- if status in ["pending"]:
- return N2VCDeploymentStatus.PENDING
- elif status in ["started"]:
- return N2VCDeploymentStatus.COMPLETED
- else:
- return N2VCDeploymentStatus.UNKNOWN
-
- return N2VCDeploymentStatus.FAILED
-
-
def obj_to_yaml(obj: object) -> str:
# dump to yaml
dump_text = yaml.dump(obj, default_flow_style=False, indent=2)