Code Sync from neutron project to newly created neutron-tempest-plugin
* The following commit sync the code from following hash:
start_hash: 7279aa35851110a4933a10b58b2758a2bc3933a3
end_hash: 6e911a49a9e630878f4c46f61fde3964be550880
Change-Id: I371aa4d5f043f695df04b98b0f485c8f0548f2b3
diff --git a/neutron_tempest_plugin/scenario/base.py b/neutron_tempest_plugin/scenario/base.py
index e810490..5cc085f 100644
--- a/neutron_tempest_plugin/scenario/base.py
+++ b/neutron_tempest_plugin/scenario/base.py
@@ -12,6 +12,7 @@
# 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 subprocess
import netaddr
from oslo_log import log
@@ -173,7 +174,7 @@
LOG.debug("Created subnet %s", self.subnet['id'])
secgroup = self.os_primary.network_client.create_security_group(
- name=data_utils.rand_name('secgroup-'))
+ name=data_utils.rand_name('secgroup'))
LOG.debug("Created security group %s",
secgroup['security_group']['name'])
self.security_groups.append(secgroup['security_group'])
@@ -276,3 +277,41 @@
nic=None, mtu=None, fragmentation=True):
self.assertTrue(self._check_remote_connectivity(
source, dest, should_succeed, nic, mtu, fragmentation))
+
+ def ping_ip_address(self, ip_address, should_succeed=True,
+ ping_timeout=None, mtu=None):
+ # the code is taken from tempest/scenario/manager.py in tempest git
+ timeout = ping_timeout or CONF.validation.ping_timeout
+ cmd = ['ping', '-c1', '-w1']
+
+ if mtu:
+ cmd += [
+ # don't fragment
+ '-M', 'do',
+ # ping receives just the size of ICMP payload
+ '-s', str(net_utils.get_ping_payload_size(mtu, 4))
+ ]
+ cmd.append(ip_address)
+
+ def ping():
+ proc = subprocess.Popen(cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
+ proc.communicate()
+
+ return (proc.returncode == 0) == should_succeed
+
+ caller = test_utils.find_test_caller()
+ LOG.debug('%(caller)s begins to ping %(ip)s in %(timeout)s sec and the'
+ ' expected result is %(should_succeed)s', {
+ 'caller': caller, 'ip': ip_address, 'timeout': timeout,
+ 'should_succeed':
+ 'reachable' if should_succeed else 'unreachable'
+ })
+ result = test_utils.call_until_true(ping, timeout, 1)
+ LOG.debug('%(caller)s finishes ping %(ip)s in %(timeout)s sec and the '
+ 'ping result is %(result)s', {
+ 'caller': caller, 'ip': ip_address, 'timeout': timeout,
+ 'result': 'expected' if result else 'unexpected'
+ })
+ return result