renamed test file to lib_edit_test.py
Project: http://git-wip-us.apache.org/repos/asf/libcloud/repo Commit: http://git-wip-us.apache.org/repos/asf/libcloud/commit/b3fa472e Tree: http://git-wip-us.apache.org/repos/asf/libcloud/tree/b3fa472e Diff: http://git-wip-us.apache.org/repos/asf/libcloud/diff/b3fa472e Branch: refs/heads/trunk Commit: b3fa472e5ab7d952e9034eb29c4a8c5acc21599b Parents: 3ebd1db Author: mitch <[email protected]> Authored: Wed Aug 15 11:34:01 2018 -0400 Committer: mitch <[email protected]> Committed: Wed Aug 15 11:34:01 2018 -0400 ---------------------------------------------------------------------- libcloud/compute/drivers/nttcis.py | 180 ++++++++++++++++++---- tests/lib_edit_test.py | 97 ++++++++++++ tests/lib_list_test.py | 254 ++++++++++++++++++++++++++++++++ tests/libtest.py | 247 ------------------------------- 4 files changed, 501 insertions(+), 277 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/libcloud/blob/b3fa472e/libcloud/compute/drivers/nttcis.py ---------------------------------------------------------------------- diff --git a/libcloud/compute/drivers/nttcis.py b/libcloud/compute/drivers/nttcis.py index 42272a3..5775e63 100644 --- a/libcloud/compute/drivers/nttcis.py +++ b/libcloud/compute/drivers/nttcis.py @@ -861,6 +861,21 @@ class NttCisNodeDriver(NodeDriver): ).object ) + def list_snapshot_windows(self, location: str, plan: str) -> list: + """ + List snapshot windows in a given location + :param location: a location object or location id such as "NA9" + :param plan: 'ESSENTIALS' or 'ADVANCED' + :return: dictionary with keys id, day_of_week, start_hour, availability + :rtype: dict + """ + params = {} + params['datacenterId'] = self._location_to_location_id(location) + params['servicePlan'] = plan + return self._to_windows(self.connection.request_with_orgId_api_2( + 'infrastructure/snapshotWindow', + params=params).object) + def list_networks(self, location=None): """ List networks deployed across all data center locations for your @@ -1070,7 +1085,22 @@ class NttCisNodeDriver(NodeDriver): nodes_obj = self._list_nodes_single_page(params) yield self._to_nodes(nodes_obj) - def ex_start_node(self, node): + def ex_edit_metadata(self, node: Node, name: str = None, description: str = None, drs_eligible: str = None) -> bool: + request_elem = ET.Element('editServerMetadata', {'xmlns': TYPES_URN, 'id': node.id}) + if name is not None: + ET.SubElement(request_elem, 'name').text = name + if description is not None: + ET.SubElement(request_elem, 'description').text = description + if drs_eligible is not None: + ET.SubElement(request_elem, 'drsEligible').text = drs_eligible + body = self.connection.request_with_orgId_api_2( + 'server/editServerMetadata', + method="POST", + data=ET.tostring(request_elem)).object + response_code = findtext(body, 'responseCode', TYPES_URN) + return response_code in ['IN_PROGRESS', 'OK'] + + def ex_start_node(self, node: Node) -> bool: """ Powers on an existing deployed server @@ -1088,7 +1118,7 @@ class NttCisNodeDriver(NodeDriver): response_code = findtext(body, 'responseCode', TYPES_URN) return response_code in ['IN_PROGRESS', 'OK'] - def ex_shutdown_graceful(self, node): + def ex_shutdown_graceful(self, node: Node) -> bool: """ This function will attempt to "gracefully" stop a server by initiating a shutdown sequence within the guest operating system. @@ -1214,6 +1244,70 @@ class NttCisNodeDriver(NodeDriver): response_code = findtext(body, 'result', GENERAL_NS) return response_code in ['IN_PROGRESS', 'SUCCESS'] + def ex_enable_snapshots(self, node: str, window: str, plan: str='ADVANCED', initiate: str = 'true') -> bool: + update_node = ET.Element('enableSnapshotService', + {'xmlns': TYPES_URN}) + window_id = window + plan = plan + update_node.set('serverId', node) + ET.SubElement(update_node, 'servicePlan').text = plan + ET.SubElement(update_node, 'windowId').text = window_id + ET.SubElement(update_node, 'initiateManualSnapshot').text = initiate + result = self.connection.request_with_orgId_api_2( + 'snapshot/enableSnapshotService', + method='POST', + data=ET.tostring(update_node)).object + + response_code = findtext(result, 'responseCode', TYPES_URN) + return response_code in ['IN_PROGRESS', 'OK'] + + def list_snapshots(self, node: Node) -> list: + params = {} + params['serverId'] = self.list_nodes(ex_name=node)[0].id + return self._to_snapshots(self.connection.request_with_orgId_api_2( + 'snapshot/snapshot', + params=params).object) + + def ex_disable_snapshots(self, node: str) -> bool: + update_node = ET.Element('disableSnapshotService', + {'xmlns': TYPES_URN}) + + update_node.set('serverId', node) + result = self.connection.request_with_orgId_api_2( + 'snapshot/disableSnapshotService', + method='POST', + data=ET.tostring(update_node)).object + + response_code = findtext(result, 'responseCode', TYPES_URN) + return response_code in ['IN_PROGRESS', 'OK'] + + def ex_initiate_manual_snapshot(self, name: str, server_id: str = None) -> bool: + """ + Initiate a manual snapshot on the fly + :param node: A node object from which to get the node id + + :return: True of False + :rtype: ``bool`` + """ + + if server_id is None: + node = self.list_nodes(ex_name=name) + if len(node) > 1: + raise RuntimeError("Found more than one server Id, " + "please use one the following along with name parameter: {}".format([n.id for n in node])) + else: + node = [] + node.append(self.ex_get_node_by_id(server_id)) + update_node = ET.Element('initiateManualSnapshot', + {'xmlns': TYPES_URN}) + update_node.set('serverId', node[0].id) + result = self.connection.request_with_orgId_api_2( + 'snapshot/initiateManualSnapshot', + method='POST', + data=ET.tostring(update_node)).object + response_code = findtext(result, 'responseCode', TYPES_URN) + return response_code in ['IN_PROGRESS', 'OK'] + def ex_create_anti_affinity_rule(self, node_list): """ Create an anti affinity rule given a list of nodes @@ -2484,8 +2578,8 @@ class NttCisNodeDriver(NodeDriver): response_code = findtext(result, 'result', GENERAL_NS) return response_code in ['IN_PROGRESS', 'SUCCESS'] - def ex_reconfigure_node(self, node, memory_gb, cpu_count, cores_per_socket, - cpu_performance): + def ex_reconfigure_node(self, node, memory_gb=None, cpu_count=None, cores_per_socket=None, + cpu_performance=None): """ Reconfigure the virtual hardware specification of a node @@ -2546,52 +2640,56 @@ class NttCisNodeDriver(NodeDriver): image_description = '' node_id = self._node_to_node_id(node) - + """ + Removing anything below 2.4 # Version 2.3 and lower if LooseVersion(self.connection.active_api_version) < LooseVersion( '2.4'): response = self.connection.request_with_orgId_api_1( 'server/%s?clone=%s&desc=%s' % (node_id, image_name, image_description)).object - + # Version 2.4 and higher else: - clone_server_elem = ET.Element('cloneServer', - {'id': node_id, - 'xmlns': TYPES_URN}) + """ + clone_server_elem = ET.Element('cloneServer', + {'id': node_id, + 'xmlns': TYPES_URN}) - ET.SubElement(clone_server_elem, 'imageName').text = image_name + ET.SubElement(clone_server_elem, 'imageName').text = image_name - if image_description is not None: - ET.SubElement(clone_server_elem, 'description').text = \ - image_description + if image_description is not None: + ET.SubElement(clone_server_elem, 'description').text = \ + image_description - if cluster_id is not None: - ET.SubElement(clone_server_elem, 'clusterId').text = \ - cluster_id + if cluster_id is not None: + ET.SubElement(clone_server_elem, 'clusterId').text = \ + cluster_id - if is_guest_Os_Customization is not None: - ET.SubElement(clone_server_elem, 'guestOsCustomization')\ - .text = is_guest_Os_Customization + if is_guest_Os_Customization is not None: + ET.SubElement(clone_server_elem, 'guestOsCustomization')\ + .text = is_guest_Os_Customization - if tag_key_id is not None: - tag_elem = ET.SubElement(clone_server_elem, 'tagById') - ET.SubElement(tag_elem, 'tagKeyId').text = tag_key_id - - if tag_value is not None: - ET.SubElement(tag_elem, 'value').text = tag_value + if tag_key_id is not None: + tag_elem = ET.SubElement(clone_server_elem, 'tagById') + ET.SubElement(tag_elem, 'tagKeyId').text = tag_key_id - response = self.connection.request_with_orgId_api_2( - 'server/cloneServer', - method='POST', - data=ET.tostring(clone_server_elem)).object + if tag_value is not None: + ET.SubElement(tag_elem, 'value').text = tag_value + response = self.connection.request_with_orgId_api_2( + 'server/cloneServer', + method='POST', + data=ET.tostring(clone_server_elem)).object + """ + removing references to anything lower than 2.4 # Version 2.3 and lower if LooseVersion(self.connection.active_api_version) < LooseVersion( '2.4'): response_code = findtext(response, 'result', GENERAL_NS) else: - response_code = findtext(response, 'responseCode', TYPES_URN) + """ + response_code = findtext(response, 'responseCode', TYPES_URN) return response_code in ['IN_PROGRESS', 'SUCCESS'] @@ -4470,6 +4568,28 @@ class NttCisNodeDriver(NodeDriver): state=element.get('state') ) + def _to_snapshots(self, object: ET): + snapshot_elements = object.findall(fixxpath('snapshot', TYPES_URN)) + return [self._to_snapshot(el) for el in snapshot_elements] + + def _to_snapshot(self, element: ET): + return {'id': element.get('id'), 'start_time': findtext(element, 'startTime', TYPES_URN), + 'end_time': findtext(element, 'endTime', TYPES_URN), + 'expiry_time':findtext(element, 'expiryTime', TYPES_URN), + 'type': findtext(element, 'type', TYPES_URN), + 'state': findtext(element, 'state', TYPES_URN)} + #'server_config': self.to_snapshot_conf_elems(findtext(element, 'serverConfig', TYPES_URN)} + + + + def _to_windows(self, object): + snapshot_window_elements = object.findall(fixxpath('snapshotWindow', TYPES_URN)) + return [self._to_window(el) for el in snapshot_window_elements] + + def _to_window(self, element): + return {'id': element.get('id'), 'day_of_week': element.get('dayOfWeek'), + 'start_hour': element.get('startHour'), 'availability_status': element.get('availabilityStatus')} + def _to_nodes(self, object): node_elements = object.findall(fixxpath('server', TYPES_URN)) return [self._to_node(el) for el in node_elements] http://git-wip-us.apache.org/repos/asf/libcloud/blob/b3fa472e/tests/lib_edit_test.py ---------------------------------------------------------------------- diff --git a/tests/lib_edit_test.py b/tests/lib_edit_test.py new file mode 100644 index 0000000..8c6596f --- /dev/null +++ b/tests/lib_edit_test.py @@ -0,0 +1,97 @@ +import pytest +import libcloud +from libcloud import loadbalancer +from libcloud.common.nttcis import NttCisAPIException + + +def test_disable_node_snapshot(compute_driver): + node = '040fefdb-78be-4b17-8ef9-86820bad67d9' + assert compute_driver.ex_disable_snapshots(node) is True + + +def test_list_windows(compute_driver): + loc = 'EU6' + service_plan = 'ADVANCED' + windows = compute_driver.list_snapshot_windows(loc, service_plan) + for window in windows: + print(window) + assert 'day_of_week' in window + + +def test_enable_snapshot(compute_driver): + """ + This will enable a snapshot window and create an initial + snapshot when it has done so. A node object and a window id are required + :param compute_driver: The driver object for compute nodes. + :return: True or False + :rtype: ``bool`` + """ + window_id = 'ea646520-4272-11e8-838c-180373fb68df' + node = '040fefdb-78be-4b17-8ef9-86820bad67d9' + result = compute_driver.ex_enable_snapshots(node, window_id) + assert result is True + + +def test_initiate_manual_snapshot_warn(compute_driver): + with pytest.raises(RuntimeError, match=r'Found more than one server Id .*'): + compute_driver.ex_initiate_manual_snapshot('sdk_server_1', 'dc637783-2bb2-4b92-838a-99a899b5e29b') + + +def test_initiate_manual_snapshot(compute_driver): + compute_driver.ex_initiate_manual_snapshot('sdk_server_1', 'dc637783-2bb2-4b92-838a-99a899b5e29b') + + +def test_shutdown_server_1(compute_driver): + node = compute_driver.ex_get_node_by_id('040fefdb-78be-4b17-8ef9-86820bad67d9 ') + result = compute_driver.ex_shutdown_graceful(node) + assert result is True + + +def test_start_server_1(compute_driver): + node = compute_driver.ex_get_node_by_id('040fefdb-78be-4b17-8ef9-86820bad67d9 ') + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_shutdown_server_2(compute_driver): + nodes = compute_driver.list_nodes(ex_name='sdk_server_1') + for node in nodes: + result = compute_driver.ex_shutdown_graceful(node) + assert result is True + + +def test_start_server_2(compute_driver): + nodes = compute_driver.list_nodes(ex_name='sdk_server_1') + for node in nodes: + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_edit_metadata(compute_driver): + node = compute_driver.ex_get_node_by_id('040fefdb-78be-4b17-8ef9-86820bad67d9 ') + description = 'SDK Test server' + name = 'sdk_server_1' + result = compute_driver.ex_edit_metadata(node, name=name, description=description) + assert result is True + + +def test_edit_metadata_fails(compute_driver): + node = compute_driver.ex_get_node_by_id('040fefdb-78be-4b17-8ef9-86820bad67d9 ') + description = 'Test server' + ip_address = 'EU6 Ubuntu' + with pytest.raises(TypeError): + result = compute_driver.ex_edit_metadata(node, ip_address=ip_address, description=description) + + +def test_reconfigure_node(compute_driver): + node = compute_driver.ex_get_node_by_id('040fefdb-78be-4b17-8ef9-86820bad67d9 ') + cpu_performance = 'HIGHPERFORMANCE' + result = compute_driver.ex_reconfigure_node(node, cpu_performance=cpu_performance) + assert result is True + + +def test_list_locations(compute_driver): + locations = compute_driver.list_locations() + for location in locations: + print(location) + http://git-wip-us.apache.org/repos/asf/libcloud/blob/b3fa472e/tests/lib_list_test.py ---------------------------------------------------------------------- diff --git a/tests/lib_list_test.py b/tests/lib_list_test.py new file mode 100644 index 0000000..9a090ae --- /dev/null +++ b/tests/lib_list_test.py @@ -0,0 +1,254 @@ +import pytest +import libcloud +from libcloud import loadbalancer + + +def test_list_node_all(compute_driver): + nodes = compute_driver.list_nodes() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6'], node.extra['window']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_node_location(compute_driver): + nodes = compute_driver.list_nodes(ex_location='EU6') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_node_name(compute_driver): + nodes = compute_driver.list_nodes(ex_name='sdk_server_1') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_node_ipv6(compute_driver): + nodes = compute_driver.list_nodes(ex_ipv6='2a00:47c0:111:1331:6140:e432:729b:eef6') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_node_ipv4(compute_driver): + nodes = compute_driver.list_nodes(ex_ipv4='10.1.1.6') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +""" + requires retrieving vlan Id first +""" +def test_list_node_vlan(compute_driver): + nodes = compute_driver.list_nodes(ex_vlan='eb05a24e-85a6-46e3-a7c9-f1765737476d') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +""" +Libcloud docs say this works but it is not in our API docs +def test_list_node_image(compute_driver): + nodes = compute_driver.list_nodes(ex_image='46096745-5a89-472b-9b3b-89a6a07bb60b') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 +""" + + +def test_list_node_started(compute_driver): + nodes = compute_driver.list_nodes(ex_started='true') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_node_deployed(compute_driver): + nodes = compute_driver.list_nodes(ex_deployed='true') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_node_state(compute_driver): + nodes = compute_driver.list_nodes(ex_state='NORMAL') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_netwokr_domain_id(compute_driver): + nodes = compute_driver.list_nodes(ex_network_domain='6aafcf08-cb0b-432c-9c64-7371265db086') + print() + for node in nodes: + print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], + node.private_ips, node.extra['ipv6']) + assert isinstance(nodes, list) and len(nodes) > 0 + + +def test_list_vlans(compute_driver): + vlans = compute_driver.ex_list_vlans() + print() + for vlan in vlans: + print(vlan.id, vlan.name, vlan.location.id, vlan.ipv4_gateway, vlan.ipv6_gateway, vlan.ipv6_range_address, vlan.ipv6_range_size, + vlan.private_ipv4_range_address, vlan.private_ipv4_range_size, vlan.status) + assert isinstance(vlans, list) and len(vlans) > 0 + + +def test_list_vlan(compute_driver): + vlan = compute_driver.ex_get_vlan('eb05a24e-85a6-46e3-a7c9-f1765737476d') + print() + print(vlan.id, vlan.name, vlan.location.id, vlan.ipv4_gateway, vlan.ipv6_gateway, vlan.ipv6_range_address, vlan.ipv6_range_size, + vlan.private_ipv4_range_address, vlan.private_ipv4_range_size, vlan.status) + assert vlan.name == 'sdk_vlan1' + + +def test_list_firewall_rules(compute_driver): + rules = compute_driver.ex_list_firewall_rules('6aafcf08-cb0b-432c-9c64-7371265db086') + print() + for rule in rules: + print("id {}, name {}, action {}. location {}, ip ver {}, protocol {}, any ip {}, ip {}, prefix {}," + " port range {} {} , src address {}, src port list {}, dest. any__ip {}, dest address {}, " + "dest prefix {}, dest port range {} {}, dest address list id {}" + ", dest port list id {}".format( + rule.id, rule.name, rule.action, + rule.location.name, rule.ip_version, + rule.protocol, rule.source.any_ip, + rule.source.ip_address, + rule.source.ip_prefix_size, + rule.source.port_begin, rule.source.port_end, + rule.source.address_list_id, + rule.source.port_list_id, + rule.destination.any_ip, + rule.destination.ip_address, + rule.destination.ip_prefix_size, + rule.destination.port_begin, + rule.destination.port_end, + rule.destination.address_list_id, + rule.destination.port_list_id, + )) + + +def test_list_address_lists(compute_driver): + address_lists = compute_driver.ex_list_ip_address_list('6aafcf08-cb0b-432c-9c64-7371265db086') + print() + for address_list in address_lists: + print(address_list) + assert isinstance(address_lists, list) and len(address_lists) > 0 + + +def test_list_port_lists(compute_driver): + port_lists = compute_driver.ex_list_portlist('6aafcf08-cb0b-432c-9c64-7371265db086') + print() + for portlist in port_lists: + print(portlist) + assert isinstance(port_lists, list) and len(port_lists) > 0 + + +def test_list_nat_rules(compute_driver): + nat_rules = compute_driver.ex_list_nat_rules(compute_driver.ex_get_network_domain('6aafcf08-cb0b-432c-9c64-7371265db086')) + print() + for nat_rule in nat_rules: + print(nat_rule, nat_rule.external_ip, nat_rule.internal_ip) + assert isinstance(nat_rules, list) and len(nat_rules) > 0 + + +def test_list_balancers(lbdriver): + balancers = lbdriver.list_balancers(ex_network_domain_id="6aafcf08-cb0b-432c-9c64-7371265db086") + print() + for balancer in balancers: + print(balancer.id, balancer.ip, balancer.name, balancer.port) + assert isinstance(balancers, list) + + +def test_get_listener(lbdriver): + listener = lbdriver.get_balancer("59abe126-2bba-48ac-8616-1aba51aabac5") + print() + print(listener.ip, listener.name, listener.port) + assert listener.ip == '168.128.13.127' + + +def test_vip_nodes(lbdriver): + vips = lbdriver.ex_get_nodes("6aafcf08-cb0b-432c-9c64-7371265db086") + print() + for vip in vips: + print(vip, vip.ip, vip.name) + assert isinstance(vips, list) and len(vips) > 0 + + +def test_list_lb_pools(lbdriver): + pools = lbdriver.ex_get_pools(ex_network_domain_id="6aafcf08-cb0b-432c-9c64-7371265db086") + print() + for pool in pools: + print(pool.id, pool.name, pool.description, pool.health_monitor_id, pool.load_balance_method, pool.slow_ramp_time, pool.status) + assert isinstance(pools, list) + + +def test_list_lb_pool_members(lbdriver): + balancer = lbdriver.get_balancer("59abe126-2bba-48ac-8616-1aba51aabac5") + pool_members = lbdriver.balancer_list_members(balancer) + print() + for pool_member in pool_members: + print(pool_member) + assert isinstance(pool_members, list) + + +def test_get_pool_member(lbdriver): + pool_member = lbdriver.ex_get_pool_member("9382e488-7f95-4db0-b2de-0b807aab825b") + print() + print(pool_member.ip, pool_member.port, pool_member.name) + assert pool_member.ip == '10.1.1.8' + + +def test_get_node(lbdriver): + node = lbdriver.ex_get_node("5c647a74-d181-4ed8-82d3-55ae443a06dd") + print() + print(node.name, node.ip, node.connection_limit, node.connection_rate_limit) + assert isinstance(node, object) + + +def test_list_snapshots(compute_driver): + snapshots = compute_driver.list_snapshots('web1') + for snapshot in snapshots: + print(snapshot) + assert 'expiry_time' in snapshot + + +def test_images(compute_driver): + images = compute_driver.list_images() + print() + print(images) + assert isinstance(images, list) and len(images) > 0 + http://git-wip-us.apache.org/repos/asf/libcloud/blob/b3fa472e/tests/libtest.py ---------------------------------------------------------------------- diff --git a/tests/libtest.py b/tests/libtest.py deleted file mode 100644 index 4b95b6b..0000000 --- a/tests/libtest.py +++ /dev/null @@ -1,247 +0,0 @@ -import pytest -import libcloud -from libcloud import loadbalancer - - -def test_list_node_all(compute_driver): - nodes = compute_driver.list_nodes() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6'], node.extra['window']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_node_location(compute_driver): - nodes = compute_driver.list_nodes(ex_location='EU6') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_node_name(compute_driver): - nodes = compute_driver.list_nodes(ex_name='sdk_server_1') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_node_ipv6(compute_driver): - nodes = compute_driver.list_nodes(ex_ipv6='2a00:47c0:111:1331:6140:e432:729b:eef6') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_node_ipv4(compute_driver): - nodes = compute_driver.list_nodes(ex_ipv4='10.1.1.6') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -""" - requires retrieving vlan Id first -""" -def test_list_node_vlan(compute_driver): - nodes = compute_driver.list_nodes(ex_vlan='eb05a24e-85a6-46e3-a7c9-f1765737476d') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -""" -Libcloud docs say this works but it is not in our API docs -def test_list_node_image(compute_driver): - nodes = compute_driver.list_nodes(ex_image='46096745-5a89-472b-9b3b-89a6a07bb60b') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 -""" - - -def test_list_node_started(compute_driver): - nodes = compute_driver.list_nodes(ex_started='true') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_node_deployed(compute_driver): - nodes = compute_driver.list_nodes(ex_deployed='true') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_node_state(compute_driver): - nodes = compute_driver.list_nodes(ex_state='NORMAL') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_netwokr_domain_id(compute_driver): - nodes = compute_driver.list_nodes(ex_network_domain='6aafcf08-cb0b-432c-9c64-7371265db086') - print() - for node in nodes: - print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], - [disk for disk in node.extra['disks']], node.extra['memoryMb'], node.extra['OS_displayName'], - node.private_ips, node.extra['ipv6']) - assert isinstance(nodes, list) and len(nodes) > 0 - - -def test_list_vlans(compute_driver): - vlans = compute_driver.ex_list_vlans() - print() - for vlan in vlans: - print(vlan.id, vlan.name, vlan.location.id, vlan.ipv4_gateway, vlan.ipv6_gateway, vlan.ipv6_range_address, vlan.ipv6_range_size, - vlan.private_ipv4_range_address, vlan.private_ipv4_range_size, vlan.status) - assert isinstance(vlans, list) and len(vlans) > 0 - - -def test_list_vlan(compute_driver): - vlan = compute_driver.ex_get_vlan('eb05a24e-85a6-46e3-a7c9-f1765737476d') - print() - print(vlan.id, vlan.name, vlan.location.id, vlan.ipv4_gateway, vlan.ipv6_gateway, vlan.ipv6_range_address, vlan.ipv6_range_size, - vlan.private_ipv4_range_address, vlan.private_ipv4_range_size, vlan.status) - assert vlan.name == 'sdk_vlan1' - - -def test_list_firewall_rules(compute_driver): - rules = compute_driver.ex_list_firewall_rules('6aafcf08-cb0b-432c-9c64-7371265db086') - print() - for rule in rules: - print("id {}, name {}, action {}. location {}, ip ver {}, protocol {}, any ip {}, ip {}, prefix {}," - " port range {} {} , src address {}, src port list {}, dest. any__ip {}, dest address {}, " - "dest prefix {}, dest port range {} {}, dest address list id {}" - ", dest port list id {}".format( - rule.id, rule.name, rule.action, - rule.location.name, rule.ip_version, - rule.protocol, rule.source.any_ip, - rule.source.ip_address, - rule.source.ip_prefix_size, - rule.source.port_begin, rule.source.port_end, - rule.source.address_list_id, - rule.source.port_list_id, - rule.destination.any_ip, - rule.destination.ip_address, - rule.destination.ip_prefix_size, - rule.destination.port_begin, - rule.destination.port_end, - rule.destination.address_list_id, - rule.destination.port_list_id, - )) - - -def test_list_address_lists(compute_driver): - address_lists = compute_driver.ex_list_ip_address_list('6aafcf08-cb0b-432c-9c64-7371265db086') - print() - for address_list in address_lists: - print(address_list) - assert isinstance(address_lists, list) and len(address_lists) > 0 - - -def test_list_port_lists(compute_driver): - port_lists = compute_driver.ex_list_portlist('6aafcf08-cb0b-432c-9c64-7371265db086') - print() - for portlist in port_lists: - print(portlist) - assert isinstance(port_lists, list) and len(port_lists) > 0 - - -def test_list_nat_rules(compute_driver): - nat_rules = compute_driver.ex_list_nat_rules(compute_driver.ex_get_network_domain('6aafcf08-cb0b-432c-9c64-7371265db086')) - print() - for nat_rule in nat_rules: - print(nat_rule, nat_rule.external_ip, nat_rule.internal_ip) - assert isinstance(nat_rules, list) and len(nat_rules) > 0 - - -def test_list_balancers(lbdriver): - balancers = lbdriver.list_balancers(ex_network_domain_id="6aafcf08-cb0b-432c-9c64-7371265db086") - print() - for balancer in balancers: - print(balancer.id, balancer.ip, balancer.name, balancer.port) - assert isinstance(balancers, list) - - -def test_get_listener(lbdriver): - listener = lbdriver.get_balancer("59abe126-2bba-48ac-8616-1aba51aabac5") - print() - print(listener.ip, listener.name, listener.port) - assert listener.ip == '168.128.13.127' - - -def test_vip_nodes(lbdriver): - vips = lbdriver.ex_get_nodes("6aafcf08-cb0b-432c-9c64-7371265db086") - print() - for vip in vips: - print(vip, vip.ip, vip.name) - assert isinstance(vips, list) and len(vips) > 0 - - -def test_list_lb_pools(lbdriver): - pools = lbdriver.ex_get_pools(ex_network_domain_id="6aafcf08-cb0b-432c-9c64-7371265db086") - print() - for pool in pools: - print(pool.id, pool.name, pool.description, pool.health_monitor_id, pool.load_balance_method, pool.slow_ramp_time, pool.status) - assert isinstance(pools, list) - - -def test_list_lb_pool_members(lbdriver): - balancer = lbdriver.get_balancer("59abe126-2bba-48ac-8616-1aba51aabac5") - pool_members = lbdriver.balancer_list_members(balancer) - print() - for pool_member in pool_members: - print(pool_member) - assert isinstance(pool_members, list) - - -def test_get_pool_member(lbdriver): - pool_member = lbdriver.ex_get_pool_member("9382e488-7f95-4db0-b2de-0b807aab825b") - print() - print(pool_member.ip, pool_member.port, pool_member.name) - assert pool_member.ip == '10.1.1.8' - - -def test_get_node(lbdriver): - node = lbdriver.ex_get_node("5c647a74-d181-4ed8-82d3-55ae443a06dd") - print() - print(node.name, node.ip, node.connection_limit, node.connection_rate_limit) - assert isinstance(node, object) - - -def test_images(compute_driver): - images = compute_driver.list_images() - print() - print(images) - assert isinstance(images, list) and len(images) > 0 -
