Add OpenFixture and get rid of 'open' mocks

This patch adds a new fixture that allows a specific
file to be mocked for 'open' calls while leaving the
normal 'open' behavior for everything else. This is to
prevent cases like the related bug where a library we
depend on can unexpectedly call 'open' on its own
(in that case it was debtcollector->warnings->linecache).

It also replaces all open mocks in the unit tests with
this new fixture.

Related-Bug: #1503847
Change-Id: Ie9e1a01acd5cb231718efd028d30df6c8dbcd3f0

diff --git a/neutron/tests/tools.py b/neutron/tests/tools.py
index 496dfbe5382883e5dfa353c3708557395d11da8d..28f0303e132752f9c962ef09ded6ef34af46daad 100644 (file)
--- a/neutron/tests/tools.py
+++ b/neutron/tests/tools.py
@@ -13,6 +13,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import mock
 import random
 import string
 import warnings
@@ -67,6 +68,28 @@ class WarningsFixture(fixtures.Fixture):
                 "always", category=wtype, module='^neutron\\.')
 
 
+class OpenFixture(fixtures.Fixture):
+    """Mock access to a specific file while preserving open for others."""
+
+    def __init__(self, filepath, contents=''):
+        self.path = filepath
+        self.contents = contents
+
+    def _setUp(self):
+        self.mock_open = mock.mock_open(read_data=self.contents)
+        self._orig_open = open
+
+        def replacement_open(name, *args, **kwargs):
+            if name == self.path:
+                return self.mock_open(name, *args, **kwargs)
+            return self._orig_open(name, *args, **kwargs)
+
+        self._patch = mock.patch('six.moves.builtins.open',
+                                 new=replacement_open)
+        self._patch.start()
+        self.addCleanup(self._patch.stop)
+
+
 class SafeCleanupFixture(fixtures.Fixture):
     """Catch errors in daughter fixture cleanup."""
 

diff --git a/neutron/tests/unit/agent/linux/test_daemon.py b/neutron/tests/unit/agent/linux/test_daemon.py
index c50b4c8762442f37ec715a7a886c9ff5194ab75a..749b1e264d6b846ae9bcea257841c8e484eb9d6d 100644 (file)
--- a/neutron/tests/unit/agent/linux/test_daemon.py
+++ b/neutron/tests/unit/agent/linux/test_daemon.py
@@ -23,6 +23,7 @@ import testtools
 from neutron.agent.linux import daemon
 from neutron.common import exceptions
 from neutron.tests import base
+from neutron.tests import tools
 
 FAKE_FD = 8
 
@@ -171,43 +172,37 @@ class TestPidfile(base.BaseTestCase):
         self.assertEqual(34, p.read())
 
     def test_is_running(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            p = daemon.Pidfile('thefile', 'python')
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.readline.return_value = 'python'
+        mock_open = self.useFixture(
+            tools.OpenFixture('/proc/34/cmdline', 'python')).mock_open
+        p = daemon.Pidfile('thefile', 'python')
 
-            with mock.patch.object(p, 'read') as read:
-                read.return_value = 34
-                self.assertTrue(p.is_running())
+        with mock.patch.object(p, 'read') as read:
+            read.return_value = 34
+            self.assertTrue(p.is_running())
 
-            mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
+        mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
 
     def test_is_running_uuid_true(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            p = daemon.Pidfile('thefile', 'python', uuid='1234')
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.readline.return_value = 'python 1234'
+        mock_open = self.useFixture(
+            tools.OpenFixture('/proc/34/cmdline', 'python 1234')).mock_open
+        p = daemon.Pidfile('thefile', 'python', uuid='1234')
 
-            with mock.patch.object(p, 'read') as read:
-                read.return_value = 34
-                self.assertTrue(p.is_running())
+        with mock.patch.object(p, 'read') as read:
+            read.return_value = 34
+            self.assertTrue(p.is_running())
 
-            mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
+        mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
 
     def test_is_running_uuid_false(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            p = daemon.Pidfile('thefile', 'python', uuid='6789')
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.readline.return_value = 'python 1234'
+        mock_open = self.useFixture(
+            tools.OpenFixture('/proc/34/cmdline', 'python 1234')).mock_open
+        p = daemon.Pidfile('thefile', 'python', uuid='6789')
 
-            with mock.patch.object(p, 'read') as read:
-                read.return_value = 34
-                self.assertFalse(p.is_running())
+        with mock.patch.object(p, 'read') as read:
+            read.return_value = 34
+            self.assertFalse(p.is_running())
 
-            mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
+        mock_open.assert_called_once_with('/proc/34/cmdline', 'r')
 
 
 class TestDaemon(base.BaseTestCase):

diff --git a/neutron/tests/unit/agent/linux/test_dhcp.py b/neutron/tests/unit/agent/linux/test_dhcp.py
index 1a25247837cc130cf04585eb6f1e999ed41dc39e..847d7a04f1ea4eb13c8a779433ab47ca57786f6e 100644 (file)
--- a/neutron/tests/unit/agent/linux/test_dhcp.py
+++ b/neutron/tests/unit/agent/linux/test_dhcp.py
@@ -29,6 +29,7 @@ from neutron.common import constants
 from neutron.common import utils
 from neutron.extensions import extra_dhcp_opt as edo_ext
 from neutron.tests import base
+from neutron.tests import tools
 
 LOG = logging.getLogger(__name__)
 
@@ -985,12 +986,11 @@ class TestDhcpLocalProcess(TestBase):
         parent.assert_has_calls(expected)
 
     def test_get_interface_name(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.read.return_value = 'tap0'
-            lp = LocalChild(self.conf, FakeDualNetwork())
-            self.assertEqual(lp.interface_name, 'tap0')
+        net = FakeDualNetwork()
+        path = '/dhcp/%s/interface' % net.id
+        self.useFixture(tools.OpenFixture(path, 'tap0'))
+        lp = LocalChild(self.conf, net)
+        self.assertEqual(lp.interface_name, 'tap0')
 
     def test_set_interface_name(self):
         with mock.patch('neutron.agent.linux.utils.replace_file') as replace:
@@ -1557,23 +1557,23 @@ class TestDnsmasq(TestBase):
          exp_addn_name, exp_addn_data,
          exp_opt_name, exp_opt_data,) = self._test_reload_allocation_data
 
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.readline.return_value = None
-
-            test_pm = mock.Mock()
-            dm = self._get_dnsmasq(FakeDualNetwork(), test_pm)
-            dm.reload_allocations()
-            self.assertTrue(test_pm.register.called)
-            self.external_process().enable.assert_called_once_with(
-                reload_cfg=True)
-
-            self.safe.assert_has_calls([
-                mock.call(exp_host_name, exp_host_data),
-                mock.call(exp_addn_name, exp_addn_data),
-                mock.call(exp_opt_name, exp_opt_data),
-            ])
+        net = FakeDualNetwork()
+        hpath = '/dhcp/%s/host' % net.id
+        ipath = '/dhcp/%s/interface' % net.id
+        self.useFixture(tools.OpenFixture(hpath))
+        self.useFixture(tools.OpenFixture(ipath))
+        test_pm = mock.Mock()
+        dm = self._get_dnsmasq(net, test_pm)
+        dm.reload_allocations()
+        self.assertTrue(test_pm.register.called)
+        self.external_process().enable.assert_called_once_with(
+            reload_cfg=True)
+
+        self.safe.assert_has_calls([
+            mock.call(exp_host_name, exp_host_data),
+            mock.call(exp_addn_name, exp_addn_data),
+            mock.call(exp_opt_name, exp_opt_data),
+        ])
 
     def test_release_unused_leases(self):
         dnsmasq = self._get_dnsmasq(FakeDualNetwork())
@@ -1699,15 +1699,12 @@ class TestDnsmasq(TestBase):
 
     def test_read_hosts_file_leases(self):
         filename = '/path/to/file'
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            lines = ["00:00:80:aa:bb:cc,inst-name,192.168.0.1",
-                     "00:00:80:aa:bb:cc,inst-name,[fdca:3ba5:a17a::1]"]
-            mock_open.return_value.readlines.return_value = lines
-
-            dnsmasq = self._get_dnsmasq(FakeDualNetwork())
-            leases = dnsmasq._read_hosts_file_leases(filename)
+        lines = ["00:00:80:aa:bb:cc,inst-name,192.168.0.1",
+                 "00:00:80:aa:bb:cc,inst-name,[fdca:3ba5:a17a::1]"]
+        mock_open = self.useFixture(
+            tools.OpenFixture(filename, '\n'.join(lines))).mock_open
+        dnsmasq = self._get_dnsmasq(FakeDualNetwork())
+        leases = dnsmasq._read_hosts_file_leases(filename)
 
         self.assertEqual(set([("192.168.0.1", "00:00:80:aa:bb:cc", None),
                               ("fdca:3ba5:a17a::1", "00:00:80:aa:bb:cc",
@@ -1716,16 +1713,13 @@ class TestDnsmasq(TestBase):
 
     def test_read_hosts_file_leases_with_client_id(self):
         filename = '/path/to/file'
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            lines = ["00:00:80:aa:bb:cc,id:client1,inst-name,192.168.0.1",
-                     "00:00:80:aa:bb:cc,id:client2,inst-name,"
-                     "[fdca:3ba5:a17a::1]"]
-            mock_open.return_value.readlines.return_value = lines
-
-            dnsmasq = self._get_dnsmasq(FakeDualNetwork())
-            leases = dnsmasq._read_hosts_file_leases(filename)
+        lines = ["00:00:80:aa:bb:cc,id:client1,inst-name,192.168.0.1",
+                 "00:00:80:aa:bb:cc,id:client2,inst-name,"
+                 "[fdca:3ba5:a17a::1]"]
+        mock_open = self.useFixture(
+            tools.OpenFixture(filename, '\n'.join(lines))).mock_open
+        dnsmasq = self._get_dnsmasq(FakeDualNetwork())
+        leases = dnsmasq._read_hosts_file_leases(filename)
 
         self.assertEqual(set([("192.168.0.1", "00:00:80:aa:bb:cc", 'client1'),
                               ("fdca:3ba5:a17a::1", "00:00:80:aa:bb:cc",

diff --git a/neutron/tests/unit/agent/linux/test_external_process.py b/neutron/tests/unit/agent/linux/test_external_process.py
index 4cc3836306dbdfb7d512a1a78f59ceafeca95095..a94774d7a507ffe106ac72dd5943e36f7d7ad7af 100644 (file)
--- a/neutron/tests/unit/agent/linux/test_external_process.py
+++ b/neutron/tests/unit/agent/linux/test_external_process.py
@@ -18,6 +18,7 @@ import os.path
 from neutron.agent.linux import external_process as ep
 from neutron.common import utils as common_utils
 from neutron.tests import base
+from neutron.tests import tools
 
 
 TEST_UUID = 'test-uuid'
@@ -215,20 +216,14 @@ class TestProcessManager(base.BaseTestCase):
         self.assertEqual(retval, '/var/path/uuid.pid')
 
     def test_pid(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.read.return_value = '5'
-            manager = ep.ProcessManager(self.conf, 'uuid')
-            self.assertEqual(manager.pid, 5)
+        self.useFixture(tools.OpenFixture('/var/path/uuid.pid', '5'))
+        manager = ep.ProcessManager(self.conf, 'uuid')
+        self.assertEqual(manager.pid, 5)
 
     def test_pid_no_an_int(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.read.return_value = 'foo'
-            manager = ep.ProcessManager(self.conf, 'uuid')
-            self.assertIsNone(manager.pid, 5)
+        self.useFixture(tools.OpenFixture('/var/path/uuid.pid', 'foo'))
+        manager = ep.ProcessManager(self.conf, 'uuid')
+        self.assertIsNone(manager.pid)
 
     def test_pid_invalid_file(self):
         with mock.patch.object(ep.ProcessManager, 'get_pid_file_name') as name:
@@ -237,17 +232,15 @@ class TestProcessManager(base.BaseTestCase):
             self.assertIsNone(manager.pid)
 
     def test_active(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.readline.return_value = \
-                'python foo --router_id=uuid'
-            with mock.patch.object(ep.ProcessManager, 'pid') as pid:
-                pid.__get__ = mock.Mock(return_value=4)
-                manager = ep.ProcessManager(self.conf, 'uuid')
-                self.assertTrue(manager.active)
+        mock_open = self.useFixture(
+            tools.OpenFixture('/proc/4/cmdline', 'python foo --router_id=uuid')
+        ).mock_open
+        with mock.patch.object(ep.ProcessManager, 'pid') as pid:
+            pid.__get__ = mock.Mock(return_value=4)
+            manager = ep.ProcessManager(self.conf, 'uuid')
+            self.assertTrue(manager.active)
 
-            mock_open.assert_called_once_with('/proc/4/cmdline', 'r')
+        mock_open.assert_called_once_with('/proc/4/cmdline', 'r')
 
     def test_active_none(self):
         dummy_cmd_line = 'python foo --router_id=uuid'
@@ -258,14 +251,13 @@ class TestProcessManager(base.BaseTestCase):
             self.assertFalse(manager.active)
 
     def test_active_cmd_mismatch(self):
-        with mock.patch('six.moves.builtins.open') as mock_open:
-            mock_open.return_value.__enter__ = lambda s: s
-            mock_open.return_value.__exit__ = mock.Mock()
-            mock_open.return_value.readline.return_value = \
-                'python foo --router_id=anotherid'
-            with mock.patch.object(ep.ProcessManager, 'pid') as pid:
-                pid.__get__ = mock.Mock(return_value=4)
-                manager = ep.ProcessManager(self.conf, 'uuid')
-                self.assertFalse(manager.active)
+        mock_open = self.useFixture(
+            tools.OpenFixture('/proc/4/cmdline',
+                              'python foo --router_id=anotherid')
+        ).mock_open
+        with mock.patch.object(ep.ProcessManager, 'pid') as pid:
+            pid.__get__ = mock.Mock(return_value=4)
+            manager = ep.ProcessManager(self.conf, 'uuid')
+            self.assertFalse(manager.active)
 
-            mock_open.assert_called_once_with('/proc/4/cmdline', 'r')
+        mock_open.assert_called_once_with('/proc/4/cmdline', 'r')

diff --git a/neutron/tests/unit/common/test_ipv6_utils.py b/neutron/tests/unit/common/test_ipv6_utils.py
index e7d77c5abb002f17f74c5ec185e91b5b2811f994..4f891a14afed062424e4ccc5cac0395f3900b52a 100644 (file)
--- a/neutron/tests/unit/common/test_ipv6_utils.py
+++ b/neutron/tests/unit/common/test_ipv6_utils.py
@@ -19,6 +19,7 @@ import mock
 from neutron.common import constants
 from neutron.common import ipv6_utils
 from neutron.tests import base
+from neutron.tests import tools
 
 
 class IPv6byEUI64TestCase(base.BaseTestCase):
@@ -65,31 +66,31 @@ class TestIsEnabled(base.BaseTestCase):
         self.addCleanup(reset_detection_flag)
         self.mock_exists = mock.patch("os.path.exists",
                                       return_value=True).start()
-        mock_open = mock.patch("six.moves.builtins.open").start()
-        self.mock_read = mock_open.return_value.__enter__.return_value.read
+        self.proc_path = '/proc/sys/net/ipv6/conf/default/disable_ipv6'
 
     def test_enabled(self):
-        self.mock_read.return_value = "0"
+        self.useFixture(tools.OpenFixture(self.proc_path, '0'))
         enabled = ipv6_utils.is_enabled()
         self.assertTrue(enabled)
 
     def test_disabled(self):
-        self.mock_read.return_value = "1"
+        self.useFixture(tools.OpenFixture(self.proc_path, '1'))
         enabled = ipv6_utils.is_enabled()
         self.assertFalse(enabled)
 
     def test_disabled_non_exists(self):
+        mo = self.useFixture(tools.OpenFixture(self.proc_path, '1')).mock_open
         self.mock_exists.return_value = False
         enabled = ipv6_utils.is_enabled()
         self.assertFalse(enabled)
-        self.assertFalse(self.mock_read.called)
+        self.assertFalse(mo.called)
 
     def test_memoize(self):
-        self.mock_read.return_value = "0"
+        mo = self.useFixture(tools.OpenFixture(self.proc_path, '0')).mock_open
         ipv6_utils.is_enabled()
         enabled = ipv6_utils.is_enabled()
         self.assertTrue(enabled)
-        self.mock_read.assert_called_once_with()
+        mo.assert_called_once_with(self.proc_path, 'r')
 
 
 class TestIsAutoAddressSubnet(base.BaseTestCase):

diff --git a/neutron/tests/unit/db/test_migration.py b/neutron/tests/unit/db/test_migration.py
index dda1cbce28e84ce0a22427dffc5c705c409afdfd..62b794bedc3a70846484528c2db9446af107eb1d 100644 (file)
--- a/neutron/tests/unit/db/test_migration.py
+++ b/neutron/tests/unit/db/test_migration.py
@@ -337,35 +337,31 @@ class TestCli(base.BaseTestCase):
         if file_heads is None:
             file_heads = []
         fake_config = self.configs[0]
+        mock_open = self.useFixture(
+            tools.OpenFixture(cli._get_head_file_path(fake_config),
+                              '\n'.join(file_heads))
+        ).mock_open
         with mock.patch('alembic.script.ScriptDirectory.from_config') as fc,\
                 mock.patch.object(cli, '_use_separate_migration_branches',
                                   return_value=not branchless):
             fc.return_value.get_heads.return_value = heads
-            with mock.patch.object(cli, 'open') as mock_open:
-                mock_open.return_value.__enter__ = lambda s: s
-                mock_open.return_value.__exit__ = mock.Mock()
-                mock_open.return_value.read.return_value = (
-                    '\n'.join(file_heads))
-
-                if not branchless or all(head in file_heads for head in heads):
-                    cli.validate_head_file(fake_config)
-                else:
-                    self.assertRaises(
-                        SystemExit,
-                        cli.validate_head_file,
-                        fake_config
-                    )
-                    self.assertTrue(self.mock_alembic_err.called)
 
-                if branchless:
-                    mock_open.assert_called_with(
-                        cli._get_head_file_path(fake_config))
-                else:
-                    self.assertFalse(mock_open.called)
+            if not branchless or all(head in file_heads for head in heads):
+                cli.validate_head_file(fake_config)
+            else:
+                self.assertRaises(
+                    SystemExit,
+                    cli.validate_head_file,
+                    fake_config
+                )
+                self.assertTrue(self.mock_alembic_err.called)
 
             if branchless:
+                mock_open.assert_called_with(
+                    cli._get_head_file_path(fake_config))
                 fc.assert_called_once_with(fake_config)
             else:
+                self.assertFalse(mock_open.called)
                 self.assertFalse(fc.called)
 
     def test_validate_head_file_multiple_heads(self):