Test failed DRS creation not supported between the selected data centers
Project: http://git-wip-us.apache.org/repos/asf/libcloud/repo Commit: http://git-wip-us.apache.org/repos/asf/libcloud/commit/10c4a31b Tree: http://git-wip-us.apache.org/repos/asf/libcloud/tree/10c4a31b Diff: http://git-wip-us.apache.org/repos/asf/libcloud/diff/10c4a31b Branch: refs/heads/trunk Commit: 10c4a31b4ea9b9776ff8fd94553aa294ceb2a1d1 Parents: 62d9eb6 Author: mitch <[email protected]> Authored: Mon Oct 29 13:41:24 2018 -0400 Committer: mitch <[email protected]> Committed: Mon Oct 29 13:41:24 2018 -0400 ---------------------------------------------------------------------- .gitignore | 2 +- libcloud/drs/drivers/nttcis.py | 47 ++++ libcloud/utils/xml.py | 14 -- tests/conftest.py | 40 ++++ tests/lib_create_test.py | 252 ++++++++++++++++++++ tests/lib_edit_test.py | 452 ++++++++++++++++++++++++++++++++++++ tests/lib_list_test.py | 393 +++++++++++++++++++++++++++++++ tests/lib_misc_test.py | 10 + tests/test_lib_list.py | 344 --------------------------- 9 files changed, 1195 insertions(+), 359 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/.gitignore ---------------------------------------------------------------------- diff --git a/.gitignore b/.gitignore index 7b71b9b..d5765e7 100644 --- a/.gitignore +++ b/.gitignore @@ -31,5 +31,5 @@ lib/ pip-selfcheck.json report.html .pytest_cache -tests/ + http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/libcloud/drs/drivers/nttcis.py ---------------------------------------------------------------------- diff --git a/libcloud/drs/drivers/nttcis.py b/libcloud/drs/drivers/nttcis.py index 8d4cb37..e368560 100644 --- a/libcloud/drs/drivers/nttcis.py +++ b/libcloud/drs/drivers/nttcis.py @@ -4,6 +4,8 @@ from libcloud.common.nttcis import API_ENDPOINTS from libcloud.common.nttcis import DEFAULT_REGION from libcloud.drs.types import Provider from libcloud.drs.base import Driver +from libcloud.common.nttcis import TYPES_URN +from libcloud.utils.xml import fixxpath, findtext, findall from libcloud.common.types import LibcloudError @@ -47,3 +49,48 @@ class NttCisDRSDriver(Driver): self)._ex_connection_class_kwargs() kwargs['region'] = self.selected_region return kwargs + + def create_consistency_group(self, name, journal_size_gb , + source_server_id, target_server_id, + description=None): + """ + Create a consistency group + + :param name: Name of consistency group + :type name: ``str`` + :param journal_size_gb: Journal size in GB + :type journal_size_gb: ``str`` + :param source_server_id: Id of the server to copy + :type source_server_id: ``str`` + :param target_server_id: Id of the server to receive the copy + :type: ``str`` + :param description: (Optional) Description of consistency group + :type: ``str`` + :return: :class: `NttCisConsistenccyGroup` + """ + + consistency_group_elm = ET.Element('createConsistencyGroup', + {'xmlns': TYPES_URN}) + ET.SubElement(consistency_group_elm, "name").text = name + if description is not None: + ET.SubElement( + consistency_group_elm,"description").text = description + ET.SubElement( + consistency_group_elm, "journalSizeGb").text = journal_size_gb + server_pair = ET.SubElement(consistency_group_elm, "serverPair") + ET.SubElement( + server_pair, "sourceServerId").text = source_server_id + ET.SubElement( + server_pair, "targetServerId").text = target_server_id + response = self.connection.request_with_orgId_api_2( + "consistencyGroup/createConsistencyGroup", + method="POST", + data=ET.tostring(consistency_group_elm)).object + response_code = findtext(response, 'responseCode', TYPES_URN) + try: + assert response_code in ['IN_PROGRESS', 'OK'] + except AssertionError: + return response_code + else: + info = findtext(response, "info", TYPES_URN) + print(info) http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/libcloud/utils/xml.py ---------------------------------------------------------------------- diff --git a/libcloud/utils/xml.py b/libcloud/utils/xml.py index 3a34d84..55b82a6 100644 --- a/libcloud/utils/xml.py +++ b/libcloud/utils/xml.py @@ -47,17 +47,3 @@ def findattr(element, xpath, namespace=None): def findall(element, xpath, namespace=None): return element.findall(fixxpath(xpath=xpath, namespace=namespace)) - - -def return_all(parent_element): - elem_dict = {} - if parent_element.items(): - elem_dict.update(dict(parent_element.items())) - for element in parent_element: - if element.items(): - elem_dict.update(dict(element.items())) - elif element.text: - elem_dict.update({element.tag.split('}')[1]: element.text}) - else: - elem_dict.update(element.attrib) - return elem_dict http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/tests/conftest.py ---------------------------------------------------------------------- diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..077e463 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,40 @@ +import pytest +import libcloud + + [email protected](scope="module") +def compute_driver(): + cls = libcloud.get_driver(libcloud.DriverType.COMPUTE, + libcloud.DriverType.COMPUTE.NTTCIS) + compute_driver = cls('mitchgeo-test', 'Snmpv2c!', region='eu') + return compute_driver + + [email protected](scope="module") +def lbdriver(): + cd = libcloud.get_driver(libcloud.DriverType.COMPUTE, + libcloud.DriverType.COMPUTE.NTTCIS) + compute_driver = cd('mitchgeo-test', 'Snmpv2c!', region='eu') + net_domain_name = 'sdk_test_1' + net_domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain_id = [d for d in net_domains if d.name == net_domain_name][0].id + cls = libcloud.get_driver(libcloud.DriverType.LOADBALANCER, + libcloud.DriverType.LOADBALANCER.NTTCIS) + lbdriver = cls('mitchgeo-test', net_domain_id, 'Snmpv2c!', region='eu') + return lbdriver + + [email protected](scope="module") +def na_compute_driver(): + cls = libcloud.get_driver(libcloud.DriverType.COMPUTE, + libcloud.DriverType.COMPUTE.NTTCIS) + na_compute_driver = cls('mitchgeo-test', 'Snmpv2c!', region='na') + return na_compute_driver + + [email protected](scope="module") +def drsdriver(): + cls = libcloud.get_driver(libcloud.DriverType.DRS, + libcloud.DriverType.DRS.NTTCIS) + drsdriver = cls('mitchgeo-test', 'Snmpv2c!', region='na') + return drsdriver http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/tests/lib_create_test.py ---------------------------------------------------------------------- diff --git a/tests/lib_create_test.py b/tests/lib_create_test.py new file mode 100644 index 0000000..0b345a8 --- /dev/null +++ b/tests/lib_create_test.py @@ -0,0 +1,252 @@ +from pprint import pprint +import pytest +import libcloud + +from libcloud.compute.drivers.nttcis import NttCisPort, NttCisIpAddress, NttCisPublicIpBlock, NttCisNatRule +from libcloud.common.nttcis import NttCisFirewallRule, NttCisVlan, NttCisFirewallAddress, NttCisAPIException + + +def test_deploy_vlan(compute_driver, vlan_name='sdk_test2', network_domain_name='sdk_test_1', base_ipv4_addr='10.1.2.0'): + # Default network size is 24 bits. Interval and polling times default to 2 and 60. + interval = 3 + timeout = 60 + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + result = compute_driver.ex_create_vlan(network_domain, vlan_name, base_ipv4_addr) + assert isinstance(result, NttCisVlan) + compute_driver.ex_wait_for_state('normal', compute_driver.ex_get_vlan, interval, timeout, result.id) + return result + + +def test_deploy_vlan_2(compute_driver, vlan_name='sdk_test_3', network_domain_name='sdk_test_1', + base_ipv4_addr='10.2.0.0', private_ipv4_prefix_size=24): + # Default network size is 24 bits. Interval and polling times default to 2 and 60. + interval = 3 + timeout = 60 + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + result = compute_driver.ex_create_vlan(network_domain, vlan_name, base_ipv4_addr, + private_ipv4_prefix_size=private_ipv4_prefix_size) + assert isinstance(result, NttCisVlan) + compute_driver.ex_wait_for_state('normal', compute_driver.ex_get_vlan, interval, timeout, result.id) + return result + + +def test_create_nat_rule(compute_driver): + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + result = compute_driver.ex_create_nat_rule(network_domain, '10.1.1.7', '168.128.13.126') + assert isinstance(result, NttCisNatRule) + + +def test_deploy_server(compute_driver): + image_id = "81a36aa0-555c-4735-b965-4b64fcf0ac8f" + images = compute_driver.list_images(location='EU6') + image = [i for i in images if i.id == image_id] + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + psswd = 'Snmpv2c!' + vlan_name = "sdk_vlan1" + vlans = compute_driver.ex_list_vlans() + vlan = [v for v in vlans if v.name == vlan_name] + new_node = compute_driver.create_node("ubuntu", image[0], psswd, ex_description="auto_created_server", + ex_network_domain=net_domain[0], ex_primary_nic_vlan=vlan[0]) + compute_driver.ex_wait_for_state('running', compute_driver.ex_get_node_by_id, 2, 300, new_node.id) + assert new_node.state == 'running' + + +def test_delete_server(compute_driver): + server = compute_driver.list_nodes(ex_name="ubuntu")[0] + shut_result = compute_driver.ex_shutdown_graceful(server) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, server.id) + result = compute_driver.destroy_node(server) + assert result is True + compute_driver.ex_wait_for_state('terminated', compute_driver.ex_get_node_by_id, 2, 240, server.id) + + +def test_deploy_firewall_rule_1(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + address_list_name = 'sdk_test_address_list' + address_lists = compute_driver.ex_list_ip_address_list('6aafcf08-cb0b-432c-9c64-7371265db086') + # using lambda with filter + + # address_list = list(filter(lambda x: address_list_name, address_lists)) + # address_list_id = address_list[0].id + + # using list comprehension to filter + + address_list = [a for a in address_lists if a.name == address_list_name] + address_list_id = address_list[0].id + + port_list_name = 'sdk_test_port_list' + port_lists = compute_driver.ex_list_portlist('6aafcf08-cb0b-432c-9c64-7371265db086') + port_list = [p for p in port_lists if p.name == port_list_name] + port_list_id = port_list[0].id + dest_firewall_address = NttCisFirewallAddress(address_list_id=address_list_id, port_list_id=port_list_id) + source_firewall_address = NttCisFirewallAddress(any_ip='ANY') + rule = compute_driver.ex_create_firewall_rule(net_domain[0], 'sdk_test_firewall_rule_1', 'ACCEPT_DECISIVELY', + 'IPV4', 'TCP', source_firewall_address, dest_firewall_address, 'LAST') + print(rule) + assert isinstance(rule, NttCisFirewallRule) + + +def test_deploy_firewall_rule_2(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + source_firewall_address = NttCisFirewallAddress(any_ip='ANY') + dest_firewall_address = NttCisFirewallAddress(ip_address='10.2.0.0', ip_prefix_size='16', + port_begin='8000', port_end='8080') + + rule = compute_driver.ex_create_firewall_rule(net_domain[0], 'sdk_test_firewall_rule_2', 'ACCEPT_DECISIVELY', + 'IPV4', 'TCP', source_firewall_address, dest_firewall_address, 'LAST') + print(rule) + assert isinstance(rule, NttCisFirewallRule) + + +def test_deploy_firewall_rule_3(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + source_firewall_address = NttCisFirewallAddress(any_ip='ANY') + dest_firewall_address = NttCisFirewallAddress(ip_address='10.2.0.0', ip_prefix_size='16', + port_begin='25') + rule_name = 'sdk_test_firewall_rule_2' + rules = compute_driver.ex_list_firewall_rules(net_domain[0]) + rule = [rule for rule in rules if rule.name == rule_name] + relative_to = compute_driver.ex_get_firewall_rule(net_domain[0], rule[0].id) + rule = compute_driver.ex_create_firewall_rule(net_domain[0], 'sdk_test_firewall_rule_3', 'ACCEPT_DECISIVELY', + 'IPV4', 'TCP', source_firewall_address, dest_firewall_address, + 'BEFORE', position_relative_to_rule=relative_to) + print(rule) + assert isinstance(rule, NttCisFirewallRule) + + +def test_create_port_list(compute_driver): + """ + An optional named argument, child_portlist_list, which takes the id of an existing + port list to include in this port list. + """ + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + port_list_name = 'sdk_test_port_list' + description = 'A test port list' + port_list = [NttCisPort(begin='8000', end='8080')] + result = compute_driver.ex_create_portlist(net_domain[0], port_list_name, description, port_list) + assert result is True + + +def test_create_address_list(compute_driver): + """ + An optional named argument, child_ip_address_list, which takes the id of an existing + port list to include in this port list. + """ + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + address_list_name = 'sdk_test_address_list' + description = 'A test address list' + ip_version = 'IPV4' + # An optional prefix list can be specified as a named argument, prefix_size= + address_list = [NttCisIpAddress('10.2.0.1', end='10.2.0.11')] + + result = compute_driver.ex_create_ip_address_list(net_domain[0], address_list_name, + description, + ip_version, address_list) + assert result is True + + +def test_create_public_ip_block(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name][0] + ip_block = compute_driver.ex_add_public_ip_block_to_network_domain(net_domain) + assert isinstance(ip_block, NttCisPublicIpBlock) + print(ip_block) + + +def test_create_private_ipv4_address(compute_driver): + vlan_name = 'sdk_vlan1' + vlan = compute_driver.ex_list_vlans(name=vlan_name)[0] + ip = '10.1.1.20' + description = 'A test reserved ipv4 address' + result = compute_driver.ex_reserve_ip(vlan, ip, description) + assert result is True + + +def test_create_ipv6_addresss(compute_driver): + vlan_name = 'sdk_vlan1' + vlan = compute_driver.ex_list_vlans(name=vlan_name)[0] + ipv6 = '2a00:47c0:111:1331:7df0:9beb:43c9:5c' + result = compute_driver.ex_reserve_ip(vlan, ipv6) + assert result is True + + +def test_import_customer_image(compute_driver): + package_name = "bitnami-couchdb-2.1.2-1-r35-linux-centos-7-x86_64.mf" + name = "bitnami-couchdb-2.1.2-1-r35-linux-centos-7-x86_64" + datacenter_id = 'EU6' + is_guest_os_customization = 'false' + result = compute_driver.import_image(package_name, name, datacenter_id=datacenter_id, + is_guest_os_customization=is_guest_os_customization) + assert result is True + + +def test_create_load_balancer(lbdriver, compute_driver): + member1 = compute_driver.list_nodes(ex_name='web1')[0] + member2 = compute_driver.list_nodes(ex_name='web2')[0] + members = [member1, member2] + name = 'sdk_test_balancer' + port = '80' + listener_port = '8000' + protocol = 'TCP' + algorithm = 'LEAST_CONNECTIONS_MEMBER' + members = [m for m in members] + ex_listener_ip_address = "168.128.13.127" + lb = lbdriver.create_balancer(name, listener_port=listener_port, port=port, protocol=protocol, + algorithm=algorithm, members=members, optimization_profile='TCP', + ex_listener_ip_address=ex_listener_ip_address) + + +def test_create_vip_node(compute_driver, lbdriver): + node_address = '10.1.1.7' + node_name = "web1" + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name][0] + node = lbdriver.ex_create_node(net_domain.id, node_name, node_address) + print(node) + + +def test_add_pool_member(compute_driver, lbdriver): + pool_name = 'sdk_test_balancer' + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + pools = lbdriver.ex_get_pools(ex_network_domain_id=network_domain.id) + pool = [p for p in pools if p.name == pool_name][0] + node = lbdriver.ex_get_node("eca8dac3-1417-4fdf-83c3-2b7b848ab171") + result = lbdriver.ex_create_pool_member(pool, node, port=80) + print(result) + + +def test_create_server_monitor(compute_driver): + pass + + +def test_fail_create_drs(na_compute_driver, drsdriver): + nodes = na_compute_driver.list_nodes(ex_name='drs_test_1') + src_id = nodes[0].id + nodes = na_compute_driver.list_nodes(ex_name="drs_test_2") + target_id = nodes[0].id + with pytest.raises(NttCisAPIException)as excinfo: + result = drsdriver.create_consistency_group( + "sdk_cg", "100", src_id, target_id, description="A test consistency group") + exception_msg = excinfo.value.msg + assert exception_msg == 'DRS is not supported between source Data Center NA9 and target Data Center NA12.' http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/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..32984ee --- /dev/null +++ b/tests/lib_edit_test.py @@ -0,0 +1,452 @@ +import pytest +import libcloud +from libcloud import loadbalancer +from libcloud.compute.drivers.nttcis import NttCisPort +from libcloud.common.nttcis import NttCisIpAddress, NttCisVlan, NttCisVIPNode +from tests.lib_create_test import test_deploy_vlan + + +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_edit_vlan(compute_driver): + vlan = compute_driver.ex_list_vlans(name='sdk_test2')[0] + vlan.name = 'sdk_test_2' + vlan.description = "Second test Vlan" + result = compute_driver.ex_update_vlan(vlan) + assert isinstance(result, NttCisVlan) + + +def test_expand_vlan(compute_driver): + vlan = compute_driver.ex_list_vlans(name='sdk_test_3')[0] + vlan.private_ipv4_range_size = '23' + result = compute_driver.ex_expand_vlan(vlan) + assert isinstance(result, NttCisVlan) + + +def test_delete_vlan(compute_driver): + vlan = compute_driver.ex_list_vlans(name='sdk_test_3')[0] + result = compute_driver.ex_delete_vlan(vlan) + assert result is True + + +def test_add_disk_by_node(compute_driver): + """ + Speeds can be specified based on DataCenter + :param compute_driver: libcloud.DriverType.COMPUTE.NTTCIS + :return: NA + """ + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + result = compute_driver.ex_add_storage_to_node(20, node) + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 180, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_add_disk_by_controller_id(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + result = compute_driver.ex_add_storage_to_node(20, controller_id=node.extra['scsi_controller'][0].id) + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 180, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_changing_diskspeed(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + disk_id = 'f8a01c24-4768-46be-af75-9fe877f8c588' + result = compute_driver.ex_change_storage_speed(disk_id, 'HIGHPERFORMANCE') + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 240, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_changing_diskspeed_iops(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + disk_id = 'f8a01c24-4768-46be-af75-9fe877f8c588' + result = compute_driver.ex_change_storage_speed(disk_id, 'PROVISIONEDIOPS', iops=60) + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 240, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_add_scsi_controller(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + adapter_type = 'VMWARE_PARAVIRTUAL' + result = compute_driver.ex_add_scsi_controller_to_node(node.id, adapter_type) + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 240, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_remove_scsi_controller(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + result = compute_driver.ex_remove_scsi_controller('f1126751-c6d5-4d64-893c-8902b8481f90') + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 240, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_update_vmware_tools(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + result = compute_driver.ex_update_vm_tools(node) + assert result is True + compute_driver.ex_wait_for_state('running', compute_driver.ex_get_node_by_id, 2, 240, node.id) + + +def test_add_node_to_vlan(compute_driver): + vlan = test_deploy_vlan(compute_driver, "test_vlan_create", "6aafcf08-cb0b-432c-9c64-7371265db086", "10.0.2.0") + assert isinstance(vlan, NttCisVlan) + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + result = compute_driver.ex_attach_node_to_vlan(node, vlan=vlan) + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 240, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_remove_nic(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + result = compute_driver.ex_disable_snapshots(node.id) + assert result is True + result = compute_driver.ex_destroy_nic("e9cdea1b-c4f2-4769-93a8-57e24248abdd") + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 240, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + +"""" +No wayt to get nic id's via libcloud +def test_exchange_nic_vlans(compute_driver): + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + print(node.extra) +""" + + +def test_change_nic_type(compute_driver): + nic_id = "7a27b2b1-7b20-404f-be53-4695023c2734" + nic_type = 'VMXNET3' + node = compute_driver.ex_get_node_by_id('803e5e00-b22a-450a-8827-066ff15ec977') + shut_result = compute_driver.ex_shutdown_graceful(node) + assert shut_result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 45, node.id) + result = compute_driver.ex_change_nic_network_adapter(nic_id, nic_type) + assert result is True + compute_driver.ex_wait_for_state('stopped', compute_driver.ex_get_node_by_id, 2, 240, node.id) + result = compute_driver.ex_start_node(node) + assert result is True + + +def test_edit_firewall_rule(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + rule_name = 'sdk_test_firewall_rule_2' + rules = compute_driver.ex_list_firewall_rules(net_domain[0]) + rule = [rule for rule in rules if rule.name == rule_name] + rule[0].destination.port_end = None + result = compute_driver.ex_edit_firewall_rule(rule[0]) + print(compute_driver.ex_get_firewall_rule(net_domain[0].id, rule[0].id)) + assert result is True + + +def test_delete_firewall_rule(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + rule_name = 'sdk_test_firewall_rule_2' + rules = compute_driver.ex_list_firewall_rules(net_domain[0]) + rule = [rule for rule in rules if rule.name == rule_name] + result = compute_driver.ex_delete_firewall_rule(rule[0]) + assert result is True + + +def test_create_anti_affinity_rule(compute_driver): + server1 = compute_driver.ex_get_node_by_id("d0425097-202f-4bba-b268-c7a73b8da129") + server2 = compute_driver.ex_get_node_by_id("803e5e00-b22a-450a-8827-066ff15ec977") + servers = [server1, server2] + result = compute_driver.ex_create_anti_affinity_rule(servers) + assert isinstance(result, ) + + +def test_delete_anti_affinity_rule(compute_driver): + anti_affinity_rule = "40d83160-0fa2-418d-a73e-5f15fe1354fc" + result = compute_driver.ex_delete_anti_affinity_rule(anti_affinity_rule) + assert result is True + + +def test_delete_port_list(compute_driver): + portlists = compute_driver.ex_list_portlist('6aafcf08-cb0b-432c-9c64-7371265db086') + port_list_to_delete = [plist for plist in portlists if plist.name == 'sdk_test_port_list'] + result = compute_driver.ex_delete_portlist(port_list_to_delete[0]) + assert result is True + + +def test_edit_address_list(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name][0] + addr_list = compute_driver.ex_get_ip_address_list(net_domain, 'sdk_test_address_list') + assert addr_list[0].ip_version == 'IPV4' + ip_address_1 = NttCisIpAddress(begin='190.2.2.100') + ip_address_2 = NttCisIpAddress(begin='190.2.2.106', end='190.2.2.108') + ip_address_3 = NttCisIpAddress(begin='190.2.2.0', prefix_size='24') + ip_address_4 = NttCisIpAddress(begin='10.2.0.0', prefix_size='24') + ip_address_collection = [ip_address_1, ip_address_2, ip_address_3, ip_address_4] + + result = compute_driver.ex_edit_ip_address_list("d32aa8d4-831b-4fd6-95da-c639768834f0", + ip_address_collection=ip_address_collection) + assert result is True + + +def test_delete_public_ip_block(compute_driver): + block = compute_driver.ex_get_public_ip_block("813b87a8-18e1-11e5-8d4f-180373fb68df") + result = compute_driver.ex_delete_public_ip_block(block) + assert result is True + + +def test_edit_address_list_2(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name][0] + # An ip address list object can be used as an argument or the id of the address list + addr_list = compute_driver.ex_get_ip_address_list(net_domain, 'sdk_test_address_list') + + result = compute_driver.ex_edit_ip_address_list("d32aa8d4-831b-4fd6-95da-c639768834f0", + description='nil') + assert result is True + + +def test_delete_address_list(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name][0] + addresslist_to_delete = compute_driver.ex_get_ip_address_list(net_domain, 'sdk_test_address_list') + print(addresslist_to_delete) + + +def test_edit_port_list_1(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name] + port_list_name = 'sdk_test_port_list' + port_lists = compute_driver.ex_list_portlist(net_domain[0]) + port_list = [port for port in port_lists if port.name == port_list_name][0] + port_collection = [NttCisPort(begin='8000', end='8080'), NttCisPort(begin='9000')] + result = compute_driver.ex_edit_portlist(port_list.id, port_collection=port_collection) + assert result is True + + +def test_unreserve_ip_address(compute_driver): + vlan_name = 'sdk_vlan1' + vlan = compute_driver.ex_list_vlans(name=vlan_name)[0] + ip = '2a00:47c0:111:1331:7df0:9beb:43c9:5c' + result = compute_driver.ex_unreserve_ip_addresses(vlan, ip) + assert result is True + + +def test_list_locations(compute_driver): + locations = compute_driver.list_locations() + for location in locations: + print(location) + + +def test_delete_nat_rule(compute_driver): + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + rule = compute_driver.ex_get_nat_rule(network_domain, '74f0897f-5536-4c17-84b0-d52b1fb3aea6') + result = compute_driver.ex_delete_nat_rule(rule) + assert result is True + + +def test_update_health_monitor(compute_driver, lbdriver): + pool_name = 'sdk_test_balancer' + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + pools = lbdriver.ex_get_pools(ex_network_domain_id=network_domain.id) + pool = [p for p in pools if p.name == pool_name][0] + pool.health_monitor_id = '9f79487a-1b6d-11e5-8d4f-180373fb68df' + result = lbdriver.ex_update_pool(pool) + assert result is True + + +def test_update_node_monitor(compute_driver, lbdriver): + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + nodes = lbdriver.ex_get_nodes(ex_network_domain_id=network_domain.id) + #pool = [p for p in pools if p.name == pool_name][0] + health_monitor_id = '9f79a126-1b6d-11e5-8d4f-180373fb68df' + for node in nodes: + node.health_monitor_id = health_monitor_id + result = lbdriver.ex_update_node(node) + assert isinstance(result, NttCisVIPNode) + + +def test_remove_node(compute_driver, lbdriver): + node_name = 'web1' + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + nodes = lbdriver.ex_get_nodes(ex_network_domain_id=network_domain.id) + node = [n for n in nodes if n.name == node_name][0] + pool_name = "sdk_test_balancer" + pools = lbdriver.ex_get_pools(ex_network_domain_id=network_domain.id) + pool = [p for p in pools if p.name == pool_name][0] + pool_members = lbdriver.ex_get_pool_members(pool.id) + pool_member = [pm for pm in pool_members if pm.node_id == node.id][0] + result = lbdriver.ex_destroy_pool_member(pool_member) + assert result is True + + +def test_delete_node(compute_driver, lbdriver): + node_name = 'web1' + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + nodes = lbdriver.ex_get_nodes(ex_network_domain_id=network_domain.id) + node = [n for n in nodes if n.name == node_name][0] + result = lbdriver.ex_destroy_node(node.id) + assert result is True + + +def test_remove_pool(compute_driver, lbdriver): + listener_name = "sdk_test_balancer" + listeners = lbdriver.list_balancers(ex_network_domain_id=lbdriver.network_domain_id) + listener = [l for l in listeners if l.name == listener_name][0] + pool_id = None + result = lbdriver.ex_update_listener(listener, poolId=pool_id) + assert result is True + + +def test_delete_pool(compute_driver, lbdriver): + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + pool_name = "sdk_test_balancer" + pools = lbdriver.ex_get_pools(ex_network_domain_id=network_domain.id) + pool = [p for p in pools if p.name == pool_name][0] + result = lbdriver.ex_destroy_pool(pool) + assert result is True + + +def test_delete_listener(compute_driver, lbdriver): + listener_name = "sdk_test_balancer" + listeners = lbdriver.list_balancers(ex_network_domain_id=lbdriver.network_domain_id) + listener = [l for l in listeners if l.name == listener_name][0] + result = lbdriver.destroy_balancer(listener) + assert result is True http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/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..de39a33 --- /dev/null +++ b/tests/lib_list_test.py @@ -0,0 +1,393 @@ +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) + #print(node.extra['networkDomainId'], node.extra['datacenterId'], node.uuid, node.state, node.name, node.extra['cpu'], + # node.extra['scsi_controller'], 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) + #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) + #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 + + +def test_list_images(compute_driver): + images = compute_driver.list_images(location='EU6') + print() + for image in images: + print(image.id, image.name) + assert isinstance(images, list) and len(images) > 0 + + +def test_list_os(compute_driver): + oss = compute_driver.ex_list_os(location='EU6') + + +def test_list_node_by_image(compute_driver): + nodes = compute_driver.list_nodes(ex_image='81a36aa0-555c-4735-b965-4b64fcf0ac8f') + 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_network_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_datacenter_object_creation(compute_driver): + datacenter = compute_driver.ex_get_datacenter('EU6') + + +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_list_nics(compute_driver): + result = compute_driver.ex_list_ + + +def test_list_vlans(compute_driver): + vlans = compute_driver.ex_list_vlans() + print(vlans) + assert isinstance(vlans, list) + + +def test_list_anti_affinity_rules(compute_driver): + # Could use network domain or node but not both + # net_domain = compute_driver.ex_get_network_domain('6aafcf08-cb0b-432c-9c64-7371265db086') + node = compute_driver.ex_get_node_by_id("803e5e00-b22a-450a-8827-066ff15ec977") + anti_affinity_rules = compute_driver.ex_list_anti_affinity_rules(node=node) + assert len(anti_affinity_rules) > 1 + + +def test_list_no_anti_affinity_rules(compute_driver): + # Could use network domain or node but not both + # net_domain = compute_driver.ex_get_network_domain('6aafcf08-cb0b-432c-9c64-7371265db086') + node = compute_driver.ex_get_node_by_id("803e5e00-b22a-450a-8827-066ff15ec977") + anti_affinity_rules = compute_driver.ex_list_anti_affinity_rules(node=node) + assert len(anti_affinity_rules) == 0 + + +def test_list_locations(compute_driver): + locations = compute_driver.list_locations() + for location in locations: + print(location) + + +""" +def test_list_sizes(compute_driver): + properties = compute_driver.list_locations() + for property in properties: + print(property) +""" + + +def test_images(compute_driver): + images = compute_driver.list_images() + print() + print(images) + assert isinstance(images, list) and len(images) > 0 + + +def test_list_public_ip_blocks(compute_driver): + domain_name = 'sdk_test_1' + domains = compute_driver.ex_list_network_domains(location='EU6') + net_domain = [d for d in domains if d.name == domain_name][0] + blocks = compute_driver.ex_list_public_ip_blocks(net_domain) + print(blocks) + + +def test_list_private_ipv4_addresses_vlan(compute_driver): + vlan_name = 'sdk_vlan1' + vlan = compute_driver.ex_list_vlans(name=vlan_name)[0] + ip_addresses = compute_driver.ex_list_reserved_ipv4(vlan=vlan) + for ip_address in ip_addresses: + print(ip_address) + + +def test_list_private_ipv4_addresses_datacenter(compute_driver): + datacenter_id = 'EU8' + ip_addresses = compute_driver.ex_list_reserved_ipv4(datacenter_id=datacenter_id) + for ip_address in ip_addresses: + print(ip_address) + + +def test_list_private_ipv4_addresses_all(compute_driver): + ip_addresses = compute_driver.ex_list_reserved_ipv4() + for ip_address in ip_addresses: + print(ip_address) + + +def test_list_reserved_ipv6_address_vlan(compute_driver): + vlan_name = 'sdk_vlan1' + vlan = compute_driver.ex_list_vlans(name=vlan_name)[0] + ip_addresses = compute_driver.ex_list_reserved_ipv6(vlan=vlan) + for ip_address in ip_addresses: + print(ip_address) + + +def test_list_nat_rules(compute_driver): + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + rules = compute_driver.ex_list_nat_rules(network_domain) + for rule in rules: + print(rule) + + +def test_list_customer_images(compute_driver): + location = 'EU6' + images = compute_driver.ex_list_customer_images(location) + for image in images: + print(image, image.extra) + + +def test_get_customer_image(compute_driver): + imagee_id = '84da095f-c8c7-4ace-9fb6-eceb1047027c' + image = compute_driver.ex_get_image_by_id(imagee_id) + print(image, image.extra) + + +def test_list_health_monitors(compute_driver, lbdriver): + network_domain_name = "sdk_test_1" + network_domains = compute_driver.ex_list_network_domains(location='EU6') + network_domain = [nd for nd in network_domains if nd.name == network_domain_name][0] + monitors = lbdriver.ex_get_default_health_monitors(network_domain) + for monitor in monitors: + print(monitor) + http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/tests/lib_misc_test.py ---------------------------------------------------------------------- diff --git a/tests/lib_misc_test.py b/tests/lib_misc_test.py new file mode 100644 index 0000000..d8be518 --- /dev/null +++ b/tests/lib_misc_test.py @@ -0,0 +1,10 @@ +import pytest +import libcloud +from libcloud import loadbalancer +from libcloud.common.nttcis import NttCisAPIException + + +def test_server_clone_to_image(compute_driver): + node = compute_driver.ex_get_node_by_id('040fefdb-78be-4b17-8ef9-86820bad67d9 ') + result = compute_driver.ex_clone_node_to_image(node, 'sdk_test_image', image_description='A test image for libcloud') + assert result is True http://git-wip-us.apache.org/repos/asf/libcloud/blob/10c4a31b/tests/test_lib_list.py ---------------------------------------------------------------------- diff --git a/tests/test_lib_list.py b/tests/test_lib_list.py deleted file mode 100644 index 48550cd..0000000 --- a/tests/test_lib_list.py +++ /dev/null @@ -1,344 +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'], - node.extra['scsi_controller'], 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 - - -def test_list_images(compute_driver): - images = compute_driver.list_images(location='EU6') - print() - for image in images: - print(image.id, image.name) - assert isinstance(images, list) and len(images) > 0 - - -def test_list_os(compute_driver): - oss = compute_driver.ex_list_os(location='EU6') - - -def test_list_node_by_image(compute_driver): - nodes = compute_driver.list_nodes(ex_image='81a36aa0-555c-4735-b965-4b64fcf0ac8f') - 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_network_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_datacenter_object_creation(compute_driver): - datacenter = compute_driver.ex_get_datacenter('EU6') - - -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_list_nics(compute_driver): - result = compute_driver.ex_list_ - - -def test_list_vlans(compute_driver): - vlans = compute_driver.ex_list_vlans() - print(vlans) - assert isinstance(vlans, list) - - -def test_list_anti_affinity_rules(compute_driver): - # Could use network domain or node but not both - # net_domain = compute_driver.ex_get_network_domain('6aafcf08-cb0b-432c-9c64-7371265db086') - node = compute_driver.ex_get_node_by_id("803e5e00-b22a-450a-8827-066ff15ec977") - anti_affinity_rules = compute_driver.ex_list_anti_affinity_rules(node=node) - assert len(anti_affinity_rules) > 1 - - -def test_list_no_anti_affinity_rules(compute_driver): - # Could use network domain or node but not both - # net_domain = compute_driver.ex_get_network_domain('6aafcf08-cb0b-432c-9c64-7371265db086') - node = compute_driver.ex_get_node_by_id("803e5e00-b22a-450a-8827-066ff15ec977") - anti_affinity_rules = compute_driver.ex_list_anti_affinity_rules(node=node) - assert len(anti_affinity_rules) == 0 - - - -""" -def test_list_sizes(compute_driver): - properties = compute_driver.list_locations() - for property in properties: - print(property) -""" - - -def test_images(compute_driver): - images = compute_driver.list_images() - print() - print(images) - assert isinstance(images, list) and len(images) > 0 - - -def test_list_public_ip_blocks(compute_driver): - domain_name = 'sdk_test_1' - domains = compute_driver.ex_list_network_domains(location='EU6') - net_domain = [d for d in domains if d.name == domain_name][0] - blocks = compute_driver.ex_list_public_ip_blocks(net_domain) - print(blocks) - - -def test_list_private_ipv4_addresses_vlan(compute_driver): - vlan_name = 'sdk_vlan1' - vlan = compute_driver.ex_list_vlans(name=vlan_name)[0] - ip_addresses = compute_driver.ex_list_reserved_ipv4(vlan=vlan) - for ip_address in ip_addresses: - print(ip_address) - - -def test_list_private_ipv4_addresses_datacenter(compute_driver): - datacenter_id = 'EU8' - ip_addresses = compute_driver.ex_list_reserved_ipv4(datacenter_id=datacenter_id) - for ip_address in ip_addresses: - print(ip_address) - - -def test_list_private_ipv4_addresses_all(compute_driver): - ip_addresses = compute_driver.ex_list_reserved_ipv4() - for ip_address in ip_addresses: - print(ip_address) \ No newline at end of file
