From 2892744d5b7bd4e440e7d0aedcb836811a04ae6b Mon Sep 17 00:00:00 2001 From: Jamie Lennox Date: Mon, 27 Jun 2016 13:45:59 +1000 Subject: [PATCH] Use the mocking fixture in notifier tests All the notifier tests end up mocking out the notifier object so lets just do it once for all tests. This also splits out the test that tries to use the logging notifier because it's a different setup. Change-Id: I159a2d34fe833b0bbac70c41d521bbe1dccbc09c --- .../tests/unit/audit/test_audit_middleware.py | 193 +++++++++--------- .../tests/unit/audit/test_logging_notifier.py | 45 ++++ 2 files changed, 137 insertions(+), 101 deletions(-) create mode 100644 keystonemiddleware/tests/unit/audit/test_logging_notifier.py diff --git a/keystonemiddleware/tests/unit/audit/test_audit_middleware.py b/keystonemiddleware/tests/unit/audit/test_audit_middleware.py index 3c42404f..c3c307da 100644 --- a/keystonemiddleware/tests/unit/audit/test_audit_middleware.py +++ b/keystonemiddleware/tests/unit/audit/test_audit_middleware.py @@ -13,82 +13,91 @@ import uuid +import fixtures import mock -from testtools import matchers import webob from keystonemiddleware import audit from keystonemiddleware.tests.unit.audit import base -@mock.patch('oslo_messaging.get_transport', mock.MagicMock()) class AuditMiddlewareTest(base.BaseAuditMiddlewareTest): + def setUp(self): + self.notifier = mock.MagicMock() + + p = 'keystonemiddleware.audit._notifier.create_notifier' + f = fixtures.MockPatch(p, return_value=self.notifier) + self.notifier_fixture = self.useFixture(f) + + super(AuditMiddlewareTest, self).setUp() + def test_api_request(self): req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo_messaging.Notifier.info') as notify: - self.middleware(req) - # Check first notification with only 'request' - call_args = notify.call_args_list[0][0] - self.assertEqual('audit.http.request', call_args[1]) - self.assertEqual('/foo/bar', call_args[2]['requestPath']) - self.assertEqual('pending', call_args[2]['outcome']) - self.assertNotIn('reason', call_args[2]) - self.assertNotIn('reporterchain', call_args[2]) - # Check second notification with request + response - call_args = notify.call_args_list[1][0] - self.assertEqual('audit.http.response', call_args[1]) - self.assertEqual('/foo/bar', call_args[2]['requestPath']) - self.assertEqual('success', call_args[2]['outcome']) - self.assertIn('reason', call_args[2]) - self.assertIn('reporterchain', call_args[2]) + self.middleware(req) + + # Check first notification with only 'request' + call_args = self.notifier.notify.call_args_list[0][0] + self.assertEqual('audit.http.request', call_args[1]) + self.assertEqual('/foo/bar', call_args[2]['requestPath']) + self.assertEqual('pending', call_args[2]['outcome']) + self.assertNotIn('reason', call_args[2]) + self.assertNotIn('reporterchain', call_args[2]) + + # Check second notification with request + response + call_args = self.notifier.notify.call_args_list[1][0] + self.assertEqual('audit.http.response', call_args[1]) + self.assertEqual('/foo/bar', call_args[2]['requestPath']) + self.assertEqual('success', call_args[2]['outcome']) + self.assertIn('reason', call_args[2]) + self.assertIn('reporterchain', call_args[2]) def test_api_request_failure(self): self.middleware = audit.AuditMiddleware( base.FakeFailingApp(), audit_map_file=self.audit_map, service_name='pycadf') + req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo_messaging.Notifier.info') as notify: - try: - self.middleware(req) - self.fail('Application exception has not been re-raised') - except Exception: - pass - # Check first notification with only 'request' - call_args = notify.call_args_list[0][0] - self.assertEqual('audit.http.request', call_args[1]) - self.assertEqual('/foo/bar', call_args[2]['requestPath']) - self.assertEqual('pending', call_args[2]['outcome']) - self.assertNotIn('reporterchain', call_args[2]) - # Check second notification with request + response - call_args = notify.call_args_list[1][0] - self.assertEqual('audit.http.response', call_args[1]) - self.assertEqual('/foo/bar', call_args[2]['requestPath']) - self.assertEqual('unknown', call_args[2]['outcome']) - self.assertIn('reporterchain', call_args[2]) + try: + self.middleware(req) + self.fail('Application exception has not been re-raised') + except Exception: + pass + + # Check first notification with only 'request' + call_args = self.notifier.notify.call_args_list[0][0] + self.assertEqual('audit.http.request', call_args[1]) + self.assertEqual('/foo/bar', call_args[2]['requestPath']) + self.assertEqual('pending', call_args[2]['outcome']) + self.assertNotIn('reporterchain', call_args[2]) + + # Check second notification with request + response + call_args = self.notifier.notify.call_args_list[1][0] + self.assertEqual('audit.http.response', call_args[1]) + self.assertEqual('/foo/bar', call_args[2]['requestPath']) + self.assertEqual('unknown', call_args[2]['outcome']) + self.assertIn('reporterchain', call_args[2]) def test_process_request_fail(self): req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) req.context = {} - with mock.patch('oslo_messaging.Notifier.info', - side_effect=Exception('error')) as notify: - self.middleware._process_request(req) - self.assertTrue(notify.called) + + self.middleware._process_request(req) + self.assertTrue(self.notifier.notify.called) def test_process_response_fail(self): req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) req.context = {} - with mock.patch('oslo_messaging.Notifier.info', - side_effect=Exception('error')) as notify: - self.middleware._process_response(req, webob.response.Response()) - self.assertTrue(notify.called) + + self.middleware._process_response(req, webob.response.Response()) + self.assertTrue(self.notifier.notify.called) def test_ignore_req_opt(self): self.middleware = audit.AuditMiddleware(base.FakeApp(), @@ -100,43 +109,23 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest): environ=self.get_environ_header('PUT')) req2 = webob.Request.blank('/accept/foo', environ=self.get_environ_header('POST')) - with mock.patch('oslo_messaging.Notifier.info') as notify: - # Check GET/PUT request does not send notification - self.middleware(req) - self.middleware(req1) - self.assertEqual([], notify.call_args_list) - # Check non-GET/PUT request does send notification - self.middleware(req2) - self.assertThat(notify.call_args_list, matchers.HasLength(2)) - call_args = notify.call_args_list[0][0] - self.assertEqual('audit.http.request', call_args[1]) - self.assertEqual('/accept/foo', call_args[2]['requestPath']) + # Check GET/PUT request does not send notification + self.middleware(req) + self.middleware(req1) + self.assertFalse(self.notifier.notify.called) - call_args = notify.call_args_list[1][0] - self.assertEqual('audit.http.response', call_args[1]) - self.assertEqual('/accept/foo', call_args[2]['requestPath']) + # Check non-GET/PUT request does send notification + self.middleware(req2) + self.assertEqual(2, self.notifier.notify.call_count) - def test_api_request_no_messaging(self): - req = webob.Request.blank('/foo/bar', - environ=self.get_environ_header('GET')) - with mock.patch('keystonemiddleware.audit._notifier.oslo_messaging', - None): - with mock.patch('keystonemiddleware.audit._LOG.info') as log: - self.middleware = audit.AuditMiddleware( - base.FakeApp(), - audit_map_file=self.audit_map) + call_args = self.notifier.notify.call_args_list[0][0] + self.assertEqual('audit.http.request', call_args[1]) + self.assertEqual('/accept/foo', call_args[2]['requestPath']) - self.middleware(req) - # Check first notification with only 'request' - call_args = log.call_args_list[0][0] - self.assertEqual('audit.http.request', - call_args[1]['event_type']) - - # Check second notification with request + response - call_args = log.call_args_list[1][0] - self.assertEqual('audit.http.response', - call_args[1]['event_type']) + call_args = self.notifier.notify.call_args_list[1][0] + self.assertEqual('audit.http.response', call_args[1]) + self.assertEqual('/accept/foo', call_args[2]['requestPath']) def test_cadf_event_context_scoped(self): middleware = audit.AuditMiddleware( @@ -145,17 +134,17 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest): service_name='pycadf') req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo_messaging.Notifier.info') as notify: - middleware(req) - self.assertEqual(2, notify.call_count) - first, second = [a[0] for a in notify.call_args_list] + middleware(req) - # the Context is the first argument. Let's verify it. - self.assertIsInstance(first[0], dict) + self.assertEqual(2, self.notifier.notify.call_count) + first, second = [a[0] for a in self.notifier.notify.call_args_list] - # ensure exact same context is used between request and response - self.assertIs(first[0], second[0]) + # the Context is the first argument. Let's verify it. + self.assertIsInstance(first[0], dict) + + # ensure exact same context is used between request and response + self.assertIs(first[0], second[0]) def test_cadf_event_scoped_to_request(self): middleware = audit.AuditMiddleware( @@ -164,35 +153,37 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest): service_name='pycadf') req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo_messaging.Notifier.info') as notify: - middleware(req) - self.assertIsNotNone(req.environ.get('cadf_event')) + middleware(req) + self.assertIsNotNone(req.environ.get('cadf_event')) - # ensure exact same event is used between request and response - self.assertEqual(notify.call_args_list[0][0][2]['id'], - notify.call_args_list[1][0][2]['id']) + # ensure exact same event is used between request and response + self.assertEqual(self.notifier.calls[0].payload['id'], + self.notifier.calls[1].payload['id']) def test_cadf_event_scoped_to_request_on_error(self): middleware = audit.AuditMiddleware( base.FakeApp(), audit_map_file=self.audit_map, service_name='pycadf') + req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) req.context = {} - with mock.patch('oslo_messaging.Notifier.info', - side_effect=Exception('error')) as notify: - middleware._process_request(req) - self.assertTrue(notify.called) + self.notifier.notify.side_effect = Exception('error') + + middleware._process_request(req) + self.assertTrue(self.notifier.notify.called) + req2 = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) req2.context = {} - with mock.patch('oslo_messaging.Notifier.info') as notify: - middleware._process_response(req2, webob.response.Response()) - self.assertTrue(notify.called) - # ensure event is not the same across requests - self.assertNotEqual(req.environ['cadf_event'].id, - notify.call_args_list[0][0][2]['id']) + self.notifier.reset_mock() + + middleware._process_response(req2, webob.response.Response()) + self.assertTrue(self.notifier.notify.called) + # ensure event is not the same across requests + self.assertNotEqual(req.environ['cadf_event'].id, + self.notifier.notify.call_args_list[0][0][2]['id']) def test_project_name_from_oslo_config(self): self.assertEqual(self.PROJECT_NAME, diff --git a/keystonemiddleware/tests/unit/audit/test_logging_notifier.py b/keystonemiddleware/tests/unit/audit/test_logging_notifier.py new file mode 100644 index 00000000..b3c20e87 --- /dev/null +++ b/keystonemiddleware/tests/unit/audit/test_logging_notifier.py @@ -0,0 +1,45 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# 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 fixtures +import mock +import webob + +from keystonemiddleware import audit +from keystonemiddleware.tests.unit.audit import base + + +class TestLoggingNotifier(base.BaseAuditMiddlewareTest): + + def setUp(self): + p = 'keystonemiddleware.audit._notifier.oslo_messaging' + f = fixtures.MockPatch(p, None) + self.messaging_fixture = self.useFixture(f) + + super(TestLoggingNotifier, self).setUp() + + @mock.patch('keystonemiddleware.audit._LOG.info') + def test_api_request_no_messaging(self, log): + middleware = audit.AuditMiddleware(base.FakeApp(), + audit_map_file=self.audit_map) + + req = webob.Request.blank('/foo/bar', + environ=self.get_environ_header('GET')) + req.get_response(middleware) + + # Check first notification with only 'request' + call_args = log.call_args_list[0][0] + self.assertEqual('audit.http.request', call_args[1]['event_type']) + + # Check second notification with request + response + call_args = log.call_args_list[1][0] + self.assertEqual('audit.http.response', call_args[1]['event_type'])