Refactor Pinger class

This change refactors Pinger class by:
* removing testcase attribute in order to reduce coupling between
  testcases and Pinger instances,
* defining client namespace in Pinger.__init__ instead of assert_*
  methods because it's required by daughter change.

Change-Id: I3aefbcaaa96bcd86319b2fb938b7c87f05cc882b

diff --git a/neutron/tests/functional/agent/linux/base.py b/neutron/tests/functional/agent/linux/base.py
index 50e8e6b520f230c10197e71dfea2d882175d93c6..26e7792996eb752fb49fcfe4b46395e36a2c9cb2 100644 (file)
--- a/neutron/tests/functional/agent/linux/base.py
+++ b/neutron/tests/functional/agent/linux/base.py
@@ -20,7 +20,6 @@ from neutron.agent.linux import ovs_lib
 from neutron.agent.linux import utils
 from neutron.common import constants as n_const
 from neutron.openstack.common import uuidutils
-from neutron.tests.functional.agent.linux import helpers
 from neutron.tests.functional import base as functional_base
 from neutron.tests import sub_base
 
@@ -165,7 +164,6 @@ class BaseIPVethTestCase(BaseLinuxTestCase):
     def setUp(self):
         super(BaseIPVethTestCase, self).setUp()
         self.check_sudo_enabled()
-        self.pinger = helpers.Pinger(self)
 
     @staticmethod
     def _set_ip_up(device, cidr, broadcast, ip_version=4):

diff --git a/neutron/tests/functional/agent/linux/helpers.py b/neutron/tests/functional/agent/linux/helpers.py
index 3c0ff49401677fb422ddf7ca66db7cb12a0c1715..46a1ea2446177b301e70c092d806b4ba71660464 100644 (file)
--- a/neutron/tests/functional/agent/linux/helpers.py
+++ b/neutron/tests/functional/agent/linux/helpers.py
@@ -25,6 +25,7 @@ import eventlet
 from neutron.agent.common import config
 from neutron.agent.linux import ip_lib
 from neutron.agent.linux import utils
+from neutron.tests import tools
 
 CHILD_PROCESS_TIMEOUT = os.environ.get('OS_TEST_CHILD_PROCESS_TIMEOUT', 20)
 CHILD_PROCESS_SLEEP = os.environ.get('OS_TEST_CHILD_PROCESS_SLEEP', 0.5)
@@ -129,29 +130,24 @@ def pid_invoked_with_cmdline(pid, expected_cmd):
 
 
 class Pinger(object):
-    def __init__(self, testcase, timeout=1, max_attempts=1):
-        self.testcase = testcase
+    def __init__(self, namespace, timeout=1, max_attempts=1):
+        self.namespace = namespace
         self._timeout = timeout
         self._max_attempts = max_attempts
 
-    def _ping_destination(self, src_namespace, dest_address):
-        src_namespace.netns.execute(['ping', '-c', self._max_attempts,
-                                     '-W', self._timeout, dest_address])
+    def _ping_destination(self, dest_address):
+        self.namespace.netns.execute(['ping', '-c', self._max_attempts,
+                                      '-W', self._timeout, dest_address])
 
-    def assert_ping_from_ns(self, src_ns, dst_ip):
-        try:
-            self._ping_destination(src_ns, dst_ip)
-        except RuntimeError:
-            self.testcase.fail("destination ip %(dst_ip)s is not replying "
-                               "to ping from namespace %(src_ns)s" %
-                               {'src_ns': src_ns.namespace, 'dst_ip': dst_ip})
+    def assert_ping(self, dst_ip):
+        self._ping_destination(dst_ip)
 
-    def assert_no_ping_from_ns(self, src_ns, dst_ip):
+    def assert_no_ping(self, dst_ip):
         try:
-            self._ping_destination(src_ns, dst_ip)
-            self.testcase.fail("destination ip %(dst_ip)s is replying to ping"
-                               "from namespace %(src_ns)s, but it shouldn't" %
-                               {'src_ns': src_ns.namespace, 'dst_ip': dst_ip})
+            self._ping_destination(dst_ip)
+            tools.fail("destination ip %(dst_ip)s is replying to ping"
+                       "from namespace %(ns)s, but it shouldn't" %
+                       {'ns': self.namespace.namespace, 'dst_ip': dst_ip})
         except RuntimeError:
             pass
 

diff --git a/neutron/tests/functional/agent/linux/test_ipset.py b/neutron/tests/functional/agent/linux/test_ipset.py
index ebe45ff52f58e27b13632f95e4bf7f8eca66bdc9..499386423f1db56c78e3673698879af7ae1052fc 100644 (file)
--- a/neutron/tests/functional/agent/linux/test_ipset.py
+++ b/neutron/tests/functional/agent/linux/test_ipset.py
@@ -16,6 +16,7 @@
 from neutron.agent.linux import ipset_manager
 from neutron.agent.linux import iptables_manager
 from neutron.tests.functional.agent.linux import base
+from neutron.tests.functional.agent.linux import helpers
 
 IPSET_SET = 'test-set'
 IPSET_ETHERTYPE = 'IPv4'
@@ -36,6 +37,7 @@ class IpsetBase(base.BaseIPVethTestCase):
             namespace=self.dst_ns.namespace)
 
         self._add_iptables_ipset_rules(self.dst_iptables)
+        self.pinger = helpers.Pinger(self.src_ns)
 
     def _create_ipset_manager_and_set(self, dst_ns, set_name):
         ipset = ipset_manager.IpsetManager(
@@ -59,28 +61,28 @@ class IpsetBase(base.BaseIPVethTestCase):
 class IpsetManagerTestCase(IpsetBase):
 
     def test_add_member_allows_ping(self):
-        self.pinger.assert_no_ping_from_ns(self.src_ns, self.DST_ADDRESS)
+        self.pinger.assert_no_ping(self.DST_ADDRESS)
         self.ipset._add_member_to_set(IPSET_SET, self.SRC_ADDRESS)
-        self.pinger.assert_ping_from_ns(self.src_ns, self.DST_ADDRESS)
+        self.pinger.assert_ping(self.DST_ADDRESS)
 
     def test_del_member_denies_ping(self):
         self.ipset._add_member_to_set(IPSET_SET, self.SRC_ADDRESS)
-        self.pinger.assert_ping_from_ns(self.src_ns, self.DST_ADDRESS)
+        self.pinger.assert_ping(self.DST_ADDRESS)
 
         self.ipset._del_member_from_set(IPSET_SET, self.SRC_ADDRESS)
-        self.pinger.assert_no_ping_from_ns(self.src_ns, self.DST_ADDRESS)
+        self.pinger.assert_no_ping(self.DST_ADDRESS)
 
     def test_refresh_ipset_allows_ping(self):
         self.ipset._refresh_set(IPSET_SET, [UNRELATED_IP], IPSET_ETHERTYPE)
-        self.pinger.assert_no_ping_from_ns(self.src_ns, self.DST_ADDRESS)
+        self.pinger.assert_no_ping(self.DST_ADDRESS)
 
         self.ipset._refresh_set(IPSET_SET, [UNRELATED_IP, self.SRC_ADDRESS],
                                 IPSET_ETHERTYPE)
-        self.pinger.assert_ping_from_ns(self.src_ns, self.DST_ADDRESS)
+        self.pinger.assert_ping(self.DST_ADDRESS)
 
         self.ipset._refresh_set(IPSET_SET, [self.SRC_ADDRESS, UNRELATED_IP],
                                 IPSET_ETHERTYPE)
-        self.pinger.assert_ping_from_ns(self.src_ns, self.DST_ADDRESS)
+        self.pinger.assert_ping(self.DST_ADDRESS)
 
     def test_destroy_ipset_set(self):
         self.assertRaises(RuntimeError, self.ipset._destroy, IPSET_SET)

diff --git a/neutron/tests/functional/agent/linux/test_iptables.py b/neutron/tests/functional/agent/linux/test_iptables.py
index 684c7f6869688cf11a47e5cce343875ac2695d7c..eb9c828a5f5dcace3750afa53faccab8fa77c963 100644 (file)
--- a/neutron/tests/functional/agent/linux/test_iptables.py
+++ b/neutron/tests/functional/agent/linux/test_iptables.py
@@ -83,27 +83,29 @@ class IptablesManagerTestCase(base.BaseIPVethTestCase):
         self.assertTrue(netcat.test_connectivity(True))
 
     def test_icmp(self):
-        self.pinger.assert_ping_from_ns(self.client_ns, self.DST_ADDRESS)
+        pinger = helpers.Pinger(self.client_ns)
+        pinger.assert_ping(self.DST_ADDRESS)
         self.server_fw.ipv4['filter'].add_rule('INPUT', base.ICMP_BLOCK_RULE)
         self.server_fw.apply()
-        self.pinger.assert_no_ping_from_ns(self.client_ns, self.DST_ADDRESS)
+        pinger.assert_no_ping(self.DST_ADDRESS)
         self.server_fw.ipv4['filter'].remove_rule('INPUT',
                                                  base.ICMP_BLOCK_RULE)
         self.server_fw.apply()
-        self.pinger.assert_ping_from_ns(self.client_ns, self.DST_ADDRESS)
+        pinger.assert_ping(self.DST_ADDRESS)
 
     def test_mangle_icmp(self):
-        self.pinger.assert_ping_from_ns(self.client_ns, self.DST_ADDRESS)
+        pinger = helpers.Pinger(self.client_ns)
+        pinger.assert_ping(self.DST_ADDRESS)
         self.server_fw.ipv4['mangle'].add_rule('INPUT', base.ICMP_MARK_RULE)
         self.server_fw.ipv4['filter'].add_rule('INPUT', base.MARKED_BLOCK_RULE)
         self.server_fw.apply()
-        self.pinger.assert_no_ping_from_ns(self.client_ns, self.DST_ADDRESS)
+        pinger.assert_no_ping(self.DST_ADDRESS)
         self.server_fw.ipv4['mangle'].remove_rule('INPUT',
                                                   base.ICMP_MARK_RULE)
         self.server_fw.ipv4['filter'].remove_rule('INPUT',
                                                   base.MARKED_BLOCK_RULE)
         self.server_fw.apply()
-        self.pinger.assert_ping_from_ns(self.client_ns, self.DST_ADDRESS)
+        pinger.assert_ping(self.DST_ADDRESS)
 
     def test_tcp_input_port(self):
         self._test_with_nc(self.server_fw, 'ingress', self.port, udp=False)

diff --git a/neutron/tests/tools.py b/neutron/tests/tools.py
index 682ce5afa2d8ccef677c7ef09c9d8a9b588237ad..59381ea739ce4ea6c8682ad42b34f8d52416f6d9 100644 (file)
--- a/neutron/tests/tools.py
+++ b/neutron/tests/tools.py
@@ -34,6 +34,8 @@ expected_calls_and_values is a list of (expected_call, return_value):
   A return value or an exception can be specified.
 """
 
+import unittest
+
 
 def setup_mock_calls(mocked_call, expected_calls_and_values):
     return_values = [call[1] for call in expected_calls_and_values]
@@ -44,3 +46,12 @@ def verify_mock_calls(mocked_call, expected_calls_and_values,
                       any_order=False):
     expected_calls = [call[0] for call in expected_calls_and_values]
     mocked_call.assert_has_calls(expected_calls, any_order=any_order)
+
+
+def fail(msg=None):
+    """Fail immediatly, with the given message.
+
+    This method is equivalent to TestCase.fail without requiring a
+    testcase instance (usefully for reducing coupling).
+    """
+    raise unittest.TestCase.failureException(msg)