connectivity test cases added for boot from volume servers, limits test cases, and negative connectivity test cases

Change-Id: I2bef2e0389019b05ffa4f5ca12bfcde664a0a136
This commit is contained in:
Basavaraj Lamani 2018-08-13 15:43:32 +05:30
parent 0d0da0f8c8
commit 53918afca0
5 changed files with 480 additions and 5 deletions

View File

@ -337,16 +337,18 @@ class ServerFromVolumeV1Fixture(BlockstorageIntegrationFixture):
class ServerFromVolumeV2Fixture(BlockstorageIntegrationFixture):
@classmethod
def create_server(cls, flavor_ref=None, key_name=None):
def create_server(cls, flavor_ref=None, key_name=None, networks=None):
"""
@summary: Base fixture for compute tests creating the Boot from Volume
Version 2 Instance Changes between the two versions are block device
mapping is deprecated in favor of block device which is now creating
the volume behind the scenes
@param flavor_ref: The flavor used to build the server.
@type key_name: String
@param key_name: Generated key for the instance
@type flavor_ref: String
@param key_name: Generated key for the instance
@type key_name: String
@param networks: Networks for the server.
@type networks: List
@return: Response Object containing response code and
the server domain object
@rtype: Request Response Object
@ -362,7 +364,7 @@ class ServerFromVolumeV2Fixture(BlockstorageIntegrationFixture):
# Creating the Boot from Volume Version 2 Instance
cls.server_response = cls.volume_server_behaviors.create_active_server(
block_device=cls.block_device_matrix, flavor_ref=flavor_ref,
key_name=key_name)
key_name=key_name, networks=networks)
cls.server = cls.server_response.entity
cls.resources.add(cls.server.id, cls.servers_client.delete_server)
return cls.server

View File

@ -0,0 +1,198 @@
"""
Copyright 2018 Rackspace
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import re
from cafe.drivers.unittest.decorators import tags
from cafe.engine.clients.ping import PingClient
from cloudcafe.networking.networks.common.tools.connectivity import \
Connectivity
from cloudcafe.networking.networks.personas import ServerPersona
from cloudroast.compute.fixtures import ServerFromVolumeV2Fixture
from cloudroast.networking.networks.fixtures import NetworkingComputeFixture
from cloudroast.networking.networks.scenario.common import ScenarioMixin
class TestBFVConnectivity(NetworkingComputeFixture, ServerFromVolumeV2Fixture,
ScenarioMixin):
"""Testing connectivity between boot from volume servers"""
NAMES_PREFIX = 'bfv_connectivity'
PRIVATE_KEY_PATH = '/root/pkey'
MAX_RETRIES = 5
SSH_COMMAND = ('ssh -o UserKnownHostsFile=/dev/null '
'-o StrictHostKeyChecking=no -o ConnectTimeout=60 '
'-i {private_key_path} {user}@{ip_address}')
ssh_msg = ('Failed remote ssh connection from '
'server {0} to server {1}')
@classmethod
def setUpClass(cls):
super(TestBFVConnectivity, cls).setUpClass()
network_name = 'network_{0}'.format(cls.NAMES_PREFIX)
cls._create_keypair()
cls.delete_keypairs.append(cls.keypair.name)
network, subnet, port = (
cls.net.behaviors.create_network_subnet_port(name=network_name))
cls.delete_ports.append(port.id)
cls.delete_networks.append(network.id)
cls.delete_subnets.append(subnet.id)
server1 = cls.create_server(key_name=cls.keypair.name,
networks=[{'uuid': cls.public_network_id},
{'uuid': cls.service_network_id},
{'uuid': network.id}])
server2 = cls.create_server(key_name=cls.keypair.name,
networks=[{'uuid': cls.public_network_id},
{'uuid': cls.service_network_id},
{'uuid': network.id}])
# Defining the server personas
cls.sp1 = ServerPersona(
server=server1, pnet=True, snet=True, inet=True,
network=network, keypair=cls.keypair, ssh_username='root')
cls.sp2 = ServerPersona(
server=server2, pnet=True, snet=True, inet=True,
network=network, keypair=cls.keypair, ssh_username='root')
server_ids = [cls.sp1.server.id, cls.sp2.server.id]
cls.delete_servers.extend(server_ids)
cls._transfer_private_key_to_vm(cls.sp1.remote_client.ssh_client,
cls.keypair.private_key,
cls.PRIVATE_KEY_PATH
)
cls._transfer_private_key_to_vm(cls.sp2.remote_client.ssh_client,
cls.keypair.private_key,
cls.PRIVATE_KEY_PATH
)
@tags('bfv_server_test', 'positive')
def test_server_ifconfig(self):
"""Testing ifconfig on BFV servers"""
servers = [self.sp1, self.sp2]
for server in servers:
ips = []
ips.extend(server.pnet_fix_ipv4)
ips.extend(server.snet_fix_ipv4)
ips.extend(server.inet_fix_ipv4)
rm_client = server.remote_client
ifconfig_ips = []
stdout = None
retry_count = 0
while stdout is None or len(ifconfig_ips) != 3:
del ifconfig_ips[:]
if retry_count < self.MAX_RETRIES:
ifconfig_output = rm_client.ssh_client. \
execute_shell_command("hostname -I")
stdout = ifconfig_output.stdout
pattern = re.compile(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
matches = pattern.finditer(stdout)
for match in matches:
ifconfig_ips.append(match.group())
if len(ifconfig_ips) == 3 or retry_count == self.MAX_RETRIES:
break
retry_count += 1
server_ip_not_found = False
for ip in ips:
if ip not in ifconfig_ips:
server_ip_not_found = True
break
self.assertFalse(server_ip_not_found,
msg="server {} ip {} not found in output of "
"ifconfig {}".
format(server, ip, ifconfig_ips))
@tags('bfv_server_test', 'positive')
def test_public_ping(self):
"""Testing ping on BFV servers with public network"""
msg_err = 'Public ping to IP address {0} - FAILED'
msg_ok = 'Public ping to IP address {0} - OK'
pub_ipv4_addr = []
pub_ipv4_addr.extend(self.sp1.pnet_fix_ipv4)
pub_ipv4_addr.extend(self.sp2.pnet_fix_ipv4)
all_pub_ips_ping_result = []
failure_flag = False
for ip_addr in pub_ipv4_addr:
ip_addr_reachable = PingClient.ping(ip_addr, 4)
if ip_addr_reachable:
all_pub_ips_ping_result.append(msg_ok.format(ip_addr))
else:
all_pub_ips_ping_result.append(msg_err.format(ip_addr))
failure_flag = True
msg = 'Got connectivity failures. Ping Results: {0}'
# Fail the test if any ping failure is found
self.assertFalse(failure_flag, msg.format(all_pub_ips_ping_result))
@tags('bfv_server_test', 'positive')
def test_remote_public_ping(self):
"""Testing public network remote ping on BFV servers"""
self._test_remote_ping(port_type='pnet')
@tags('bfv_server_test', 'positive')
def test_remote_private_ping(self):
"""Testing private network remote ping on BFV servers"""
self._test_remote_ping(port_type='snet')
@tags('bfv_server_test', 'positive')
def test_remote_isolated_ping(self):
"""Testing isolated network remote ping on BFV servers"""
self._test_remote_ping(port_type='inet')
def _test_remote_ping(self, port_type):
"""Testing remote ping on BFV servers"""
conn = Connectivity(self.sp2, self.sp1)
icmp_basic = dict(port_type=port_type, protocol='icmp', ip_version=4)
rp = conn.verify_personas_conn(**icmp_basic)
result = rp[0]
ping_result = result['connection']
self.assertTrue(ping_result, rp)
@tags('bfv_server_test', 'positive')
def test_remote_public_ssh(self):
"""Testing Public remote ssh on BFV servers"""
self._test_remote_ssh(self.sp1.pnet_fix_ipv4[0])
@tags('bfv_server_test', 'positive')
def test_remote_private_ssh(self):
"""Testing ServiceNet remote ssh on BFV servers"""
self._test_remote_ssh(self.sp1.snet_fix_ipv4[0])
@tags('bfv_server_test', 'positive')
def test_remote_isolated_ssh(self):
"""Testing isolated network remote ssh on BFV servers"""
self._test_remote_ssh(self.sp1.inet_fix_ipv4[0])
def _test_remote_ssh(self, target_ip_addr):
"""Testing remote ssh on BFV servers"""
rc2 = self.sp2.remote_client
ssh_cmd = self.SSH_COMMAND.format(
private_key_path=self.PRIVATE_KEY_PATH,
user=self.sp1.ssh_username, ip_address=target_ip_addr)
stdout = None
ssh_connection_established = False
retry_count = 0
while stdout is None or not stdout.endswith('# '):
if retry_count < self.MAX_RETRIES:
output = rc2.ssh_client.execute_shell_command(ssh_cmd)
stdout = output.stdout
retry_count += 1
if stdout.endswith('# '):
ssh_connection_established = True
self.assertTrue(ssh_connection_established, self.ssh_msg.format(
self.sp2.pnet_fix_ipv4[0], target_ip_addr))

View File

@ -152,7 +152,6 @@ class TestConnectivity(NetworkingComputeFixture, ScenarioMixin):
"""Testing isolated network remote ping on servers"""
self._test_remote_ping(port_type='inet')
@tags('connectivity', 'positive')
def _test_remote_ping(self, port_type):
"""Testing remote ping on servers"""
conn = Connectivity(self.sp2, self.sp1)
@ -191,6 +190,8 @@ class TestConnectivity(NetworkingComputeFixture, ScenarioMixin):
output = rc2.ssh_client.execute_shell_command(ssh_cmd)
stdout = output.stdout
retry_count += 1
if retry_count == self.MAX_RETRIES:
break
if stdout.endswith('# '):
ssh_connection_established = True
self.assertTrue(ssh_connection_established, self.ssh_msg.format(

View File

@ -0,0 +1,128 @@
"""
Copyright 2018 Rackspace
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from cafe.drivers.unittest.decorators import tags
from cafe.engine.clients.ping import PingClient
from cloudcafe.networking.networks.common.tools.connectivity import \
Connectivity
from cloudcafe.networking.networks.personas import ServerPersona
from cloudroast.networking.networks.fixtures import NetworkingComputeFixture
from cloudroast.networking.networks.scenario.common import ScenarioMixin
class TestLimits(NetworkingComputeFixture, ScenarioMixin):
"""Testing connectivity between servers with 100 number of pings"""
NAMES_PREFIX = 'limits_connectivity'
PRIVATE_KEY_PATH = '/root/pkey'
NUM_PINGS = 100
IP_VERSION = 4
@classmethod
def setUpClass(cls):
super(TestLimits, cls).setUpClass()
network_name = 'network_{0}'.format(cls.NAMES_PREFIX)
svr_name_1 = 'svr_1_{0}'.format(cls.NAMES_PREFIX)
svr_name_2 = 'svr_2_{0}'.format(cls.NAMES_PREFIX)
cls._create_keypair()
cls.delete_keypairs.append(cls.keypair.name)
network, subnet, port = (
cls.net.behaviors.create_network_subnet_port(name=network_name))
cls.delete_ports.append(port.id)
cls.delete_networks.append(network.id)
cls.delete_subnets.append(subnet.id)
servers = cls.net.behaviors.create_multiple_servers(
names=[svr_name_1, svr_name_2], pnet=True, snet=True,
keypair_name=cls.keypair.name, networks=[network.id])
svr_names = servers.keys()
svr_names.sort()
# Defining the server personas
cls.sp1 = ServerPersona(
server=servers[svr_names[0]], pnet=True, snet=True, inet=True,
network=network, keypair=cls.keypair,
ssh_username='root')
cls.sp2 = ServerPersona(
server=servers[svr_names[1]], pnet=True, snet=True, inet=True,
network=network, keypair=cls.keypair,
ssh_username='root')
server_ids = [cls.sp1.server.id, cls.sp2.server.id]
cls.delete_servers.extend(server_ids)
cls._transfer_private_key_to_vm(cls.sp1.remote_client.ssh_client,
cls.keypair.private_key,
cls.PRIVATE_KEY_PATH
)
cls._transfer_private_key_to_vm(cls.sp2.remote_client.ssh_client,
cls.keypair.private_key,
cls.PRIVATE_KEY_PATH
)
@tags('connectivity', 'positive')
def test_public_ping(self):
"""Testing 100 number of pings on servers with public network"""
msg_err = 'Public ping to IP address {0} - FAILED'
msg_ok = 'Public ping to IP address {0} - OK'
pub_ipv4_addr = []
pub_ipv4_addr.extend(self.sp1.pnet_fix_ipv4)
pub_ipv4_addr.extend(self.sp2.pnet_fix_ipv4)
all_pub_ips_ping_result = []
failure_flag = False
for ip_addr in pub_ipv4_addr:
ip_addr_reachable = PingClient.ping(
ip=ip_addr, ip_address_version=self.IP_VERSION,
num_pings=self.NUM_PINGS)
if ip_addr_reachable:
all_pub_ips_ping_result.append(msg_ok.format(ip_addr))
else:
all_pub_ips_ping_result.append(msg_err.format(ip_addr))
failure_flag = True
msg = 'Got connectivity failures. Ping Results: {0}'
# Fail the test if any ping failure is found
self.assertFalse(failure_flag, msg.format(all_pub_ips_ping_result))
@tags('connectivity', 'positive')
def test_remote_public_ping(self):
"""
Testing 100 number of pings on remote servers with public network
"""
self._test_remote_ping(port_type='pnet')
@tags('connectivity', 'positive')
def test_remote_private_ping(self):
"""
Testing 100 number of pings on remote servers with private network
"""
self._test_remote_ping(port_type='snet')
@tags('connectivity', 'positive')
def test_remote_isolated_ping(self):
"""
Testing 100 number of pings on remote servers with isolated network
"""
self._test_remote_ping(port_type='inet')
def _test_remote_ping(self, port_type):
"""Testing remote ping on servers"""
conn = Connectivity(self.sp2, self.sp1)
icmp_basic = dict(port_type=port_type,
protocol='icmp', ip_version=self.IP_VERSION,
count=self.NUM_PINGS)
rp = conn.verify_personas_conn(**icmp_basic)
result = rp[0]
ping_result = result['connection']
self.assertTrue(ping_result, rp)

View File

@ -0,0 +1,146 @@
"""
Copyright 2018 Rackspace
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from cafe.drivers.unittest.decorators import tags
from cloudcafe.networking.networks.common.tools.connectivity import \
Connectivity
from cloudcafe.networking.networks.personas import ServerPersona
from cloudroast.networking.networks.fixtures import NetworkingComputeFixture
from cloudroast.networking.networks.scenario.common import ScenarioMixin
class TestConnectivity(NetworkingComputeFixture, ScenarioMixin):
"""
Testing no network connectivity across servers on different networks
"""
NAMES_PREFIX = 'connectivity'
PRIVATE_KEY_PATH = '/root/pkey'
MAX_RETRIES = 5
SSH_COMMAND = ('ssh -o UserKnownHostsFile=/dev/null '
'-o StrictHostKeyChecking=no -o ConnectTimeout=60 '
'-i {private_key_path} {user}@{ip_address}')
msg = ('Unexpected remote connection from server on {0} network'
'with server on another isolated or private network {1}. '
'Complete results: {2}')
@classmethod
def setUpClass(cls):
super(TestConnectivity, cls).setUpClass()
network_name = 'network_1_{0}'.format(cls.NAMES_PREFIX)
cls.network1 = cls.create_server_network(name=network_name, ipv4=True)
cls.delete_networks.append(cls.network1.id)
keypair_name = 'key_{0}'.format(cls.NAMES_PREFIX)
keypair = cls.create_keypair(name=keypair_name)
cls.delete_keypairs.append(keypair.name)
svr_name_1 = 'svr_1_{0}'.format(cls.NAMES_PREFIX)
network_ids = [cls.public_network_id, cls.service_network_id,
cls.network1.id]
cls.server1 = cls.create_test_server(
name=svr_name_1, key_name=keypair.name,
network_ids=network_ids, active_server=False)
cls.sp1 = ServerPersona(
server=cls.server1, pnet=True, snet=True, inet=True,
network=cls.network1, keypair=keypair, ssh_username='root')
network_name = 'network_2_{0}'.format(cls.NAMES_PREFIX)
cls.network2 = cls.create_server_network(name=network_name, ipv4=True)
svr_name_2 = 'svr_2_{0}'.format(cls.NAMES_PREFIX)
network_ids = [cls.public_network_id]
cls.server2 = cls.create_test_server(
name=svr_name_2, key_name=keypair.name,
network_ids=network_ids, active_server=False)
cls.sp2 = ServerPersona(
server=cls.server2, pnet=True, snet=False, inet=False,
network=cls.network2, keypair=keypair, ssh_username='root')
server_ids = [cls.sp1.server.id, cls.sp2.server.id]
cls.delete_servers.extend(server_ids)
cls._transfer_private_key_to_vm(cls.sp1.remote_client.ssh_client,
keypair.private_key,
cls.PRIVATE_KEY_PATH
)
cls._transfer_private_key_to_vm(cls.sp2.remote_client.ssh_client,
keypair.private_key,
cls.PRIVATE_KEY_PATH
)
@tags('connectivity', 'negative')
def test_remote_private_ping(self):
"""
Testing there is no private connectivity from servers
without private networks.
"""
self._test_remote_ping(port_type='snet')
@tags('connectivity', 'negative')
def test_remote_isolated_ping(self):
"""
Testing there is no isolated connectivity from servers
without isolated networks.
"""
self._test_remote_ping(port_type='inet')
@tags('connectivity', 'negative')
def _test_remote_ping(self, port_type):
"""
Testing no remote ping on servers from another server
which doesn't have private or isolated networks.
"""
conn = Connectivity(self.sp2, self.sp1)
icmp_basic = dict(port_type=port_type, protocol='icmp', ip_version=4)
rp = conn.verify_personas_conn(**icmp_basic)
result = rp[0]
ping_result = result['connection']
self.assertFalse(ping_result, self.msg.format(
self.network2, self.network1, rp))
@tags('connectivity', 'negative')
def test_remote_private_ssh(self):
"""
Testing ServiceNet remote ssh on servers is unavailable.
"""
self._test_remote_ssh(self.sp1.snet_fix_ipv4[0])
@tags('connectivity', 'negative')
def test_remote_isolated_ssh(self):
"""
Testing isolated network remote ssh on servers is unavailable.
"""
self._test_remote_ssh(self.sp1.inet_fix_ipv4[0])
def _test_remote_ssh(self, target_ip_addr):
"""
Testing no remote ssh on servers from another server
which doesn't have private or isolated networks.
"""
rc2 = self.sp2.remote_client
ssh_cmd = self.SSH_COMMAND.format(
private_key_path=self.PRIVATE_KEY_PATH,
user=self.sp1.ssh_username, ip_address=target_ip_addr)
stdout = None
ssh_connection_established = False
retry_count = 0
while stdout is None or not stdout.endswith('# '):
if retry_count < self.MAX_RETRIES:
output = rc2.ssh_client.execute_shell_command(ssh_cmd)
stdout = output.stdout
retry_count += 1
if retry_count == self.MAX_RETRIES:
break
if stdout.endswith('# '):
ssh_connection_established = True
self.assertFalse(ssh_connection_established, self.msg.format(
self.network2, self.network1, stdout))