diff options
Diffstat (limited to 'keystonemiddleware-moon/keystonemiddleware/tests/unit/test_audit_middleware.py')
| -rw-r--r-- | keystonemiddleware-moon/keystonemiddleware/tests/unit/test_audit_middleware.py | 93 |
1 files changed, 81 insertions, 12 deletions
diff --git a/keystonemiddleware-moon/keystonemiddleware/tests/unit/test_audit_middleware.py b/keystonemiddleware-moon/keystonemiddleware/tests/unit/test_audit_middleware.py index 89e5aa44..48ff9a4f 100644 --- a/keystonemiddleware-moon/keystonemiddleware/tests/unit/test_audit_middleware.py +++ b/keystonemiddleware-moon/keystonemiddleware/tests/unit/test_audit_middleware.py @@ -18,11 +18,11 @@ import uuid import mock from oslo_config import cfg from pycadf import identifier -import testtools from testtools import matchers import webob from keystonemiddleware import audit +from keystonemiddleware.tests.unit import utils class FakeApp(object): @@ -40,7 +40,7 @@ class FakeFailingApp(object): raise Exception('It happens!') -class BaseAuditMiddlewareTest(testtools.TestCase): +class BaseAuditMiddlewareTest(utils.BaseTestCase): def setUp(self): super(BaseAuditMiddlewareTest, self).setUp() self.fd, self.audit_map = tempfile.mkstemp() @@ -90,13 +90,13 @@ class BaseAuditMiddlewareTest(testtools.TestCase): return env_headers -@mock.patch('oslo.messaging.get_transport', mock.MagicMock()) +@mock.patch('oslo_messaging.get_transport', mock.MagicMock()) class AuditMiddlewareTest(BaseAuditMiddlewareTest): 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: + 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] @@ -121,7 +121,7 @@ class AuditMiddlewareTest(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: + with mock.patch('oslo_messaging.Notifier.info') as notify: try: self.middleware(req) self.fail('Application exception has not been re-raised') @@ -144,7 +144,7 @@ class AuditMiddlewareTest(BaseAuditMiddlewareTest): def test_process_request_fail(self): req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo.messaging.Notifier.info', + with mock.patch('oslo_messaging.Notifier.info', side_effect=Exception('error')) as notify: self.middleware._process_request(req) self.assertTrue(notify.called) @@ -152,7 +152,7 @@ class AuditMiddlewareTest(BaseAuditMiddlewareTest): def test_process_response_fail(self): req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo.messaging.Notifier.info', + 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) @@ -167,7 +167,7 @@ class AuditMiddlewareTest(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: + with mock.patch('oslo_messaging.Notifier.info') as notify: # Check GET/PUT request does not send notification self.middleware(req) self.middleware(req1) @@ -207,7 +207,7 @@ class AuditMiddlewareTest(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: + with mock.patch('oslo_messaging.Notifier.info') as notify: middleware(req) self.assertIsNotNone(req.environ.get('cadf_event')) @@ -222,13 +222,13 @@ class AuditMiddlewareTest(BaseAuditMiddlewareTest): service_name='pycadf') req = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo.messaging.Notifier.info', + with mock.patch('oslo_messaging.Notifier.info', side_effect=Exception('error')) as notify: middleware._process_request(req) self.assertTrue(notify.called) req2 = webob.Request.blank('/foo/bar', environ=self.get_environ_header('GET')) - with mock.patch('oslo.messaging.Notifier.info') as notify: + 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 @@ -236,7 +236,7 @@ class AuditMiddlewareTest(BaseAuditMiddlewareTest): notify.call_args_list[0][0][2]['id']) -@mock.patch('oslo.messaging', mock.MagicMock()) +@mock.patch('oslo_messaging.rpc', mock.MagicMock()) class AuditApiLogicTest(BaseAuditMiddlewareTest): def api_request(self, method, url): @@ -483,3 +483,72 @@ class AuditApiLogicTest(BaseAuditMiddlewareTest): self.middleware._process_request(req) payload = req.environ['cadf_event'].as_dict() self.assertEqual(payload['target']['id'], identifier.norm_ns('nova')) + + def test_endpoint_missing_internal_url(self): + env_headers = {'HTTP_X_SERVICE_CATALOG': + '''[{"endpoints_links": [], + "endpoints": [{"adminURL": + "http://admin_host:8774", + "region": "RegionOne", + "publicURL": + "http://public_host:8774"}], + "type": "compute", + "name": "nova"},]''', + 'HTTP_X_USER_ID': 'user_id', + 'HTTP_X_USER_NAME': 'user_name', + 'HTTP_X_AUTH_TOKEN': 'token', + 'HTTP_X_PROJECT_ID': 'tenant_id', + 'HTTP_X_IDENTITY_STATUS': 'Confirmed', + 'REQUEST_METHOD': 'GET'} + req = webob.Request.blank('http://admin_host:8774/v2/' + + str(uuid.uuid4()) + '/servers', + environ=env_headers) + self.middleware._process_request(req) + payload = req.environ['cadf_event'].as_dict() + self.assertEqual((payload['target']['addresses'][1]['url']), "unknown") + + def test_endpoint_missing_public_url(self): + env_headers = {'HTTP_X_SERVICE_CATALOG': + '''[{"endpoints_links": [], + "endpoints": [{"adminURL": + "http://admin_host:8774", + "region": "RegionOne", + "internalURL": + "http://internal_host:8774"}], + "type": "compute", + "name": "nova"},]''', + 'HTTP_X_USER_ID': 'user_id', + 'HTTP_X_USER_NAME': 'user_name', + 'HTTP_X_AUTH_TOKEN': 'token', + 'HTTP_X_PROJECT_ID': 'tenant_id', + 'HTTP_X_IDENTITY_STATUS': 'Confirmed', + 'REQUEST_METHOD': 'GET'} + req = webob.Request.blank('http://admin_host:8774/v2/' + + str(uuid.uuid4()) + '/servers', + environ=env_headers) + self.middleware._process_request(req) + payload = req.environ['cadf_event'].as_dict() + self.assertEqual((payload['target']['addresses'][2]['url']), "unknown") + + def test_endpoint_missing_admin_url(self): + env_headers = {'HTTP_X_SERVICE_CATALOG': + '''[{"endpoints_links": [], + "endpoints": [{"region": "RegionOne", + "publicURL": + "http://public_host:8774", + "internalURL": + "http://internal_host:8774"}], + "type": "compute", + "name": "nova"},]''', + 'HTTP_X_USER_ID': 'user_id', + 'HTTP_X_USER_NAME': 'user_name', + 'HTTP_X_AUTH_TOKEN': 'token', + 'HTTP_X_PROJECT_ID': 'tenant_id', + 'HTTP_X_IDENTITY_STATUS': 'Confirmed', + 'REQUEST_METHOD': 'GET'} + req = webob.Request.blank('http://public_host:8774/v2/' + + str(uuid.uuid4()) + '/servers', + environ=env_headers) + self.middleware._process_request(req) + payload = req.environ['cadf_event'].as_dict() + self.assertEqual((payload['target']['addresses'][0]['url']), "unknown") |