From: Cedric Brandily Date: Sat, 20 Dec 2014 22:27:30 +0000 (+0100) Subject: Refactor Pinger class X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=753d3162b52d700d20f83c0416ac7a196f68db70;p=openstack-build%2Fneutron-build.git 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 50e8e6b52..26e779299 100644 --- 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 3c0ff4940..46a1ea244 100644 --- 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 ebe45ff52..499386423 100644 --- 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 684c7f686..eb9c828a5 100644 --- 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 682ce5afa..59381ea73 100644 --- 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)