from http import HTTPStatus
from n2vc.loggable import Loggable
from n2vc.exceptions import N2VCBadArgumentsException
+import yaml
from osm_common.dbmongo import DbException
url: str,
username: str,
vca_config: dict,
- on_update_db = None
+ on_update_db=None
):
"""Initialize N2VC abstract connector. It defines de API for VCA connectors
if fs is None:
raise N2VCBadArgumentsException('Argument fs is mandatory', ['fs'])
- self.info('url={}, username={}, vca_config={}'.format(url, username, vca_config))
+ self.log.info('url={}, username={}, vca_config={}'.format(
+ url, username, {k: v for k, v in vca_config.items() if k not in ("host", "port", "user", "secret",
+ "public_key", "ca_cert")}))
# store arguments into self
self.db = db
self.on_update_db = on_update_db
# 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):
+ async def get_status(self, namespace: str, yaml_format: bool = True):
"""Get namespace status
:param namespace: we obtain ns from namespace
+ :param yaml_format: returns a yaml string
"""
# TODO: review which public key
- async def get_public_key(self) -> str:
+ def get_public_key(self) -> str:
"""Get the VCA ssh-public-key
Returns the SSH public key from local mahine, to be injected into virtual machines to
public_key = ''
# Find the path where we expect our key lives (~/.ssh)
- homedir = os.environ['HOME']
+ homedir = os.environ.get('HOME')
+ if not homedir:
+ self.warning('No HOME environment variable, using /tmp')
+ homedir = '/tmp'
sshdir = "{}/.ssh".format(homedir)
if not os.path.exists(sshdir):
os.mkdir(sshdir)
entity_type: str
):
if not db_dict:
- self.debug('No db_dict => No database write')
+ self.log.debug('No db_dict => No database write')
return
- self.debug('status={} / detailed-status={} / VCA-status={} / entity_type={}'
- .format(str(status.value), detailed_status, vca_status, entity_type))
+ # self.log.debug('status={} / detailed-status={} / VCA-status={} / entity_type={}'
+ # .format(str(status.value), detailed_status, vca_status, entity_type))
try:
except DbException as e:
if e.http_code == HTTPStatus.NOT_FOUND:
- self.error('NOT_FOUND error: Exception writing status to database: {}'.format(e))
+ self.log.error('NOT_FOUND error: Exception writing status to database: {}'.format(e))
else:
- self.info('Exception writing status to database: {}'.format(e))
+ self.log.info('Exception writing status to database: {}'.format(e))
def juju_status_2_osm_status(type: str, status: str) -> N2VCDeploymentStatus:
if type == 'application' or type == '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.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)
+ # split lines
+ lines = dump_text.splitlines()
+ # remove !!python/object tags
+ yaml_text = ''
+ for line in lines:
+ index = line.find('!!python/object')
+ if index >= 0:
+ line = line[:index]
+ yaml_text += line + '\n'
+ return yaml_text
+
+
+def obj_to_dict(obj: object) -> dict:
+ # convert obj to yaml
+ yaml_text = obj_to_yaml(obj)
+ # parse to dict
+ return yaml.load(yaml_text, Loader=yaml.Loader)