Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_0000_setup(self):
"""Setup the networks required for the other tests in this module.
Create an isolated org vdc network as per the configuration stated
above.
This test passes if the isolated orgvdc network is created
successfully.
"""
logger = Environment.get_default_logger()
TestNetwork._client = Environment.get_client_in_default_org(
TestNetwork._test_runner_role)
TestNetwork._system_client = Environment.get_sys_admin_client()
TestNetwork._vapp_author_client = \
Environment.get_client_in_default_org(CommonRoles.VAPP_AUTHOR)
vdc = Environment.get_test_vdc(TestNetwork._client)
logger.debug('Creating isolated orgvdc network : ' +
TestNetwork._isolated_orgvdc_network_name)
result = vdc.create_isolated_vdc_network(
network_name=TestNetwork._isolated_orgvdc_network_name,
network_cidr=TestNetwork._isolated_orgvdc_network_gateway_ip)
TestNetwork._client.get_task_monitor().wait_for_success(
task=result.Tasks.Task[0])
def test_0098_teardown(self):
"""Test the method vdc.delete_vapp().
Invoke the method for all the vApps created by setup.
This test passes if all the tasks for deleting the vApps succeed.
"""
vdc = Environment.get_test_vdc(TestVappFirewall._client)
task = vdc.delete_vapp(name=TestVappFirewall._vapp_name, force=True)
result = TestVappFirewall._client.get_task_monitor().wait_for_success(
task)
self.assertEqual(result.get('status'), TaskStatus.SUCCESS.value)
def test_9998_teardown(self):
"""Test the method vdc.delete_vapp().
Invoke the method for all the vApps created by setup.
This test passes if all the tasks for deleting the vApps succeed.
"""
vdc = Environment.get_test_vdc(TestVappFirewall._client)
task = vdc.delete_vapp(name=TestVappFirewall._vapp_name, force=True)
result = TestVappFirewall._client.get_task_monitor().wait_for_success(
task)
self.assertEqual(result.get('status'), TaskStatus.SUCCESS.value)
def test_0000_setup(self):
"""Setup the gateway required for the other tests in this module.
Create a gateway as per the configuration stated
above.
This test passes if the gateway is created successfully.
"""
TestGateway._client = Environment.get_sys_admin_client()
TestGateway._vdc = Environment.get_test_vdc(TestGateway._client)
TestGateway._org_client = Environment.get_client_in_default_org(
CommonRoles.ORGANIZATION_ADMINISTRATOR)
TestGateway._config = Environment.get_config()
TestGateway._api_version = TestGateway._config['vcd']['api_version']
external_network = Environment.get_test_external_network(
TestGateway._client)
ext_net_resource = external_network.get_resource()
ip_scopes = ext_net_resource.xpath(
'vcloud:Configuration/vcloud:IpScopes/vcloud:IpScope',
namespaces=NSMAP)
first_ipscope = ip_scopes[0]
gateway_ip = first_ipscope.Gateway.text
prefix_len = netmask_to_cidr_prefix_len(gateway_ip,
def test_0010_list_isolated_orgvdc_networks(self):
"""Test the method vdc.list_orgvdc_isolated_networks().
Fetches all isolated orgvdc networks in the current organization.
This test passes if the network created during setup is present in the
retrieved list of networks.
"""
vdc = Environment.get_test_vdc(TestNetwork._client)
networks = vdc.list_orgvdc_isolated_networks()
for network in networks:
if network.get('name') == \
TestNetwork._isolated_orgvdc_network_name:
return
self.fail('Retrieved list of isolated orgvdc network doesn\'t'
'contain ' + TestNetwork._isolated_orgvdc_network_name)
def test_0000_setup(self):
"""Setup the vms required for the other tests in this module.
Create a vApp with just one vm as per the configuration stated above.
This test passes if the vApp and vm hrefs are not None.
"""
logger = Environment.get_default_logger()
TestVM._client = Environment.get_client_in_default_org(
TestVM._test_runner_role)
TestVM._sys_admin_client = Environment.get_sys_admin_client()
vdc = Environment.get_test_vdc(TestVM._client)
TestVM._media_resource = Environment.get_test_media_resource()
logger.debug('Creating vApp ' + TestVM._test_vapp_name + '.')
TestVM._test_vapp_href = create_customized_vapp_from_template(
client=TestVM._client,
vdc=vdc,
name=TestVM._test_vapp_name,
catalog_name=Environment.get_default_catalog_name(),
template_name=Environment.get_default_template_name(),
memory_size=TestVM._test_vapp_first_vm_memory_size,
num_cpu=TestVM._test_vapp_first_vm_num_cpu,
disk_size=TestVM._test_vapp_first_vm_first_disk_size,
vm_name=TestVM._test_vapp_first_vm_name,
nw_adapter_type=TestVM._test_vapp_first_vm_network_adapter_type)
self.assertIsNotNone(TestVM._test_vapp_href)
def test_0110_attach_independent_disk(self):
vdc = Environment.get_test_vdc(TestVM._client)
idisk = vdc.get_disk(name=TestVM._idisk_name)
task = TestVM._test_vapp. \
attach_disk_to_vm(disk_href=idisk.get('href'),
vm_name=TestVM._test_vapp_first_vm_name)
result = TestVM._client.get_task_monitor().wait_for_success(
task=task)
self.assertEqual(result.get('status'), TaskStatus.SUCCESS.value)
is_disk_attached = self.__validate_is_attached_disk(
is_disk_attached=False)
self.assertTrue(is_disk_attached)
"""
vapps_to_delete = []
if VmTest._empty_vapp_href is not None:
vapps_to_delete.append(VmTest._empty_vapp_name)
vapp = VApp(VmTest._client, href=VmTest._test_old_vapp_href)
self._power_off_and_undeploy(vapp=vapp)
vapp = VApp(VmTest._client, href=VmTest._test_vapp_vmtools_href)
self._power_off_and_undeploy(vapp=vapp)
vapp = VApp(VmTest._client, href=VmTest._test_vapp_href)
self._power_off_and_undeploy(vapp=vapp)
vapps_to_delete.append(VmTest._vapp_name)
vapps_to_delete.append(VmTest._test_vapp_vmtools_name)
vapps_to_delete.append(VAppConstants.name)
self._sys_login()
vdc = Environment.get_test_vdc(VmTest._client)
vdc.delete_disk(name=self._idisk_name)
self._logout()
self._login()
for vapp_name in vapps_to_delete:
task = vdc.delete_vapp(name=vapp_name, force=True)
result = VmTest._client.get_task_monitor().wait_for_success(task)
self.assertEqual(result.get('status'), TaskStatus.SUCCESS.value)
def __convert_to_subinterface(self, client):
vdc = Environment.get_test_vdc(client)
org_vdc_routed_nw = vdc.get_routed_orgvdc_network(
TestNetwork._routed_org_vdc_network_name)
vdc_network = VdcNetwork(client, resource=org_vdc_routed_nw)
result = vdc_network.convert_to_sub_interface()
task = TestNetwork._client.get_task_monitor().wait_for_success(
task=result)
self.assertEqual(task.get('status'), TaskStatus.SUCCESS.value)
# Verify
vdc_network.reload_admin()
reloaded_vdc_network = vdc_network.admin_resource
self.assertTrue(reloaded_vdc_network.Configuration.SubInterface)
# Revert to Internal Interface
def __convert_to_distributed_interface(self, client):
self.__convert_enable_dr_in_gateway_using_sys_client(True)
vdc = Environment.get_test_vdc(client)
org_vdc_routed_nw = vdc.get_routed_orgvdc_network(
TestNetwork._routed_org_vdc_network_name)
vdc_network = VdcNetwork(client, resource=org_vdc_routed_nw)
result = vdc_network.convert_to_distributed_interface()
self.__wait_for_success(client, result)
# Verify
vdc_network.reload_admin()
reloaded_vdc_network = vdc_network.admin_resource
self.assertTrue(
reloaded_vdc_network.Configuration.DistributedInterface)
# Revert
result = vdc_network.convert_to_internal_interface()
self.__wait_for_success(client, result)