You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by re...@apache.org on 2015/10/13 22:39:16 UTC
[2/5] git commit: updated refs/heads/master to be41921
CLOUDSTACK-8934 - Adding tests to cover default routes on IsoNest and RVR nets
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/5d1cdc64
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/5d1cdc64
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/5d1cdc64
Branch: refs/heads/master
Commit: 5d1cdc64409caed239f6814a8f73266c8e5eb6fe
Parents: 45642b8
Author: Wilder Rodrigues <wr...@schubergphilis.com>
Authored: Fri Oct 9 13:31:32 2015 +0200
Committer: Wilder Rodrigues <wr...@schubergphilis.com>
Committed: Sun Oct 11 14:57:31 2015 +0200
----------------------------------------------------------------------
.../component/test_routers_network_ops.py | 705 ++++++-------------
1 file changed, 218 insertions(+), 487 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/5d1cdc64/test/integration/component/test_routers_network_ops.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_routers_network_ops.py b/test/integration/component/test_routers_network_ops.py
index 7f1e4aa..95ede49 100644
--- a/test/integration/component/test_routers_network_ops.py
+++ b/test/integration/component/test_routers_network_ops.py
@@ -26,9 +26,13 @@ from marvin.lib.utils import (cleanup_resources,
from marvin.lib.base import (ServiceOffering,
VirtualMachine,
Account,
- LoadBalancerRule,
+ ServiceOffering,
+ NATRule,
+ NetworkACL,
FireWallRule,
- NATRule)
+ PublicIPAddress,
+ NetworkOffering,
+ Network)
from marvin.lib.common import (get_zone,
get_template,
get_domain,
@@ -38,7 +42,6 @@ from marvin.lib.common import (get_zone,
list_routers,
list_nat_rules,
list_publicIP,
- list_lb_rules,
list_firewall_rules,
list_hosts)
@@ -46,12 +49,12 @@ from marvin.lib.common import (get_zone,
import time
import logging
-class TestCreatePFOnStoppedRouter(cloudstackTestCase):
+class TestRedundantIsolateNetworks(cloudstackTestCase):
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestCreatePFOnStoppedRouter, cls).getClsTestClient()
+ cls.testClient = super(TestRedundantIsolateNetworks, cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = cls.testClient.getParsedTestDataConfig()
@@ -77,20 +80,22 @@ class TestCreatePFOnStoppedRouter(cloudstackTestCase):
cls.api_client,
cls.services["service_offering"]
)
- cls.vm_1 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- templateid=template.id,
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id
- )
+
+ cls.services["nw_off_persistent_RVR"]["egress_policy"] = "true"
+
+ cls.network_offering = NetworkOffering.create(
+ cls.api_client,
+ cls.services["nw_off_persistent_RVR"],
+ conservemode=True
+ )
+ cls.network_offering.update(cls.api_client, state='Enabled')
+
cls._cleanup = [
- cls.account,
- cls.service_offering
- ]
+ cls.service_offering,
+ cls.network_offering,
+ ]
- cls.logger = logging.getLogger('TestCreatePFOnStoppedRouter')
+ cls.logger = logging.getLogger('TestRedundantIsolateNetworks')
cls.stream_handler = logging.StreamHandler()
cls.logger.setLevel(logging.DEBUG)
cls.logger.addHandler(cls.stream_handler)
@@ -100,93 +105,136 @@ class TestCreatePFOnStoppedRouter(cloudstackTestCase):
@classmethod
def tearDownClass(cls):
try:
- cls.api_client = super(
- TestCreatePFOnStoppedRouter,
- cls).getClsTestClient().getApiClient()
- # Clean up, terminate the created resources
cleanup_resources(cls.api_client, cls._cleanup)
-
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
def setUp(self):
self.apiclient = self.testClient.getApiClient()
+ self.account = Account.create(
+ self.apiclient,
+ self.services["account"],
+ admin=True,
+ domainid=self.domain.id
+ )
self.cleanup = []
+ self.cleanup.insert(0, self.account)
return
def tearDown(self):
try:
- # Clean up, terminate the created resources
- cleanup_resources(self.apiclient, self.cleanup)
+ cleanup_resources(self.api_client, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
-
- @attr(tags=["advanced", "advancedns"], required_hardware="true")
- def test_01_CreatePFOnStoppedRouter(self):
- """Stop existing router, add a PF rule and check we can access the VM """
-
- # Get router details associated for that account
- routers = list_routers(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
-
+ @attr(tags=["advanced", "advancedns", "ssh"], required_hardware="true")
+ def test_RVR_Network_FW_PF_SSH_default_routes(self):
+ """ Test redundant router internals """
+ self.logger.debug("Starting test_RVR_Network_FW_PF_SSH_default_routes...")
+
+ self.logger.debug("Creating network with network offering: %s" % self.network_offering.id)
+ network = Network.create(
+ self.apiclient,
+ self.services["network"],
+ accountid=self.account.name,
+ domainid=self.account.domainid,
+ networkofferingid=self.network_offering.id,
+ zoneid=self.zone.id
+ )
+ self.logger.debug("Created network with ID: %s" % network.id)
+
+ networks = Network.list(
+ self.apiclient,
+ id=network.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
+ isinstance(networks, list),
True,
- "Check for list routers response return valid data"
- )
- self.assertNotEqual(
- len(routers),
- 0,
- "Check list router response"
- )
- router = routers[0]
-
- self.logger.debug("Stopping router ID: %s" % router.id)
+ "List networks should return a valid response for created network"
+ )
+ nw_response = networks[0]
- # Stop the router
- cmd = stopRouter.stopRouterCmd()
- cmd.id = router.id
- self.apiclient.stopRouter(cmd)
-
- routers = list_routers(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.logger.debug("Network state: %s" % nw_response.state)
self.assertEqual(
- isinstance(routers, list),
- True,
- "Check for list routers response return valid data"
- )
- router = routers[0]
-
+ nw_response.state,
+ "Allocated",
+ "The network should be in allocated state after creation"
+ )
+
+ self.logger.debug("Listing routers for network: %s" % network.name)
+ routers = Router.list(
+ self.apiclient,
+ networkid=network.id,
+ listall=True
+ )
self.assertEqual(
- router.state,
- 'Stopped',
- "Check list router response for router state"
- )
+ routers,
+ None,
+ "Routers should not be spawned when network is in allocated state"
+ )
- public_ips = list_publicIP(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid,
- zoneid=self.zone.id
- )
+ self.logger.debug("Deploying VM in account: %s" % self.account.name)
+ virtual_machine = VirtualMachine.create(
+ self.apiclient,
+ self.services["virtual_machine"],
+ accountid=self.account.name,
+ domainid=self.account.domainid,
+ serviceofferingid=self.service_offering.id,
+ networkids=[str(network.id)]
+ )
+ self.logger.debug("Deployed VM in network: %s" % network.id)
+
+ vms = VirtualMachine.list(
+ self.apiclient,
+ id=virtual_machine.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(public_ips, list),
- True,
- "Check for list public IPs response return valid data"
- )
-
- public_ip = public_ips[0]
-
- # Open up firewall port for SSH
+ isinstance(vms, list),
+ True,
+ "List Vms should return a valid list"
+ )
+ vm = vms[0]
+ self.assertEqual(
+ vm.state,
+ "Running",
+ "VM should be in running state after deployment"
+ )
+
+ self.logger.debug("Listing routers for network: %s" % network.name)
+ routers = Router.list(
+ self.apiclient,
+ networkid=network.id,
+ listall=True
+ )
+ self.assertEqual(
+ isinstance(routers, list),
+ True,
+ "list router should return Master and backup routers"
+ )
+ self.assertEqual(
+ len(routers),
+ 2,
+ "Length of the list router should be 2 (Backup & master)"
+ )
+
+ self.logger.debug("Associating public IP for network: %s" % network.name)
+ public_ip = PublicIPAddress.create(
+ self.apiclient,
+ accountid=self.account.name,
+ zoneid=self.zone.id,
+ domainid=self.account.domainid,
+ networkid=network.id
+ )
+ self.logger.debug("Associated %s with network %s" % (
+ public_ip.ipaddress.ipaddress,
+ network.id
+ ))
+
+ self.logger.debug("Creating Firewall rule for VM ID: %s" % virtual_machine.id)
FireWallRule.create(
self.apiclient,
ipaddressid=public_ip.id,
@@ -196,75 +244,43 @@ class TestCreatePFOnStoppedRouter(cloudstackTestCase):
endport=self.services["natrule"]["publicport"]
)
- self.logger.debug("Creating NAT rule for VM ID: %s" % self.vm_1.id)
- # Create NAT rule
+ self.logger.debug("Creating NAT rule for VM ID: %s" % virtual_machine.id)
nat_rule = NATRule.create(
self.apiclient,
- self.vm_1,
+ virtual_machine,
self.services["natrule"],
public_ip.id
)
- self.logger.debug("Starting router ID: %s" % router.id)
- # Start the router
- cmd = startRouter.startRouterCmd()
- cmd.id = router.id
- self.apiclient.startRouter(cmd)
+ self.cleanup.insert(0, network)
+ self.cleanup.insert(0, virtual_machine)
- routers = list_routers(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid,
- zoneid=self.zone.id
- )
- self.assertEqual(
- isinstance(routers, list),
- True,
- "Check for list routers response return valid data"
- )
- router = routers[0]
-
- self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
- # NAT Rule should be in Active state after router start
- nat_rules = list_nat_rules(
- self.apiclient,
- id=nat_rule.id
- )
- self.assertEqual(
- isinstance(nat_rules, list),
- True,
- "Check for list NAT rules response return valid data"
- )
- self.assertEqual(
- nat_rules[0].state,
- 'Active',
- "Check list port forwarding rules"
- )
+ result = 'failed'
try:
+ ssh_command = "ping -c 3 8.8.8.8"
+ ssh = virtual_machine.get_ssh_client(ipaddress=public_ip.ipaddress.ipaddress, retries=5)
+ self.logger.debug("Ping to google.com from VM")
- self.logger.debug("SSH into VM with ID: %s" % nat_rule.ipaddress)
+ result = str(ssh.execute(ssh_command))
+ self.logger.debug("SSH result: %s; COUNT is ==> %s" % (result, result.count("3 packets received")))
+ except:
+ self.fail("Failed to SSH into VM - %s" % (public_ip.ipaddress.ipaddress))
+
+ self.assertEqual(
+ result.count("3 packets received"),
+ 1,
+ "Ping to outside world from VM should be successful"
+ )
- self.vm_1.get_ssh_client(
- ipaddress=nat_rule.ipaddress,
- port=self.services["natrule"]["publicport"])
- except Exception as e:
- self.fail(
- "SSH Access failed for %s: %s" %
- (nat_rule.ipaddress, e)
- )
return
-class TestCreateLBOnStoppedRouter(cloudstackTestCase):
+class TestIsolatedNetworks(cloudstackTestCase):
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestCreateLBOnStoppedRouter, cls).getClsTestClient()
+ cls.testClient = super(TestIsolatedNetworks, cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = cls.testClient.getParsedTestDataConfig()
@@ -290,20 +306,39 @@ class TestCreateLBOnStoppedRouter(cloudstackTestCase):
cls.api_client,
cls.services["service_offering"]
)
- cls.vm_1 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- templateid=template.id,
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id
- )
+
+ cls.services["network_offering"]["egress_policy"] = "true"
+
+ cls.network_offering = NetworkOffering.create(cls.api_client,
+ cls.services["network_offering"],
+ conservemode=True)
+
+ cls.network_offering.update(cls.api_client, state='Enabled')
+
+ cls.network = Network.create(cls.api_client,
+ cls.services["network"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ networkofferingid=cls.network_offering.id,
+ zoneid=cls.zone.id)
+
+ cls.vm_1 = VirtualMachine.create(cls.api_client,
+ cls.services["virtual_machine"],
+ templateid=template.id,
+ accountid=cls.account.name,
+ domainid=cls.domain.id,
+ serviceofferingid=cls.service_offering.id,
+ networkids=[str(cls.network.id)])
+
cls._cleanup = [
- cls.account,
- cls.service_offering
+ cls.vm_1,
+ cls.network,
+ cls.network_offering,
+ cls.service_offering,
+ cls.account
]
-
- cls.logger = logging.getLogger('TestCreateLBOnStoppedRouter')
+
+ cls.logger = logging.getLogger('TestIsolatedNetworks')
cls.stream_handler = logging.StreamHandler()
cls.logger.setLevel(logging.DEBUG)
cls.logger.addHandler(cls.stream_handler)
@@ -313,33 +348,20 @@ class TestCreateLBOnStoppedRouter(cloudstackTestCase):
@classmethod
def tearDownClass(cls):
try:
- cls.api_client = super(
- TestCreateLBOnStoppedRouter,
- cls).getClsTestClient().getApiClient()
- # Clean up, terminate the created resources
cleanup_resources(cls.api_client, cls._cleanup)
-
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
def setUp(self):
self.apiclient = self.testClient.getApiClient()
- self.cleanup = []
- return
-
- def tearDown(self):
- try:
- cleanup_resources(self.apiclient, self.cleanup)
- except Exception as e:
- raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "advancedns"], required_hardware="true")
- def test_01_CreateLBOnStoppedRouter(self):
- """Stop existing Router, add LB rule and check we can reach the VM"""
+ @attr(tags=["advanced", "advancedns", "ssh"], required_hardware="true")
+ def test_isolate_network_FW_PF_default_routes(self):
+ """Stop existing router, add a PF rule and check we can access the VM """
- # Get router details associated for that account
+ self.logger.debug("Starting test_isolate_network_FW_PF_default_routes...")
routers = list_routers(
self.apiclient,
account=self.account.name,
@@ -360,366 +382,75 @@ class TestCreateLBOnStoppedRouter(cloudstackTestCase):
router = routers[0]
- self.logger.debug("Stopping router with ID: %s" % router.id)
- # Stop the router
- cmd = stopRouter.stopRouterCmd()
- cmd.id = router.id
- self.apiclient.stopRouter(cmd)
-
- routers = list_routers(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
- self.assertEqual(
- isinstance(routers, list),
- True,
- "Check for list routers response return valid data"
- )
- router = routers[0]
-
self.assertEqual(
router.state,
- 'Stopped',
+ 'Running',
"Check list router response for router state"
)
public_ips = list_publicIP(
self.apiclient,
account=self.account.name,
- domainid=self.account.domainid
+ domainid=self.account.domainid,
+ zoneid=self.zone.id
)
+
self.assertEqual(
isinstance(public_ips, list),
True,
"Check for list public IPs response return valid data"
)
+
public_ip = public_ips[0]
- # Open up firewall port for SSH
+ self.logger.debug("Creating Firewall rule for VM ID: %s" % self.vm_1.id)
FireWallRule.create(
self.apiclient,
ipaddressid=public_ip.id,
- protocol=self.services["lbrule"]["protocol"],
+ protocol=self.services["natrule"]["protocol"],
cidrlist=['0.0.0.0/0'],
- startport=self.services["lbrule"]["publicport"],
- endport=self.services["lbrule"]["publicport"]
- )
- self.logger.debug("Creating LB rule for public IP: %s" % public_ip.id)
- # Create Load Balancer rule and assign VMs to rule
- lb_rule = LoadBalancerRule.create(
- self.apiclient,
- self.services["lbrule"],
- public_ip.id,
- accountid=self.account.name
+ startport=self.services["natrule"]["publicport"],
+ endport=self.services["natrule"]["publicport"]
)
- self.logger.debug("Assigning VM %s to LB rule: %s" % (
- self.vm_1.id,
- lb_rule.id
- ))
- lb_rule.assign(self.apiclient, [self.vm_1])
-
- # Start the router
- cmd = startRouter.startRouterCmd()
- cmd.id = router.id
- self.apiclient.startRouter(cmd)
- routers = list_routers(
+ self.logger.debug("Creating NAT rule for VM ID: %s" % self.vm_1.id)
+ # Create NAT rule
+ nat_rule = NATRule.create(
self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
- self.assertEqual(
- isinstance(routers, list),
- True,
- "Check for list routers response return valid data"
+ self.vm_1,
+ self.services["natrule"],
+ public_ip.id
)
- router = routers[0]
- self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
- # After router start, LB RUle should be in Active state
- lb_rules = list_lb_rules(
+ nat_rules = list_nat_rules(
self.apiclient,
- id=lb_rule.id
+ id=nat_rule.id
)
self.assertEqual(
- isinstance(lb_rules, list),
+ isinstance(nat_rules, list),
True,
- "Check for list LB rules response return valid data"
+ "Check for list NAT rules response return valid data"
)
self.assertEqual(
- lb_rules[0].state,
+ nat_rules[0].state,
'Active',
- "Check list load balancing rules"
- )
- self.assertEqual(
- lb_rules[0].publicport,
- str(self.services["lbrule"]["publicport"]),
- "Check list load balancing rules"
- )
-
- try:
- self.logger.debug("SSH into VM with IP: %s" % public_ip.ipaddress)
- self.vm_1.ssh_port = self.services["lbrule"]["publicport"]
- self.vm_1.get_ssh_client(public_ip.ipaddress)
- except Exception as e:
- self.fail(
- "SSH Access failed for %s: %s" %
- (self.vm_1.ipaddress, e)
- )
- return
-
-
-class TestCreateFWOnStoppedRouter(cloudstackTestCase):
-
- @classmethod
- def setUpClass(cls):
-
- cls.testClient = super(TestCreateFWOnStoppedRouter, cls).getClsTestClient()
- cls.api_client = cls.testClient.getApiClient()
-
- cls.services = cls.testClient.getParsedTestDataConfig()
- # Get Zone, Domain and templates
- cls.domain = get_domain(cls.api_client)
- cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
- cls.services['mode'] = cls.zone.networktype
- template = get_template(
- cls.api_client,
- cls.zone.id,
- cls.services["ostype"]
- )
- cls.services["virtual_machine"]["zoneid"] = cls.zone.id
-
- # Create an account, network, VM and IP addresses
- cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- domainid=cls.domain.id
- )
- cls.service_offering = ServiceOffering.create(
- cls.api_client,
- cls.services["service_offering"]
- )
- cls.vm_1 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- templateid=template.id,
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id
+ "Check list port forwarding rules"
)
- cls._cleanup = [
- cls.account,
- cls.service_offering
- ]
-
- cls.logger = logging.getLogger('TestCreateFWOnStoppedRouter')
- cls.stream_handler = logging.StreamHandler()
- cls.logger.setLevel(logging.DEBUG)
- cls.logger.addHandler(cls.stream_handler)
-
- return
-
- @classmethod
- def tearDownClass(cls):
- try:
- cls.api_client = super(
- TestCreateFWOnStoppedRouter,
- cls).getClsTestClient().getApiClient()
- # Clean up, terminate the created templates
- cleanup_resources(cls.api_client, cls._cleanup)
-
- except Exception as e:
- raise Exception("Warning: Exception during cleanup : %s" % e)
- return
-
- def tearDown(self):
+
+ result = 'failed'
try:
- cleanup_resources(self.apiclient, self.cleanup)
- except Exception as e:
- raise Exception("Warning: Exception during cleanup : %s" % e)
- return
-
- def setUp(self):
- self.apiclient = self.testClient.getApiClient()
- self.hypervisor = self.testClient.getHypervisorInfo()
- self.cleanup = []
- return
-
- @attr(tags=["advanced", "advancedns"], required_hardware="true")
- def test_01_CreateFWOnStoppedRouter(self):
- """Stop existing Router, create Firewall rules and check that the rules are applied to the router"""
-
- # Get the router details associated with account
- routers = list_routers(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
-
- self.assertEqual(
- isinstance(routers, list),
- True,
- "Check for list routers response return valid data"
- )
-
- self.assertNotEqual(
- len(routers),
- 0,
- "Check list router response"
- )
-
- router = routers[0]
-
- self.logger.debug("Stopping the router: %s" % router.id)
- # Stop the router
- cmd = stopRouter.stopRouterCmd()
- cmd.id = router.id
- self.apiclient.stopRouter(cmd)
-
- routers = list_routers(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
- self.assertEqual(
- isinstance(routers, list),
- True,
- "Check for list routers response return valid data"
- )
- router = routers[0]
-
- self.assertEqual(
- router.state,
- 'Stopped',
- "Check list router response for router state"
- )
-
- public_ips = list_publicIP(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
- self.assertEqual(
- isinstance(public_ips, list),
- True,
- "Check for list public IP response return valid data"
- )
- public_ip = public_ips[0]
-
- # Create Firewall rule with configurations from settings file
- fw_rule = FireWallRule.create(
- self.apiclient,
- ipaddressid=public_ip.id,
- protocol='TCP',
- cidrlist=[self.services["fwrule"]["cidr"]],
- startport=self.services["fwrule"]["startport"],
- endport=self.services["fwrule"]["endport"]
- )
- self.logger.debug("Created firewall rule: %s" % fw_rule.id)
-
- self.logger.debug("Starting the router: %s" % router.id)
- # Start the router
- cmd = startRouter.startRouterCmd()
- cmd.id = router.id
- self.apiclient.startRouter(cmd)
-
- routers = list_routers(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
- self.assertEqual(
- isinstance(routers, list),
- True,
- "Check for list routers response return valid data"
- )
-
- router = routers[0]
-
- self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
- # After Router start, FW rule should be in Active state
- fw_rules = list_firewall_rules(
- self.apiclient,
- id=fw_rule.id,
- )
- self.assertEqual(
- isinstance(fw_rules, list),
- True,
- "Check for list FW rules response return valid data"
- )
+ ssh_command = "ping -c 3 8.8.8.8"
+ self.logger.debug("SSH into VM with ID: %s" % nat_rule.ipaddress)
- self.assertEqual(
- fw_rules[0].state,
- 'Active',
- "Check list load balancing rules"
- )
- self.assertEqual(
- fw_rules[0].startport,
- str(self.services["fwrule"]["startport"]),
- "Check start port of firewall rule"
- )
+ ssh = self.vm_1.get_ssh_client(ipaddress=nat_rule.ipaddress, port=self.services["natrule"]["publicport"], retries=5)
+ result = str(ssh.execute(ssh_command))
+ self.logger.debug("SSH result: %s; COUNT is ==> %s" % (result, result.count("3 packets received")))
+ except:
+ self.fail("Failed to SSH into VM - %s" % (public_ip.ipaddress.ipaddress))
self.assertEqual(
- fw_rules[0].endport,
- str(self.services["fwrule"]["endport"]),
- "Check end port of firewall rule"
- )
- # For DNS and DHCP check 'dnsmasq' process status
- if (self.hypervisor.lower() == 'vmware'
- or self.hypervisor.lower() == 'hyperv'):
- result = get_process_status(
- self.apiclient.connection.mgtSvr,
- 22,
- self.apiclient.connection.user,
- self.apiclient.connection.passwd,
- router.linklocalip,
- 'iptables -t nat -L',
- hypervisor=self.hypervisor
- )
- else:
- hosts = list_hosts(
- self.apiclient,
- id=router.hostid,
- )
- self.assertEqual(
- isinstance(hosts, list),
- True,
- "Check for list hosts response return valid data"
- )
- host = hosts[0]
- host.user = self.services["configurableData"]["host"]["username"]
- host.passwd = self.services["configurableData"]["host"]["password"]
- try:
- result = get_process_status(
- host.ipaddress,
- 22,
- host.user,
- host.passwd,
- router.linklocalip,
- 'iptables -t nat -L'
- )
- except KeyError:
- self.skipTest(
- "Provide a marvin config file with host\
- credentials to run %s" %
- self._testMethodName)
-
- self.logger.debug("iptables -t nat -L: %s" % result)
- self.logger.debug("Public IP: %s" % public_ip.ipaddress)
- res = str(result)
- self.assertEqual(
- res.count(str(public_ip.ipaddress)),
- 1,
- "Check public IP address"
- )
-
+ result.count("3 packets received"),
+ 1,
+ "Ping to outside world from VM should be successful"
+ )
return