diff --git a/IM/InfrastructureInfo.py b/IM/InfrastructureInfo.py index 8e541e69c..1fd48555d 100644 --- a/IM/InfrastructureInfo.py +++ b/IM/InfrastructureInfo.py @@ -357,6 +357,9 @@ def update_radl(self, radl, deployed_vms, warn=True): with self._lock: original_radl = self.radl.clone() + # Add the description field + if radl.description: + self.radl.add(radl.description.clone(), "replace") # Add new networks ad ansible_hosts only for s in radl.networks + radl.ansible_hosts: if not self.radl.add(s.clone(), "ignore") and warn: diff --git a/IM/InfrastructureManager.py b/IM/InfrastructureManager.py index 06bf8dadf..b378cc09c 100644 --- a/IM/InfrastructureManager.py +++ b/IM/InfrastructureManager.py @@ -28,6 +28,7 @@ from IM.VMRC import VMRC from IM.AppDBIS import AppDBIS +from IM.AppDB import AppDB from IM.CloudInfo import CloudInfo from IM.auth import Authentication from IM.recipe import Recipe @@ -1464,6 +1465,40 @@ def gen_auth_from_appdb(auth): InfrastructureManager.logger.error("Error getting auth data from AppDBIS: %s" % sites) return auth + @staticmethod + def translate_egi_to_ost(auth): + # Gen OST auth for all EGI auth sent + res = [] + for auth_item in auth.auth_list: + if auth_item.get('type') == "EGI": + if 'host' in auth_item and 'vo' in auth_item and 'token' in auth_item: + ost_auth = {'id': auth_item['id'], 'type': 'OpenStack', 'username': 'egi.eu', 'tenant': 'openid', + 'password': auth_item['token'], 'auth_version': '3.x_oidc_access_token', + 'vo': auth_item['vo']} + site_id = AppDB.get_site_id(auth_item["host"], stype="openstack") + site_url = AppDB.get_site_url(site_id) + if not site_url: + InfrastructureManager.logger.error("Site name '%s' not found at AppDB." % auth_item['host']) + continue + ost_auth['host'] = site_url + projects = AppDB.get_project_ids(site_id) + # If the VO does not appear in the project IDs + if auth_item['vo'] in projects: + ost_auth['domain'] = projects[auth_item['vo']] + else: + # let's use the VO name directly + ost_auth['domain'] = auth_item['vo'] + + if 'api_version' in auth_item: + ost_auth['api_version'] = auth_item['api_version'] + + res.append(ost_auth) + else: + res.append(auth_item) + + auth.auth_list = res + return auth + @staticmethod def check_auth_data(auth): # First check if it is configured to check the users from a list @@ -1505,6 +1540,7 @@ def check_auth_data(auth): # We have to check if TTS is needed for other auth item auth = InfrastructureManager.get_auth_from_vault(auth) auth = InfrastructureManager.gen_auth_from_appdb(auth) + auth = InfrastructureManager.translate_egi_to_ost(auth) return auth @staticmethod diff --git a/IM/connectors/EGI.py b/IM/connectors/EGI.py deleted file mode 100644 index c571094e6..000000000 --- a/IM/connectors/EGI.py +++ /dev/null @@ -1,209 +0,0 @@ -# IM - Infrastructure Manager -# Copyright (C) 2011 - GRyCAP - Universitat Politecnica de Valencia -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - - -from IM.connectors.OpenStack import OpenStackCloudConnector -from IM.AppDB import AppDB -from IM.auth import Authentication -from IM.CloudInfo import CloudInfo -try: - from urlparse import urlparse -except ImportError: - from urllib.parse import urlparse - - -class EGICloudConnector(OpenStackCloudConnector): - """ - Cloud Launcher to EGI using LibCloud - """ - - type = "EGI" - """str with the name of the provider.""" - - def __init__(self, cloud_info, inf): - self.egi_auth = None - self.ost_auth = None - OpenStackCloudConnector.__init__(self, cloud_info, inf) - - def get_egi_auth(self, auths): - """ - Get a compatible auth data. - The auth must have the same vo. - - Arguments: - - auth(Authentication): parsed authentication tokens. - - Returns: a :py:class:`IM.auth.Authentication` - """ - for auth in auths: - auth_vo = auth['vo'] if 'vo' in auth else None - cloud_vo = self.cloud.extra['vo'] if 'vo' in self.cloud.extra else None - - if auth_vo and auth_vo == cloud_vo: - return auth - - raise Exception("No compatible EGI auth data has been specified (check VO).") - - def get_ost_auth(self, auth): - if self.ost_auth: - return self.ost_auth - else: - if 'host' in auth and 'vo' in auth and 'token' in auth: - ost_auth = {'id': auth['id'], 'type': 'OpenStack', 'username': 'egi.eu', 'tenant': 'openid', - 'password': auth['token'], 'auth_version': '3.x_oidc_access_token', 'vo': auth['vo']} - site_id = AppDB.get_site_id(auth["host"], stype="openstack") - site_url = AppDB.get_site_url(site_id) - if not site_url: - raise Exception("Invalid site name '%s'. Not found at AppDB." % auth['host']) - ost_auth['host'] = site_url - projects = AppDB.get_project_ids(site_id) - # If the VO does not appear in the project IDs - if auth['vo'] in projects: - ost_auth['domain'] = projects[auth['vo']] - else: - # let's use the VO name directly - ost_auth['domain'] = auth['vo'] - - if 'api_version' in auth: - ost_auth['api_version'] = auth['api_version'] - - self.ost_auth = ost_auth - return ost_auth - else: - self.log_error("No correct auth data has been specified to EGI: host, vo, and token") - return None - - def get_driver(self, auth_data): - """ - Get the driver from the auth data - - Arguments: - - auth(Authentication): parsed authentication tokens. - - Returns: a :py:class:`libcloud.compute.base.NodeDriver` or None in case of error - """ - auths = auth_data.getAuthInfo(self.type, self.cloud.server) - if not auths: - # If there are no EGI credentials - # Try with the equivalent OpenStack ones - if self.ost_auth: - ost_auth = Authentication([self.ost_auth]) - ost_cloud = CloudInfo.get_cloud_list(ost_auth)[0] - auths = auth_data.getAuthInfo(OpenStackCloudConnector.type, ost_cloud.server) - if not auths: - raise Exception("No auth data has been specified to EGI.") - else: - orig_cloud = self.cloud - self.cloud = ost_cloud - auth = self.get_auth(auths) - self.type = OpenStackCloudConnector.type - driver = OpenStackCloudConnector.get_driver(self, ost_auth) - self.type = EGICloudConnector.type - self.cloud = orig_cloud - return driver - else: - auth = self.get_egi_auth(auths) - - if self.driver and self.egi_auth.compare(auth_data, self.type, self.cloud.server): - return self.driver - else: - self.egi_auth = auth_data - - if 'host' in auth and 'vo' in auth and 'token' in auth: - new_auth = Authentication([self.get_ost_auth(auth)]) - - orig_cloud = self.cloud - self.cloud = CloudInfo.get_cloud_list(new_auth)[0] - self.type = OpenStackCloudConnector.type - driver = OpenStackCloudConnector.get_driver(self, new_auth) - self.type = EGICloudConnector.type - self.cloud = orig_cloud - - self.driver = driver - return driver - else: - self.log_error("No correct auth data has been specified to EGI: host, vo, and token") - raise Exception("No correct auth data has been specified to EG: host, vo, and token") - - def concrete_system(self, radl_system, str_url, auth_data): - url = urlparse(str_url) - protocol = url[0] - src_host = url[1].split(':')[0] - vo = url[4] - - if protocol in ["ost", "appdb"] and self.cloud.server and not self.cloud.protocol: - site_host = "" - if protocol == "ost": - site_id = AppDB.get_site_id(self.cloud.server, stype="openstack") - site_url = AppDB.get_site_url(site_id, stype="openstack") - if site_url: - site_host = urlparse(site_url)[1].split(':')[0] - elif not url[2]: - # in case if appdb url without setting the site - src_host = self.cloud.server - - if ((protocol == "ost" and site_host == src_host) or - (protocol == "appdb" and src_host == self.cloud.server)): - driver = self.get_driver(auth_data) - - # In AppDB case also check the vo name, if set in the url - if protocol == "appdb" and vo: - auths = auth_data.getAuthInfo(self.type, self.cloud.server) - if not auths: - raise Exception("No auth data has been specified to EGI.") - else: - auth = self.get_egi_auth(auths) - if auth['vo'] != vo: - return None - - vo = self.get_vo_name(auth_data) - - if protocol == "appdb": - site_url, image_id, _ = AppDB.get_image_data(str_url, "openstack", vo, site=self.cloud.server) - if not image_id: - return None - - res_system = radl_system.clone() - instance_type = self.get_instance_type(driver, res_system) - if not instance_type: - return None - self.update_system_info_from_instance(res_system, instance_type) - - if vo: - res_system.setValue("disk.0.os.image.vo", vo) - - username = res_system.getValue('disk.0.os.credentials.username') - if not username: - res_system.setValue('disk.0.os.credentials.username', self.DEFAULT_USER) - - return res_system - else: - return None - else: - return None - - @staticmethod - def _get_tenant_id(driver, auth): - """ - Workaround function to get tenant id from tenant name - """ - if 'auth_version' in auth and auth['auth_version'] == '3.x_oidc_access_token': - return auth['domain'] - else: - if 'tenant_id' in auth: - return auth['tenant_id'] - else: - return auth['tenant'] diff --git a/IM/connectors/OpenStack.py b/IM/connectors/OpenStack.py index c57301a76..00531ee51 100644 --- a/IM/connectors/OpenStack.py +++ b/IM/connectors/OpenStack.py @@ -364,6 +364,13 @@ def concrete_system(self, radl_system, str_url, auth_data): if not username: res_system.setValue('disk.0.os.credentials.username', self.DEFAULT_USER) + auths = auth_data.getAuthInfo(self.type, self.cloud.server) + if auths: + auth = self.get_auth(auths) + if auth.get("vo"): + res_system.addFeature(Feature("provider.vo", "=", auth.get("vo")), + conflict="other", missing="other") + return res_system else: return None @@ -2023,7 +2030,7 @@ def list_images(self, auth_data, filters=None): return images @staticmethod - def _get_tenant_id(driver, auth): + def _get_tenant_id(auth): """ Workaround function to get tenant id from tenant name """ @@ -2039,7 +2046,7 @@ def _get_tenant_id(driver, auth): def get_quotas(self, auth_data): driver = self.get_driver(auth_data) - tenant_id = self._get_tenant_id(driver, auth_data.getAuthInfo(self.type, self.cloud.server)[0]) + tenant_id = self._get_tenant_id(auth_data.getAuthInfo(self.type, self.cloud.server)[0]) quotas = driver.ex_get_quota_set(tenant_id) try: net_quotas = driver.ex_get_network_quotas(tenant_id) diff --git a/IM/connectors/__init__.py b/IM/connectors/__init__.py index ec08cd406..7b46a70c8 100644 --- a/IM/connectors/__init__.py +++ b/IM/connectors/__init__.py @@ -16,5 +16,5 @@ __all__ = ['CloudConnector', 'EC2', 'OCCI', 'OpenNebula', 'OpenStack', 'Docker', 'GCE', 'FogBow', - 'Azure', 'DeployedNode', 'Kubernetes', 'Dummy', 'vSphere', 'CloudStack', 'Linode', 'Orange', 'EGI', + 'Azure', 'DeployedNode', 'Kubernetes', 'Dummy', 'vSphere', 'CloudStack', 'Linode', 'Orange', 'OSCAR', 'Lambda'] diff --git a/README.md b/README.md index e2371c84d..695006022 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,7 @@ [![Codacy Badge](https://app.codacy.com/project/badge/Coverage/582a0d6e763f44bdade11133e5191439)](https://www.codacy.com/gh/grycap/im/dashboard?utm_source=github.com&utm_medium=referral&utm_content=grycap/im&utm_campaign=Badge_Coverage) [![License](https://img.shields.io/badge/license-GPL%20v3.0-brightgreen.svg)](LICENSE) [![Docs](https://img.shields.io/badge/docs-latest-brightgreen.svg)](https://imdocs.readthedocs.io/en/latest/) +[![SQAaaS badge](https://img.shields.io/badge/sqaaas%20software-gold-yellow)](https://eu.badgr.com/public/assertions/rXQbYl_9SiCN7P-QCdu7Ig) IM is a tool that deploys complex and customized virtual infrastructures on IaaS Cloud deployments (such as AWS, OpenStack, etc.). It eases the access and @@ -30,6 +31,14 @@ Miguel Caballer, Ignacio Blanquer, German Molto, and Carlos de Alfonso. Journal of Grid Computing, Volume 13, Issue 1, Pages 53-70, 2015, ISSN 1570-7873, DOI: 10.1007/s10723-014-9296-5. +## Achievements + +[![SQAaaS badge](https://github.com/EOSC-synergy/SQAaaS/raw/master/badges/badges_150x116/badge_software_gold.png)](https://eu.badgr.com/public/assertions/rXQbYl_9SiCN7P-QCdu7Ig "SQAaaS gold badge achieved") + +This software has received a gold badge according to the +[Software Quality Baseline criteria](https://github.com/indigo-dc/sqa-baseline) +defined by the [EOSC-Synergy](https://www.eosc-synergy.eu) project. + ## 1 DOCKER IMAGE (Recommended Option) The recommended option to use the Infrastructure Manager service is using the diff --git a/test/unit/connectors/EGI.py b/test/unit/connectors/EGI.py deleted file mode 100644 index cdfc0f75a..000000000 --- a/test/unit/connectors/EGI.py +++ /dev/null @@ -1,295 +0,0 @@ -#! /usr/bin/env python -# -# IM - Infrastructure Manager -# Copyright (C) 2011 - GRyCAP - Universitat Politecnica de Valencia -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -import sys -import unittest - -sys.path.append(".") -sys.path.append("..") -from .CloudConn import TestCloudConnectorBase -from IM.CloudInfo import CloudInfo -from IM.auth import Authentication -from radl import radl_parse -from IM.InfrastructureInfo import InfrastructureInfo -from IM.connectors.EGI import EGICloudConnector -from mock import patch, MagicMock - - -class TestEGIConnector(TestCloudConnectorBase): - """ - Class to test the IM connectors - """ - - def setUp(self): - self.error_in_create = True - TestCloudConnectorBase.setUp(self) - - @staticmethod - def get_egi_cloud(): - cloud_info = CloudInfo() - cloud_info.type = "EGI" - cloud_info.server = "CESGA" - cloud_info.extra['vo'] = "vo.access.egi.eu" - inf = MagicMock() - inf.id = "1" - one_cloud = EGICloudConnector(cloud_info, inf) - return one_cloud - - @patch('libcloud.compute.drivers.openstack.OpenStackNodeDriver') - @patch('IM.connectors.EGI.AppDB') - def test_10_concrete(self, appdb, get_driver): - radl_data = """ - network net () - system test ( - cpu.arch='x86_64' and - cpu.count>=1 and - memory.size>=512m and - net_interface.0.connection = 'net' and - net_interface.0.dns_name = 'test' and - disk.0.os.name = 'linux' and - disk.0.image.url = 'ost://site.com/image-id' and - disk.0.os.credentials.username = 'user' - )""" - radl = radl_parse.parse_radl(radl_data) - radl_system = radl.systems[0] - - auth = Authentication([{'id': 'egi1', 'type': 'EGI', 'host': 'CESGA', - 'vo': 'vo.access.egi.eu', 'token': 'access_token'}]) - egi_cloud = self.get_egi_cloud() - - driver = MagicMock() - get_driver.return_value = driver - - appdb.get_image_data.return_value = "", "imageid", "" - appdb.get_site_id.return_value = "site1" - appdb.get_site_url.return_value = "https://site.com:5000/v3" - appdb.get_project_ids.return_value = {"vo.access.egi.eu": "projectid"} - - node_size = MagicMock() - node_size.id = '1' - node_size.ram = 512 - node_size.price = 1 - node_size.disk = 1 - node_size.vcpus = 1 - node_size.name = "g.small" - node_size.extra = {'pci_passthrough:alias': 'GPU:2,FPGA:1'} - node_size2 = MagicMock() - node_size.id = '2' - node_size2.ram = 512 - node_size2.price = 1 - node_size2.disk = 1 - node_size2.vcpus = 1 - node_size2.name = "small" - node_size2.extra = {} - driver.list_sizes.return_value = [node_size, node_size2] - driver.ex_get_size_extra_specs.return_value = {} - - concrete = egi_cloud.concreteSystem(radl_system, auth) - self.assertEqual(len(concrete), 1) - self.assertEqual(concrete[0].getValue("instance_type"), "small") - self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue()) - - radl_system.setValue('disk.0.image.url', 'appdb://image_apc_name') - concrete = egi_cloud.concreteSystem(radl_system, auth) - self.assertEqual(len(concrete), 1) - self.assertEqual(concrete[0].getValue("instance_type"), "small") - self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue()) - - radl_system.setValue('disk.0.image.url', 'appdb://CESGA/image_apc_name?vo_name') - concrete = egi_cloud.concreteSystem(radl_system, auth) - self.assertEqual(len(concrete), 0) - self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue()) - - radl_system.setValue('disk.0.image.url', 'appdb://CESGA/image_apc_name?vo.access.egi.eu') - concrete = egi_cloud.concreteSystem(radl_system, auth) - self.assertEqual(len(concrete), 1) - self.assertNotIn("ERROR", self.log.getvalue(), msg="ERROR found in log: %s" % self.log.getvalue()) - - def create_node(self, **kwargs): - """ - Create VMs returning error only first time - """ - if self.error_in_create: - self.error_in_create = False - raise Exception("Error creating VM") - else: - node = MagicMock() - node.id = "ost1" - node.name = "ost1name" - return node - - @patch('libcloud.compute.drivers.openstack.OpenStackNodeDriver') - @patch('IM.InfrastructureList.InfrastructureList.save_data') - @patch('IM.AppDB.AppDB.get_image_data') - @patch('IM.connectors.EGI.AppDB') - def test_20_launch(self, appdb, get_image_data, save_data, get_driver): - radl_data = """ - network net1 (outbound = 'yes' and provider_id = 'public' and - outports = '8080,9000:9100' and sg_name= 'test') - network net2 (dnsserver='1.1.1.1' and create = 'yes') - system test ( - cpu.arch='x86_64' and - cpu.count=1 and - memory.size=512m and - instance_tags='key=value,key1=value2' and - net_interface.1.connection = 'net1' and - net_interface.0.connection = 'net2' and - disk.0.os.name = 'linux' and - disk.0.image.url = 'appdb://apc_image_name' and - disk.0.os.credentials.username = 'user' and - disk.1.size=1GB and - disk.1.device='hdb' and - disk.2.image.url = 'ost://server.com/vol-id' and - disk.2.device='hdc' - )""" - radl = radl_parse.parse_radl(radl_data) - radl.check() - - auth = Authentication([{'id': 'egi1', 'type': 'EGI', 'host': 'CESGA', - 'vo': 'vo.access.egi.eu', 'token': 'access_token'}, - {'type': 'InfrastructureManager', 'username': 'user', - 'password': 'pass'}]) - egi_cloud = self.get_egi_cloud() - - driver = MagicMock() - get_driver.return_value = driver - - get_image_data.return_value = "", "imageid", "" - appdb.get_site_id.return_value = "site1" - appdb.get_site_url.return_value = "https://site.com:5000/v3" - appdb.get_project_ids.return_value = {"vo.access.egi.eu": "projectid"} - - node_size = MagicMock() - node_size.ram = 512 - node_size.price = 1 - node_size.disk = 1 - node_size.vcpus = 1 - node_size.name = "small" - driver.list_sizes.return_value = [node_size] - - net1 = MagicMock() - net1.name = "public" - net1.id = "net1id" - net1.extra = {'router:external': True} - net1.cidr = None - net2 = MagicMock() - net2.name = "private" - net2.id = "net2id" - net2.cidr = "10.0.0.0/24" - driver.ex_list_networks.return_value = [net2, net1] - - sg = MagicMock() - sg.name = "sg" - driver.ex_create_security_group.return_value = sg - driver.ex_list_security_groups.return_value = [] - driver.ex_create_security_group_rule.return_value = True - - driver.features = {'create_node': ['ssh_key']} - - driver.create_node.side_effect = self.create_node - - driver.ex_create_network.return_value = net2 - subnet1 = MagicMock() - driver.ex_create_subnet.return_value = subnet1 - - router = MagicMock() - router.id = "id" - router.name = "name" - router.extra = {'external_gateway_info': {'network_id': net1.id}} - driver.ex_list_routers.return_value = [router] - driver.ex_add_router_subnet.return_value = True - - image = MagicMock() - image.id = 'imageid' - driver.get_image.return_value = image - vol = MagicMock() - vol.id = 'volid' - driver.ex_get_volume.return_value = vol - - inf = InfrastructureInfo() - inf.radl = radl - inf.auth = auth - res = egi_cloud.launch_with_retry(inf, radl, radl, 1, auth, 2, 1) - success, _ = res[0] - self.assertTrue(success, msg="ERROR: launching a VM.") - self.assertEqual(driver.create_node.call_args_list[0][1]['networks'], [net1]) - mappings = [ - {'source_type': 'image', - 'uuid': 'imageid', - 'boot_index': 0, - 'delete_on_termination': False}, - {'guest_format': 'ext3', - 'boot_index': 1, - 'volume_size': 1, - 'device_name': 'vdb', - 'source_type': 'blank', - 'destination_type': 'volume', - 'delete_on_termination': True}, - {'boot_index': 2, - 'delete_on_termination': False, - 'destination_type': 'volume', - 'device_name': 'vdc', - 'source_type': 'volume', - 'uuid': 'volid'} - ] - self.assertEqual(driver.create_node.call_args_list[0][1]['ex_blockdevicemappings'], mappings) - self.assertEqual(driver.ex_create_subnet.call_args_list[0][0][2], "10.0.1.0/24") - - @patch('libcloud.compute.drivers.openstack.OpenStackNodeDriver') - @patch('IM.connectors.EGI.AppDB') - def test_get_driver(self, appdb, get_driver): - auth = Authentication([{'id': 'egi1', 'type': 'EGI', 'host': 'CESGA', - 'vo': 'vo.access.egi.eu', 'token': 'access_token'}, - {'type': 'InfrastructureManager', 'username': 'user', - 'password': 'pass'}]) - egi_cloud = self.get_egi_cloud() - - appdb.get_site_id.return_value = "site1" - appdb.get_site_url.return_value = "https://site.com:5000/v3" - appdb.get_project_ids.return_value = {"vo.access.egi.eu": "projectid"} - - egi_cloud.get_driver(auth) - - egi_cloud.cloud.extra['vo'] = 'other_vo' - with self.assertRaises(Exception) as ex: - egi_cloud.get_driver(auth) - self.assertEqual('No compatible EGI auth data has been specified (check VO).', - str(ex.exception)) - - egi_cloud.driver = None - auth = Authentication([{'id': 'egi1', 'type': 'EGI', 'host': 'CESGA', - 'vo': 'vo.access.egi.eu', 'token': 'access_token'}, - {'id': 'egi2', 'type': 'EGI', 'host': 'CESGA', - 'vo': 'other_vo', 'token': 'access_token'}, - {'type': 'InfrastructureManager', 'username': 'user', - 'password': 'pass'}]) - egi_cloud.get_driver(auth) - self.assertEqual(appdb.get_site_url.call_count, 1) - - egi_cloud.driver = None - auth = Authentication([{'id': 'ost1', 'type': 'OpenStack', 'host': 'https://site.com:5000', - 'username': 'egi.eu', 'password': 'access_token', 'tenant': 'openid', - 'domain': 'projectid', 'auth_version': '3.x_oidc_access_token'}, - {'type': 'InfrastructureManager', 'username': 'user', - 'password': 'pass'}]) - egi_cloud.get_driver(auth) - self.assertEqual(appdb.get_site_url.call_count, 1) - - -if __name__ == '__main__': - unittest.main() diff --git a/test/unit/test_im_logic.py b/test/unit/test_im_logic.py old mode 100755 new mode 100644 index cf4c53f27..71cd6c74c --- a/test/unit/test_im_logic.py +++ b/test/unit/test_im_logic.py @@ -1491,6 +1491,20 @@ def test_search_vm(self, dummycc): self.assertEqual(res[1].name, "s0") self.assertEqual(res[1].getValue("disk.0.image.url"), "imageuri3") + @patch('IM.InfrastructureManager.AppDB') + def test_translate_egi_to_ost(self, appdb): + appdb.get_site_id.return_value = 'site_id' + appdb.get_site_url.return_value = 'https://ostsite.com:5000' + appdb.get_project_ids.return_value = {'vo_name': 'projectid'} + + auth = Authentication([{'id': 'egi', 'type': 'EGI', 'host': 'SITE_NAME', 'vo': 'vo_name', 'token': 'atoken'}, + {'type': 'InfrastructureManager', 'token': 'atoken'}]) + res = IM.translate_egi_to_ost(auth) + self.assertIn({'id': 'egi', 'type': 'OpenStack', 'username': 'egi.eu', 'password': 'atoken', + 'tenant': 'openid', 'auth_version': '3.x_oidc_access_token', 'vo': 'vo_name', + 'host': 'https://ostsite.com:5000', 'domain': 'projectid'}, res.auth_list) + self.assertIn({'type': 'InfrastructureManager', 'token': 'atoken'}, res.auth_list) + if __name__ == "__main__": unittest.main()