Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_log_err(self):
with LogCapture() as log:
utils.log_err("test errors")
assert "test errors" in str(log.records)
def get_project_id(username, password, auth_url, user_domain_name):
sess = generate_session(
username=username,
password=password,
auth_url=auth_url,
user_domain_name=user_domain_name,
)
keystone = client.Client(session=sess)
# project_list = [t.id for t in keystone.projects.list(user=sess.get_user_id())]
enabled_project = []
for project in keystone.projects.list(user=sess.get_user_id()):
if project.enabled == True:
enabled_project.append(project.id)
if len(enabled_project) == 0:
utils.log_err("Something wrong with your project. Please contact Support")
exit()
elif len(enabled_project) > 1:
return enabled_project[0]
else:
return enabled_project[0]
def start_instance(vm_id, session=None):
compute = get_nova_client(session)
try:
return compute.servers.start(vm_id)
except Exception as e:
utils.log_err(e)
def revert_size(vm_id, session=None):
compute = get_nova_client(session)
try:
return compute.servers.revert_resize(vm_id)
except Exception as e:
utils.log_err(e)
def attach_interface(vm_id, port_id, net_id, fixed_ip, session=None):
compute = get_nova_client(session)
try:
attach_ip = compute.servers.interface_attach(
vm_id, port_id, net_id, fixed_ip, tag=None
)
except Exception as e:
utils.log_err(e)
return attach_ip
def list_port(session=None):
obj_port_list = list()
neutron = get_neutron_client(session)
try:
obj_port_list = neutron.list_ports()
except Exception as e:
utils.log_err(e)
return obj_port_list
def do_delete(vol_id, session=None):
storage = get_cinder_client(session)
try:
storage.volumes.delete(vol_id)
except Exception as e:
utils.log_err("Volumes Not Delete : " + str(e))
return 0
else:
return 1
def show_router(routers, session=None):
neutron = get_neutron_client(session)
try:
obj_router = neutron.show_router(routers)
except Exception as e:
utils.log_err(e)
return obj_router
def dump_session(sess):
temp = utils.tmp_dir()
try:
with open("{}/session.pkl".format(temp), "wb") as f:
dill.dump(sess, f)
except:
utils.log_err("Dump session failed")