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:
parent
0d0da0f8c8
commit
53918afca0
|
@ -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
|
||||
|
|
|
@ -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))
|
|
@ -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(
|
||||
|
|
|
@ -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)
|
|
@ -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))
|
Loading…
Reference in New Issue