Merge "Replace mock open with namespace builtins with OpenFixture"

2016-04-14 19:11:44 +00:00
parent cf97283c1c 66b1f2bf11
commit 5d5dcf9951
4 changed files with 143 additions and 133 deletions
--- a/octavia/tests/unit/amphorae/backends/agent/api_server/test_listener.py
+++ b/octavia/tests/unit/amphorae/backends/agent/api_server/test_listener.py
@@ -14,7 +14,6 @@

 import mock
 from oslo_utils import uuidutils
-import six.moves.builtins as builtins

 from octavia.amphorae.backends.agent.api_server import listener
 from octavia.amphorae.backends.agent.api_server import util as agent_util
@@ -46,16 +45,16 @@ class ListenerTestCase(base.TestCase):
                                                 tls=True, sni=True),
            tls_tupe)

-        m = mock.mock_open(read_data=rendered_obj)
+        path = agent_util.config_path(LISTENER_ID1)
+        self.useFixture(test_utils.OpenFixture(path, rendered_obj))

-        with mock.patch.object(builtins, 'open', m, create=True):
-            res = listener._parse_haproxy_file('123')
-            self.assertEqual('TERMINATED_HTTPS', res['mode'])
-            self.assertEqual('/var/lib/octavia/sample_listener_id_1.sock',
-                             res['stats_socket'])
-            self.assertEqual(
-                '/var/lib/octavia/certs/sample_listener_id_1/FakeCN.pem',
-                res['ssl_crt'])
+        res = listener._parse_haproxy_file(LISTENER_ID1)
+        self.assertEqual('TERMINATED_HTTPS', res['mode'])
+        self.assertEqual('/var/lib/octavia/sample_listener_id_1.sock',
+                         res['stats_socket'])
+        self.assertEqual(
+            '/var/lib/octavia/certs/sample_listener_id_1/FakeCN.pem',
+            res['ssl_crt'])

        # render_template_tls_no_sni
        rendered_obj = self.jinja_cfg.render_loadbalancer_obj(
@@ -66,50 +65,46 @@ class ListenerTestCase(base.TestCase):
                private_key='ImAsdlfksdjPrivateKey',
                primary_cn="FakeCN"))

-        m = mock.mock_open(read_data=rendered_obj)
+        self.useFixture(test_utils.OpenFixture(path, rendered_obj))

-        with mock.patch.object(builtins, 'open', m, create=True):
-            res = listener._parse_haproxy_file('123')
-            self.assertEqual('TERMINATED_HTTPS', res['mode'])
-            self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
-                             res['stats_socket'])
-            self.assertEqual(
-                BASE_CRT_PATH + '/sample_listener_id_1/FakeCN.pem',
-                res['ssl_crt'])
+        res = listener._parse_haproxy_file(LISTENER_ID1)
+        self.assertEqual('TERMINATED_HTTPS', res['mode'])
+        self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
+                         res['stats_socket'])
+        self.assertEqual(
+            BASE_CRT_PATH + '/sample_listener_id_1/FakeCN.pem',
+            res['ssl_crt'])

        # render_template_http
        rendered_obj = self.jinja_cfg.render_loadbalancer_obj(
            sample_configs.sample_listener_tuple())
-        m = mock.mock_open(read_data=rendered_obj)

-        with mock.patch.object(builtins, 'open', m, create=True):
-            res = listener._parse_haproxy_file('123')
-            self.assertEqual('HTTP', res['mode'])
-            self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
-                             res['stats_socket'])
-            self.assertIsNone(res['ssl_crt'])
+        self.useFixture(test_utils.OpenFixture(path, rendered_obj))
+
+        res = listener._parse_haproxy_file(LISTENER_ID1)
+        self.assertEqual('HTTP', res['mode'])
+        self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
+                         res['stats_socket'])
+        self.assertIsNone(res['ssl_crt'])

        # template_https
        rendered_obj = self.jinja_cfg.render_loadbalancer_obj(
            sample_configs.sample_listener_tuple(proto='HTTPS'))
-        m = mock.mock_open(read_data=rendered_obj)
+        self.useFixture(test_utils.OpenFixture(path, rendered_obj))

-        with mock.patch.object(builtins, 'open', m, create=True):
-            res = listener._parse_haproxy_file('123')
-            self.assertEqual('TCP', res['mode'])
-            self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
-                             res['stats_socket'])
-            self.assertIsNone(res['ssl_crt'])
+        res = listener._parse_haproxy_file(LISTENER_ID1)
+        self.assertEqual('TCP', res['mode'])
+        self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
+                         res['stats_socket'])
+        self.assertIsNone(res['ssl_crt'])

        # Bogus format
-        m = mock.mock_open(read_data='Bogus')
-
-        with mock.patch.object(builtins, 'open', m, create=True):
-            try:
-                res = listener._parse_haproxy_file('123')
-                self.fail("No Exception?")
-            except listener.ParsingError:
-                pass
+        self.useFixture(test_utils.OpenFixture(path, 'Bogus'))
+        try:
+            res = listener._parse_haproxy_file(LISTENER_ID1)
+            self.fail("No Exception?")
+        except listener.ParsingError:
+            pass

    @mock.patch('os.path.exists')
    @mock.patch('octavia.amphorae.backends.agent.api_server'
@@ -145,22 +140,24 @@ class ListenerTestCase(base.TestCase):
    def test_vrrp_check_script_update(self, mock_sock_path, mock_get_listeners,
                                      mock_join, mock_listdir, mock_exists,
                                      mock_makedirs):
-        mock_get_listeners.return_value = ['abc', '123']
+        mock_get_listeners.return_value = ['abc', LISTENER_ID1]
        mock_sock_path.return_value = 'listener.sock'
        mock_exists.return_value = False
        cmd = 'haproxy-vrrp-check ' + ' '.join(['listener.sock']) + '; exit $?'
-        m = mock.mock_open()
-        with mock.patch.object(builtins, 'open', m, create=True):
-            listener.vrrp_check_script_update('123', 'stop')
+
+        path = agent_util.keepalived_dir()
+        m = self.useFixture(test_utils.OpenFixture(path)).mock_open
+
+        listener.vrrp_check_script_update(LISTENER_ID1, 'stop')
        handle = m()
        handle.write.assert_called_once_with(cmd)

-        mock_get_listeners.return_value = ['abc', '123']
+        mock_get_listeners.return_value = ['abc', LISTENER_ID1]
        cmd = ('haproxy-vrrp-check ' + ' '.join(['listener.sock',
                                                 'listener.sock']) + '; exit '
                                                                     '$?')
-        m = mock.mock_open()
-        with mock.patch.object(builtins, 'open', m, create=True):
-            listener.vrrp_check_script_update('123', 'start')
+
+        m = self.useFixture(test_utils.OpenFixture(path)).mock_open
+        listener.vrrp_check_script_update(LISTENER_ID1, 'start')
        handle = m()
        handle.write.assert_called_once_with(cmd)
--- a/octavia/tests/unit/controller/worker/tasks/test_compute_tasks.py
+++ b/octavia/tests/unit/controller/worker/tasks/test_compute_tasks.py
@@ -17,14 +17,14 @@ import mock
 from oslo_config import cfg
 from oslo_config import fixture as oslo_fixture
 from oslo_utils import uuidutils
-import six.moves.builtins as builtins
-

 from octavia.common import constants
 from octavia.common import exceptions
 from octavia.controller.worker.tasks import compute_tasks
+from octavia.tests.common import utils as test_utils
 import octavia.tests.unit.base as base

+
 AMP_FLAVOR_ID = 10
 AMP_IMAGE_ID = 11
 AMP_IMAGE_TAG = 'glance_tag'
@@ -257,44 +257,45 @@ class TestComputeTasks(base.TestCase):
                'build_agent_config', return_value='test_conf')
    @mock.patch('stevedore.driver.DriverManager.driver')
    def test_compute_create_cert(self, mock_driver, mock_conf, mock_jinja):
-
        createcompute = compute_tasks.CertComputeCreate()

        mock_driver.build.return_value = COMPUTE_ID
-        m = mock.mock_open(read_data='test')
-        with mock.patch.object(builtins, 'open', m, create=True):
-            # Test execute()
-            compute_id = createcompute.execute(_amphora_mock.id, 'test_cert',
-                                               server_group_id=SERVER_GRPOUP_ID
-                                               )
+        path = '/etc/octavia/certs/ca_01.pem'
+        self.useFixture(test_utils.OpenFixture(path, 'test'))

-            # Validate that the build method was called properly
-            mock_driver.build.assert_called_once_with(
-                name="amphora-" + _amphora_mock.id,
-                amphora_flavor=AMP_FLAVOR_ID,
-                image_id=AMP_IMAGE_ID,
-                image_tag=AMP_IMAGE_TAG,
-                key_name=AMP_SSH_KEY_NAME,
-                sec_groups=AMP_SEC_GROUPS,
-                network_ids=[AMP_NET],
-                port_ids=[],
-                user_data=None,
-                config_drive_files={
-                    '/etc/octavia/certs/server.pem': 'test_cert',
-                    '/etc/octavia/certs/client_ca.pem': 'test',
-                    '/etc/octavia/amphora-agent.conf': 'test_conf'},
-                server_group_id=SERVER_GRPOUP_ID)
+        # Test execute()
+        compute_id = createcompute.execute(_amphora_mock.id, 'test_cert',
+                                           server_group_id=SERVER_GRPOUP_ID
+                                           )
+
+        # Validate that the build method was called properly
+        mock_driver.build.assert_called_once_with(
+            name="amphora-" + _amphora_mock.id,
+            amphora_flavor=AMP_FLAVOR_ID,
+            image_id=AMP_IMAGE_ID,
+            image_tag=AMP_IMAGE_TAG,
+            key_name=AMP_SSH_KEY_NAME,
+            sec_groups=AMP_SEC_GROUPS,
+            network_ids=[AMP_NET],
+            port_ids=[],
+            user_data=None,
+            config_drive_files={
+                '/etc/octavia/certs/server.pem': 'test_cert',
+                '/etc/octavia/certs/client_ca.pem': 'test',
+                '/etc/octavia/amphora-agent.conf': 'test_conf'},
+            server_group_id=SERVER_GRPOUP_ID)

        # Make sure it returns the expected compute_id
-        assert(compute_id == COMPUTE_ID)
+        assert (compute_id == COMPUTE_ID)

        # Test that a build exception is raised
-        with mock.patch.object(builtins, 'open', m, create=True):
-            createcompute = compute_tasks.ComputeCreate()
-            self.assertRaises(TypeError,
-                              createcompute.execute,
-                              _amphora_mock,
-                              config_drive_files='test_cert')
+        self.useFixture(test_utils.OpenFixture(path, 'test'))
+
+        createcompute = compute_tasks.ComputeCreate()
+        self.assertRaises(TypeError,
+                          createcompute.execute,
+                          _amphora_mock,
+                          config_drive_files='test_cert')

        # Test revert()

@@ -304,8 +305,7 @@ class TestComputeTasks(base.TestCase):
        createcompute.revert(compute_id, _amphora_mock.id)

        # Validate that the delete method was called properly
-        mock_driver.delete.assert_called_once_with(
-            COMPUTE_ID)
+        mock_driver.delete.assert_called_once_with(COMPUTE_ID)

        # Test that a delete exception is not raised