diff --git a/coriolisclient/tests/v1/data/logs.yml b/coriolisclient/tests/v1/data/logs.yml new file mode 100644 index 0000000..59c1b48 --- /dev/null +++ b/coriolisclient/tests/v1/data/logs.yml @@ -0,0 +1 @@ +test_chunk1test_chunk2 \ No newline at end of file diff --git a/coriolisclient/tests/v1/test_logging.py b/coriolisclient/tests/v1/test_logging.py new file mode 100644 index 0000000..5fc6c61 --- /dev/null +++ b/coriolisclient/tests/v1/test_logging.py @@ -0,0 +1,278 @@ +# Copyright 2024 Cloudbase Solutions Srl +# All Rights Reserved. + +import ddt +import os +import requests +from unittest import mock + +from keystoneauth1.exceptions import http + +from coriolisclient import exceptions +from coriolisclient.tests import test_base +from coriolisclient.v1 import logging + + +@ddt.ddt +class LoggingClientTestCase( + test_base.CoriolisBaseTestCase): + """Test suite for the Coriolis v1 Logging Client.""" + + @mock.patch.object(logging.LoggingClient, '_get_endpoint_url') + def setUp(self, mock_get_endpoint_url): + mock_get_endpoint_url.return_value = mock.sentinel.ep_url + mock_client = mock.Mock() + super(LoggingClientTestCase, self).setUp() + self.logger = logging.LoggingClient(mock_client) + + @mock.patch.object(logging.LoggingClient, '_get_endpoint_url') + def test__init__(self, mock_get_endpoint_url): + mock_get_endpoint_url.side_effect = Exception + + with self.assertLogs(logger=logging.LOG, level="WARNING"): + logger = logging.LoggingClient(None) + + self.assertEqual( + None, + logger._ep_url + ) + + def test_get_endpoint_url(self): + self.logger._cli.get_endpoint.return_value = "url/endpoint_url/" + + result = self.logger._get_endpoint_url(mock.sentinel.name) + + self.assertEqual( + "url/endpoint_url", + result + ) + self.logger._cli.get_endpoint.assert_called_once_with( + service_type=mock.sentinel.name) + + def test_get_endpoint_url_not_found(self): + self.logger._cli.get_endpoint.return_value = None + + self.assertRaises( + exceptions.LoggingEndpointNotFound, + self.logger._get_endpoint_url, + mock.sentinel.name + ) + self.logger._cli.get_endpoint.assert_called_once_with( + service_type=mock.sentinel.name) + + def test_get_endpoint_url_http_unauthorized(self): + self.logger._cli.get_endpoint.side_effect = http.Unauthorized + + with self.assertLogs(logger=logging.LOG, level="ERROR"): + self.assertRaises( + exceptions.HTTPAuthError, + self.logger._get_endpoint_url, + mock.sentinel.name + ) + self.logger._cli.get_endpoint.assert_called_once_with( + service_type=mock.sentinel.name) + + @ddt.data( + { + "query_args": { + "arg1": None, + "arg2": None + }, + "is_websocket": True, + "expected_result": + "ws:///None/sentinel.resource" + }, + { + "query_args": { + "arg1": None, + "arg2": "mock_arg2" + }, + "_ep_url": "https:///ep_url", + "is_websocket": True, + "expected_result": + "wss:///ep_url/sentinel.resource?arg2=mock_arg2" + }, + { + "query_args": { + "arg1": "mock_arg1", + "arg2": "mock_arg2" + }, + "_ep_url": "https:///ep_url", + "is_websocket": False, + "expected_result": "https:///ep_url/sentinel.resource" + "?arg1=mock_arg1&arg2=mock_arg2" + } + ) + @mock.patch.object(logging.LoggingClient, '_get_endpoint_url') + def test_construct_url(self, data, mock_get_endpoint_url): + self.logger._ep_url = None + mock_get_endpoint_url.return_value = data.get("_ep_url", None) + + result = self.logger._construct_url( + mock.sentinel.resource, + data.get("query_args"), + is_websocket=data.get("is_websocket", False), + ) + + self.assertEqual( + data.get("expected_result"), + result + ) + + @ddt.data( + (None, None, False, False), + ("1", 1, False, False), + ("1234567890123456789", None, True, True), + ("abc", None, True, True), + ("", None, True, True), + ("10s", mock.ANY, False, True), + ) + @ddt.unpack + def test_convert_period_to_timestamp( + self, + period, + expected_result, + raises, + has_logs + ): + if raises is False: + if has_logs is False: + result = self.logger._convert_period_to_timestamp(period) + else: + with self.assertLogs(logger=logging.LOG, level="WARNING"): + result = self.logger._convert_period_to_timestamp(period) + self.assertEqual( + expected_result, + result + ) + else: + with self.assertLogs(logger=logging.LOG, level="WARNING"): + self.assertRaises( + exceptions.CoriolisException, + self.logger._convert_period_to_timestamp, + period + ) + + @mock.patch.object(requests, "get") + @mock.patch.object(logging.LoggingClient, "_construct_url") + @mock.patch.object(logging.LoggingClient, "_convert_period_to_timestamp") + def test_download_logs( + self, + mock_convert_period_to_timestamp, + mock_construct_url, + mock_get + ): + logs_path = os.path.dirname(os.path.realpath(__file__)) + logs_path = os.path.join(logs_path, 'data') + logs_path = os.path.join(logs_path, 'logs.yml') + mock_r = mock.Mock() + mock_r.iter_content.return_value = [b'test_chunk1', b'test_chunk2'] + mock_get.return_value.__enter__.return_value = mock_r + with open(logs_path, "w") as fd: + fd.write("test_logs") + + self.logger.download_logs( + mock.sentinel.app, + logs_path, + start_time=mock.sentinel.start_time, + end_time=mock.sentinel.end_time + ) + + with open(logs_path, "r") as fd: + result = fd.read() + self.assertEqual( + result, + "test_chunk1test_chunk2" + ) + mock_get.assert_called_once_with( + mock_construct_url.return_value, + headers=self.logger._auth_headers, + stream=True + ) + mock_construct_url.assert_called_once_with( + "logs/sentinel.app/", + { + "start_date": mock_convert_period_to_timestamp.return_value, + "end_date": mock_convert_period_to_timestamp.return_value, + } + ) + + def test_download_logs_no_app(self): + self.assertRaises( + exceptions.CoriolisException, + self.logger.download_logs, + "", + None + ) + + @mock.patch.object(requests, "get") + def test_list_logs(self, mock_get): + mock_get.return_value.raise_for_status.return_value = None + mock_get.return_value.json.return_value = { + "logs": ["mock_log1", "mock_log2"] + } + + result = self.logger.list_logs() + + self.assertEqual( + ['mock_log1', 'mock_log2'], + result + ) + + +class CoriolisLogDownloadManagerTestCase( + test_base.CoriolisBaseTestCase): + """Test suite for the Coriolis v1 Coriolis Log Download Manager.""" + + def setUp(self): + mock_client = mock.Mock() + super(CoriolisLogDownloadManagerTestCase, self).setUp() + self.logger = logging.CoriolisLogDownloadManager(mock_client) + self.logger._coriolis_cli = mock.Mock() + self.logger.resource_class = mock.Mock() + + def test_list(self): + self.logger._coriolis_cli.list_logs.return_value = [ + "mock_log1", "mock_log2"] + + result = self.logger.list() + + self.assertEqual( + [self.logger.resource_class.return_value, + self.logger.resource_class.return_value], + result + ) + self.logger.resource_class.assert_has_calls([ + mock.call(self.logger, "mock_log1", loaded=True), + mock.call(self.logger, "mock_log2", loaded=True) + ]) + + def test_get(self): + result = self.logger.get( + mock.sentinel.app, + mock.sentinel.to, + start_time=mock.sentinel.start_time, + end_time=mock.sentinel.end_time, + ) + + self.assertEqual( + self.logger._coriolis_cli.download_logs.return_value, + result + ) + self.logger._coriolis_cli.download_logs.assert_called_once_with( + mock.sentinel.app, + mock.sentinel.to, + start_time=mock.sentinel.start_time, + end_time=mock.sentinel.end_time, + ) + + def test_stream(self): + self.logger.stream( + app_name=mock.sentinel.app_name, + severity=mock.sentinel.severity, + ) + + self.logger._coriolis_cli.stream_logs.assert_called_once_with( + app_name=mock.sentinel.app_name, + severity=mock.sentinel.severity, + )