diff options
Diffstat (limited to 'test/logging/test_osdf_logging.py')
| -rwxr-xr-x | test/logging/test_osdf_logging.py | 182 |
1 files changed, 182 insertions, 0 deletions
diff --git a/test/logging/test_osdf_logging.py b/test/logging/test_osdf_logging.py new file mode 100755 index 0000000..982ef0b --- /dev/null +++ b/test/logging/test_osdf_logging.py @@ -0,0 +1,182 @@ +# ------------------------------------------------------------------------- +# Copyright (c) 2017-2018 AT&T Intellectual Property +# +# 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 unittest +import json +import yaml + +from osdf.logging import osdf_logging as L1 +from osdf.logging.osdf_logging import OOF_OSDFLogMessageHelper as MH +from osdf.logging.osdf_logging import OOF_OSDFLogMessageFormatter as formatter +from unittest import mock + + +class TestOSDFLogging(unittest.TestCase): + + def __init__(self, *args, **kwargs): + super(self.__class__, self).__init__(*args, **kwargs) + self.MH = MH() + self.err = Exception("Some Exception") + self.req_id = "TEST-Req-ID" + self.url = mock.MagicMock() + self.request = mock.MagicMock() + self.client = mock.MagicMock() + self.service_name = mock.MagicMock() + self.callback_url = mock.MagicMock() + self.service_name = mock.MagicMock() + self.body = mock.MagicMock() + self.desc = mock.MagicMock() + self.response = mock.MagicMock() + self.remote_addr = mock.MagicMock() + self.json_body = mock.MagicMock() + self.F = formatter + + def test_log_handlers_pre_onap(self): + res = L1.log_handlers_pre_onap() + assert type(res) == dict + + def test_format_exception(self): + res = L1.format_exception(Exception("Some error")) + + def test_accepted_valid_request(self): + res = formatter.accepted_valid_request(self.req_id, self.request) + assert res.startswith("Accepted valid request") + + def test_invalid_request(self): + res = formatter.invalid_request(self.req_id, self.err) + assert res.startswith("Invalid request") + + def test_invalid_response(self): + res = formatter.invalid_response(self.req_id, self.err) + assert res.startswith("Invalid response") + + def test_malformed_request(self): + res = formatter.malformed_request(self.request, self.err) + assert res.startswith("Malformed request") + + def test_malformed_response(self): + res = formatter.malformed_response(self.response, self.client, self.err) + assert res.startswith("Malformed response") + + def test_need_policies(self): + res = formatter.need_policies(self.req_id) + assert res.startswith("Policies required but") + + def test_policy_service_error(self): + res = formatter.policy_service_error(self.url, self.req_id, self.err) + assert res.startswith("Unable to call policy") + + def test_requesting_url(self): + res = formatter.requesting_url(self.url, self.req_id) + assert res.startswith("Making a call to URL") + + def test_requesting(self): + res = formatter.requesting(self.service_name, self.req_id) + assert res.startswith("Making a call to service") + + def test_error_requesting(self): + res = formatter.error_requesting(self.service_name, self.req_id, self.err) + assert res.startswith("Error while requesting service") + + def test_error_calling_back(self): + res = formatter.error_calling_back(self.service_name, self.callback_url, self.err) + assert res.startswith("Error while posting result to callback URL") + + def test_calling_back(self): + res = formatter.calling_back(self.req_id, self.callback_url) + assert res.startswith("Posting result to callback URL") + + def test_calling_back_with_body(self): + res = formatter.calling_back_with_body(self.req_id, self.callback_url, self.body) + assert res.startswith("Posting result to callback URL") + + def test_received_request(self): + res = formatter.received_request(self.url, self.remote_addr, self.json_body) + assert res.startswith("Received a call") + + def test_new_worker_thread(self): + res = formatter.new_worker_thread(self.req_id) + assert res.startswith("Initiating new worker thread") + + def test_inside_worker_thread(self): + res = formatter.inside_worker_thread(self.req_id) + assert res.startswith("Inside worker thread for request ID") + + def test_inside_new_thread(self): + res = formatter.inside_new_thread(self.req_id) + assert res.startswith("Spinning up a new thread for request ID") + + def test_processing(self): + res = formatter.processing(self.req_id, self.desc) + assert res.startswith("Processing request ID") + + def test_processed(self): + res = formatter.processed(self.req_id, self.desc) + assert res.startswith("Processed request ID") + + def test_error_while_processing(self): + res = formatter.error_while_processing(self.req_id, self.desc, self.err) + assert res.startswith("Error while processing request ID") + + def test_creating_local_env(self): + res = formatter.creating_local_env(self.req_id) + assert res.startswith("Creating local environment request ID") + + def test_error_local_env(self): + res = formatter.error_local_env(self.req_id, self.desc, self.err) + assert res.startswith("Error while creating local environment for request ID") + + def test_error_response_posting(self): + res = formatter.error_response_posting(self.req_id, self.desc, self.err) + assert res.startswith("Error while posting a response for a request ID") + + def test_received_http_response(self): + res = formatter.received_http_response(self.response) + assert res.startswith("Received response [code: ") + + def test_sending_response(self): + res = formatter.sending_response(self.req_id, self.desc) + assert res.startswith("Response is sent for request ID") + + def test_listening_response(self): + res = formatter.listening_response(self.req_id, self.desc) + assert res.startswith("Response is sent for request ID") + + def test_items_received(self): + res = formatter.items_received(10, "messages") + assert res == "Received 10 messages" + + def test_items_sent(self): + res = formatter.items_sent(10, "messages") + assert res == "Published 10 messages" + + def test_warn_audit_error(self): + """Log the message to error_log.warn and audit_log.warn""" + L1.warn_audit_error("Some warning message") + + def test_log_message_multi(msg): + X = L1.log_handlers_pre_onap() + wanted_methods = [ + X["error"].error, X["error"].warn, X["audit"].info, + X["metrics"].info, X["debug"].debug, X["error"].fatal + ] + L1.log_message_multi("Some log message", *wanted_methods) + + +if __name__ == "__main__": + unittest.main() + |