"Fossies" - the Fresh Open Source Software Archive

Member "monasca-api-3.1.0/monasca_api/tests/test_logs.py" (27 Sep 2019, 5885 Bytes) of package /linux/misc/openstack/monasca-api-3.1.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_logs.py": 3.1.0_vs_4.0.0.

    1 # Copyright 2016-2017 FUJITSU LIMITED
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 # not use this file except in compliance with the License. You may obtain
    5 # a copy of the License at
    6 #
    7 #      http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 # License for the specific language governing permissions and limitations
   13 # under the License.
   14 
   15 import falcon
   16 import mock
   17 import ujson as json
   18 
   19 from monasca_api.tests import base
   20 from monasca_api.v2.reference import logs
   21 
   22 ENDPOINT = '/logs'
   23 TENANT_ID = 'bob'
   24 ROLES = 'admin'
   25 
   26 
   27 def _init_resource(test):
   28     resource = logs.Logs()
   29     test.app.add_route(ENDPOINT, resource)
   30     return resource
   31 
   32 
   33 def _generate_payload(log_count=None, messages=None):
   34     if not log_count and messages:
   35         log_count = len(messages)
   36         logs = [{
   37             'message': messages[it],
   38             'dimensions': {
   39                 'hostname': 'host_%d' % it,
   40                 'component': 'component_%d' % it,
   41                 'service': 'service_%d' % it
   42             }
   43         } for it in range(log_count)]
   44     else:
   45         logs = [{
   46             'message': base.generate_unique_message(100),
   47             'dimensions': {
   48                 'hostname': 'host_%d' % it,
   49                 'component': 'component_%d' % it,
   50                 'service': 'service_%d' % it
   51             }
   52         } for it in range(log_count)]
   53     body = {
   54         'dimensions': {
   55             'origin': __name__
   56         },
   57         'logs': logs
   58     }
   59 
   60     return body, logs
   61 
   62 
   63 class TestApiLogsVersion(base.BaseApiTestCase):
   64 
   65     @mock.patch('monasca_api.v2.common.bulk_processor.BulkProcessor')
   66     def test_should_return_as_version(self, _):
   67         logs_resource = logs.Logs()
   68         self.assertEqual('v2.0', logs_resource.version)
   69 
   70 
   71 class TestApiLogs(base.BaseApiTestCase):
   72 
   73     @mock.patch('monasca_api.v2.common.bulk_processor.BulkProcessor')
   74     def test_should_pass_cross_tenant_id(self, bulk_processor):
   75         logs_resource = _init_resource(self)
   76         logs_resource._processor = bulk_processor
   77 
   78         body, logs = _generate_payload(1)
   79         payload = json.dumps(body)
   80         content_length = len(payload)
   81         response = self.simulate_request(
   82             path='/logs',
   83             method='POST',
   84             query_string='tenant_id=1',
   85             headers={
   86                 'X_ROLES': ROLES,
   87                 'Content-Type': 'application/json',
   88                 'Content-Length': str(content_length)
   89             },
   90             body=payload
   91         )
   92         self.assertEqual(falcon.HTTP_204, response.status)
   93         logs_resource._processor.send_message.assert_called_with(
   94             logs=logs,
   95             global_dimensions=body['dimensions'],
   96             log_tenant_id='1')
   97 
   98     @mock.patch('monasca_api.v2.common.bulk_processor.BulkProcessor')
   99     def test_should_fail_not_delegate_ok_cross_tenant_id(self, _):
  100         _init_resource(self)
  101         response = self.simulate_request(
  102             path='/logs',
  103             method='POST',
  104             query_string='tenant_id=1',
  105             headers={
  106                 'X-Roles': ROLES,
  107                 'Content-Type': 'application/json',
  108                 'Content-Length': '0'
  109             }
  110         )
  111         self.assertEqual(falcon.HTTP_400, response.status)
  112 
  113     @mock.patch('monasca_api.v2.common.bulk_processor.BulkProcessor')
  114     def test_should_pass_empty_cross_tenant_id_wrong_role(self,
  115                                                           bulk_processor):
  116         logs_resource = _init_resource(self)
  117         logs_resource._processor = bulk_processor
  118 
  119         body, _ = _generate_payload(1)
  120         payload = json.dumps(body)
  121         content_length = len(payload)
  122         response = self.simulate_request(
  123             path='/logs',
  124             method='POST',
  125             headers={
  126                 'X-Roles': ROLES,
  127                 'Content-Type': 'application/json',
  128                 'Content-Length': str(content_length)
  129             },
  130             body=payload
  131         )
  132         self.assertEqual(falcon.HTTP_204, response.status)
  133         self.assertEqual(1, bulk_processor.send_message.call_count)
  134 
  135     @mock.patch('monasca_api.v2.common.bulk_processor.BulkProcessor')
  136     def test_should_pass_empty_cross_tenant_id_ok_role(self,
  137                                                        bulk_processor):
  138         logs_resource = _init_resource(self)
  139         logs_resource._processor = bulk_processor
  140 
  141         body, _ = _generate_payload(1)
  142         payload = json.dumps(body)
  143         content_length = len(payload)
  144         response = self.simulate_request(
  145             path='/logs',
  146             method='POST',
  147             headers={
  148                 'X-Roles': ROLES,
  149                 'Content-Type': 'application/json',
  150                 'Content-Length': str(content_length)
  151             },
  152             body=payload
  153         )
  154         self.assertEqual(falcon.HTTP_204, response.status)
  155         self.assertEqual(1, bulk_processor.send_message.call_count)
  156 
  157 
  158 class TestUnicodeLogs(base.BaseApiTestCase):
  159 
  160     @mock.patch('monasca_api.api.core.log.log_publisher.producer.'
  161                 'KafkaProducer')
  162     def test_should_send_unicode_messages(self, _):
  163         _init_resource(self)
  164 
  165         messages = [m['input'] for m in base.UNICODE_MESSAGES]
  166         body, _ = _generate_payload(messages=messages)
  167         payload = json.dumps(body, ensure_ascii=False)
  168         response = self.simulate_request(
  169             path='/logs',
  170             method='POST',
  171             headers={
  172                 'X-Roles': ROLES,
  173                 'Content-Type': 'application/json'
  174             },
  175             body=payload
  176         )
  177         self.assertEqual(falcon.HTTP_204, response.status)