# Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You 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 os import sys from io import StringIO import zlib import requests_mock import mock import libcloud from libcloud.test import unittest from libcloud.common.base import Connection from libcloud.http import LibcloudConnection from libcloud.utils.loggingconnection import LoggingConnection EXPECTED_DATA_JSON = """ HTTP/1.1 200 OK Content-Type: application/json {"foo": "bar!"} """.strip() EXPECTED_DATA_JSON_PRETTY = """ HTTP/1.1 200 OK Content-Type: application/json { "foo": "bar!" } """.strip() EXPECTED_DATA_XML = """ HTTP/1.1 200 OK Content-Type: text/xml """.strip() EXPECTED_DATA_XML_PRETTY = """ HTTP/1.1 200 OK Content-Type: application/xml """.strip() class TestLoggingConnection(unittest.TestCase): def setUp(self): super(TestLoggingConnection, self).setUp() self._reset_environ() def tearDown(self): super(TestLoggingConnection, self).tearDown() Connection.conn_class = LibcloudConnection def test_debug_method_uses_log_class(self): with StringIO() as fh: libcloud.enable_debug(fh) conn = Connection(timeout=10) conn.connect() self.assertTrue(isinstance(conn.connection, LoggingConnection)) def test_debug_log_class_handles_request(self): with StringIO() as fh: libcloud.enable_debug(fh) conn = Connection(url='http://test.com/') conn.connect() self.assertEqual(conn.connection.host, 'http://test.com') with requests_mock.mock() as m: m.get('http://test.com/test', text='data') conn.request('/test') log = fh.getvalue() self.assertTrue(isinstance(conn.connection, LoggingConnection)) self.assertIn('-i -X GET', log) self.assertIn('data', log) def test_debug_log_class_handles_request_with_compression(self): request = zlib.compress(b'data') with StringIO() as fh: libcloud.enable_debug(fh) conn = Connection(url='http://test.com/') conn.connect() self.assertEqual(conn.connection.host, 'http://test.com') with requests_mock.mock() as m: m.get('http://test.com/test', content=request, headers={'content-encoding': 'zlib'}) conn.request('/test') log = fh.getvalue() self.assertTrue(isinstance(conn.connection, LoggingConnection)) self.assertIn('-i -X GET', log) def test_log_response_json_content_type(self): conn = LoggingConnection(host='example.com', port=80) r = self._get_mock_response('application/json', '{"foo": "bar!"}') result = conn._log_response(r).replace('\r', '') self.assertTrue(EXPECTED_DATA_JSON in result) def test_log_response_xml_content_type(self): conn = LoggingConnection(host='example.com', port=80) r = self._get_mock_response('text/xml', '') result = conn._log_response(r).replace('\r', '') self.assertTrue(EXPECTED_DATA_XML in result) def test_log_response_with_pretty_print_json_content_type(self): os.environ['LIBCLOUD_DEBUG_PRETTY_PRINT_RESPONSE'] = '1' conn = LoggingConnection(host='example.com', port=80) # body type is unicode r = self._get_mock_response('application/json', '{"foo": "bar!"}') result = conn._log_response(r).replace('\r', '') self.assertTrue(EXPECTED_DATA_JSON_PRETTY in result) # body type is bytes r = self._get_mock_response('application/json', bytes('{"foo": "bar!"}', 'utf-8')) result = conn._log_response(r).replace('\r', '') self.assertTrue(EXPECTED_DATA_JSON_PRETTY in result) def test_log_response_with_pretty_print_xml_content_type(self): os.environ['LIBCLOUD_DEBUG_PRETTY_PRINT_RESPONSE'] = '1' conn = LoggingConnection(host='example.com', port=80) r = self._get_mock_response('application/xml', '') result = conn._log_response(r).replace('\r', '') self.assertTrue(EXPECTED_DATA_XML_PRETTY in result) def _reset_environ(self): if 'LIBCLOUD_DEBUG_PRETTY_PRINT_RESPONSE' in os.environ: del os.environ['LIBCLOUD_DEBUG_PRETTY_PRINT_RESPONSE'] def _get_mock_response(self, content_type, body): header = mock.Mock() header.title.return_value = 'Content-Type' header.lower.return_value = 'content-type' r = mock.Mock() r.version = 11 r.status = '200' r.reason = 'OK' r.getheaders.return_value = [(header, content_type)] r.read.return_value = body return r if __name__ == '__main__': sys.exit(unittest.main())