diff --git a/coriolisclient/tests/v1/test_licensing.py b/coriolisclient/tests/v1/test_licensing.py
new file mode 100644
index 0000000..c8b4de8
--- /dev/null
+++ b/coriolisclient/tests/v1/test_licensing.py
@@ -0,0 +1,371 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+import requests
+
+from coriolisclient import exceptions
+from coriolisclient.tests import test_base
+from coriolisclient.v1 import licensing
+
+
+class LicensingClientTestCase(
+ test_base.CoriolisBaseTestCase):
+ """Test suite for the Coriolis v1 Licensing Client."""
+
+ def setUp(self):
+ mock_client = mock.Mock()
+ super(LicensingClientTestCase, self).setUp()
+ self.licence = licensing.LicensingClient(
+ mock_client, "endpoint_name")
+ mock_client.verify = True
+ self.licence._cli = mock_client
+
+ def test_get_licensing_endpoint_url(self):
+ self.licence._cli.get_endpoint.return_value = "url/endpoint_url/"
+
+ result = self.licence._get_licensing_endpoint_url()
+
+ self.assertEqual(
+ "url/endpoint_url",
+ result
+ )
+ self.licence._cli.get_endpoint.assert_called_once_with(
+ service_type="endpoint_name")
+
+ def test_get_licensing_endpoint_url_raises(self):
+ self.licence._cli.get_endpoint.side_effect = Exception()
+
+ with self.assertLogs(level="WARN"):
+ self.assertRaises(
+ exceptions.LicensingEndpointNotFound,
+ self.licence._get_licensing_endpoint_url
+ )
+ self.licence._cli.get_endpoint.assert_called_once_with(
+ service_type="endpoint_name")
+
+ @mock.patch.object(licensing.LicensingClient,
+ "_get_licensing_endpoint_url")
+ def test_do_req_raw(
+ self,
+ mock_get_licensing_endpoint_url
+ ):
+ mock_method = mock.Mock()
+ mock_resp = mock.Mock()
+ mock_resp.ok = True
+ mock_method.return_value = mock_resp
+ setattr(requests, "mock_method", mock_method)
+ mock_get_licensing_endpoint_url.return_value = 'url/endpoint_url/'
+ result = self.licence._do_req(
+ method_name="mock_method",
+ resource='url/resource_url/',
+ body=None,
+ response_key=None,
+ raw_response=True
+ )
+
+ self.assertEqual(
+ mock_resp,
+ result
+ )
+ mock_method.assert_called_once_with(
+ 'url/endpoint_url/url/resource_url/',
+ verify=self.licence._cli.verify
+ )
+
+ @mock.patch.object(licensing.LicensingClient,
+ "_get_licensing_endpoint_url")
+ def test_do_req_json(
+ self,
+ mock_get_licensing_endpoint_url
+ ):
+ mock_method = mock.Mock()
+ mock_resp = mock.Mock()
+ mock_resp.ok = True
+ mock_resp.json.return_value = {"response_key": mock.sentinel.data}
+ mock_method.return_value = mock_resp
+ setattr(requests, "mock_method", mock_method)
+ mock_get_licensing_endpoint_url.return_value = 'url/endpoint_url/'
+ result = self.licence._do_req(
+ method_name="mock_method",
+ resource='url/resource_url/',
+ body={"mock_body": "value"},
+ response_key='response_key',
+ raw_response=False
+ )
+
+ self.assertEqual(
+ mock.sentinel.data,
+ result
+ )
+ mock_method.assert_called_once_with(
+ 'url/endpoint_url/url/resource_url/',
+ verify=self.licence._cli.verify,
+ data='{"mock_body": "value"}'
+ )
+
+ @mock.patch.object(licensing.LicensingClient,
+ "_get_licensing_endpoint_url")
+ def test_do_req_error(
+ self,
+ mock_get_licensing_endpoint_url
+ ):
+ mock_method = mock.Mock()
+ mock_resp = mock.Mock()
+ mock_resp.ok = False
+ mock_resp.json.side_effect = Exception
+ mock_resp.raise_for_status.side_effect = exceptions.CoriolisException
+ mock_method.return_value = mock_resp
+ setattr(requests, "mock_method", mock_method)
+ mock_get_licensing_endpoint_url.return_value = 'url/endpoint_url/'
+
+ with self.assertLogs(level="DEBUG"):
+ self.assertRaises(
+ exceptions.CoriolisException,
+ self.licence._do_req,
+ method_name="mock_method",
+ resource='url/resource_url/',
+ body=None,
+ response_key='response_key',
+ raw_response=False
+ )
+ mock_method.assert_called_once_with(
+ 'url/endpoint_url/url/resource_url/',
+ verify=self.licence._cli.verify
+ )
+
+ @mock.patch.object(licensing.LicensingClient,
+ "_get_licensing_endpoint_url")
+ def test_do_req_http_error(
+ self,
+ mock_get_licensing_endpoint_url
+ ):
+ mock_method = mock.Mock()
+ mock_resp = mock.Mock()
+ mock_resp.ok = False
+ mock_resp.json.return_value = {"error": {"code": 123, "message": ""}}
+ mock_method.return_value = mock_resp
+ setattr(requests, "mock_method", mock_method)
+ mock_get_licensing_endpoint_url.return_value = 'url/endpoint_url/'
+
+ self.assertRaises(
+ exceptions.HTTPError,
+ self.licence._do_req,
+ method_name="mock_method",
+ resource='url/resource_url/',
+ body=None,
+ response_key='response_key',
+ raw_response=False
+ )
+ mock_method.assert_called_once_with(
+ 'url/endpoint_url/url/resource_url/',
+ verify=self.licence._cli.verify
+ )
+
+ @mock.patch.object(licensing.LicensingClient,
+ "_get_licensing_endpoint_url")
+ def test_do_req_response_key_error(
+ self,
+ mock_get_licensing_endpoint_url
+ ):
+ mock_method = mock.Mock()
+ mock_resp = mock.Mock()
+ mock_resp.ok = False
+ mock_resp.json.return_value = {"response_key": mock.sentinel.data}
+ mock_method.return_value = mock_resp
+ setattr(requests, "mock_method", mock_method)
+ mock_get_licensing_endpoint_url.return_value = 'url/endpoint_url/'
+
+ self.assertRaises(
+ ValueError,
+ self.licence._do_req,
+ method_name="mock_method",
+ resource='url/resource_url/',
+ body=None,
+ response_key='invalid',
+ raw_response=False
+ )
+ mock_method.assert_called_once_with(
+ 'url/endpoint_url/url/resource_url/',
+ verify=self.licence._cli.verify
+ )
+
+ def test_do_req_method_error(self):
+ setattr(requests, "mock_method", None)
+
+ self.assertRaises(
+ ValueError,
+ self.licence._do_req,
+ method_name="mock_method",
+ resource='url/resource_url/',
+ body=None,
+ response_key='invalid',
+ raw_response=False
+ )
+
+ @mock.patch.object(licensing.LicensingClient, '_do_req')
+ def test_get(self, mock_do_req):
+ result = self.licence.get(
+ resource=mock.sentinel.resource,
+ body=mock.sentinel.body,
+ response_key=mock.sentinel.response_key,
+ raw_response=False
+ )
+ self.assertEqual(
+ mock_do_req.return_value,
+ result
+ )
+ mock_do_req.assert_called_once_with(
+ 'GET',
+ mock.sentinel.resource,
+ response_key=mock.sentinel.response_key,
+ body=mock.sentinel.body,
+ raw_response=False
+ )
+
+ @mock.patch.object(licensing.LicensingClient, '_do_req')
+ def test_post(self, mock_do_req):
+ result = self.licence.post(
+ resource=mock.sentinel.resource,
+ body=mock.sentinel.body,
+ response_key=mock.sentinel.response_key,
+ raw_response=False
+ )
+ self.assertEqual(
+ mock_do_req.return_value,
+ result
+ )
+ mock_do_req.assert_called_once_with(
+ 'POST',
+ mock.sentinel.resource,
+ response_key=mock.sentinel.response_key,
+ body=mock.sentinel.body,
+ raw_response=False
+ )
+
+ @mock.patch.object(licensing.LicensingClient, '_do_req')
+ def test_delete(self, mock_do_req):
+ result = self.licence.delete(
+ resource=mock.sentinel.resource,
+ body=mock.sentinel.body,
+ response_key=mock.sentinel.response_key,
+ raw_response=False
+ )
+ self.assertEqual(
+ mock_do_req.return_value,
+ result
+ )
+ mock_do_req.assert_called_once_with(
+ 'DELETE',
+ mock.sentinel.resource,
+ response_key=mock.sentinel.response_key,
+ body=mock.sentinel.body,
+ raw_response=False
+ )
+
+
+class LicensingManagerTestCase(
+ test_base.CoriolisBaseTestCase):
+ """Test suite for the Coriolis v1 Licensing Client."""
+
+ @mock.patch.object(licensing, 'LicensingClient')
+ def setUp(self, mock_LicensingClient):
+ mock_client = mock.Mock()
+ super(LicensingManagerTestCase, self).setUp()
+ self.licence = licensing.LicensingManager(mock_client)
+ self.licence._licensing_cli = mock_LicensingClient
+
+ def test_status(self):
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.status(mock.sentinel.appliance_id)
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.licence._licensing_cli.get.assert_called_once_with(
+ '/appliances/%s/status' % mock.sentinel.appliance_id,
+ response_key='appliance_licence_status')
+ mock_resource_class.assert_called_once_with(
+ self.licence, self.licence._licensing_cli.get.return_value,
+ loaded=True)
+
+ def test_list(self):
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+ self.licence._licensing_cli.get.return_value = {
+ "licence1": "mock_licence1",
+ "licence2": "mock_licence2"
+ }
+
+ result = self.licence.list(mock.sentinel.appliance_id)
+
+ self.assertEqual(
+ [mock_resource_class.return_value,
+ mock_resource_class.return_value],
+ result
+ )
+ self.licence._licensing_cli.get.assert_called_once_with(
+ '/appliances/%s/licences' % mock.sentinel.appliance_id,
+ response_key='licences')
+ mock_resource_class.assert_has_calls([
+ mock.call(self.licence, "licence1", loaded=True),
+ mock.call(self.licence, "licence2", loaded=True)
+ ])
+
+ def test_register(self):
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.register(
+ mock.sentinel.appliance_id, mock.sentinel.licence)
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.licence._licensing_cli.post.assert_called_once_with(
+ '/appliances/%s/licences' % mock.sentinel.appliance_id,
+ body=mock.sentinel.licence,
+ response_key='licence')
+ mock_resource_class.assert_called_once_with(
+ self.licence, self.licence._licensing_cli.post.return_value,
+ loaded=True)
+
+ def test_show(self):
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.show(
+ mock.sentinel.appliance_id, mock.sentinel.licence_id)
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.licence._licensing_cli.get.assert_called_once_with(
+ '/appliances/%s/licences/%s' % (mock.sentinel.appliance_id,
+ mock.sentinel.licence_id),
+ response_key='licence')
+ mock_resource_class.assert_called_once_with(
+ self.licence, self.licence._licensing_cli.get.return_value,
+ loaded=True)
+
+ def test_delete(self):
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.delete(
+ mock.sentinel.appliance_id, mock.sentinel.licence_id)
+
+ self.assertEqual(
+ self.licence._licensing_cli.delete.return_value,
+ result
+ )
+ self.licence._licensing_cli.delete.assert_called_once_with(
+ '/appliances/%s/licences/%s' % (mock.sentinel.appliance_id,
+ mock.sentinel.licence_id),
+ raw_response=True)
diff --git a/coriolisclient/tests/v1/test_licensing_appliances.py b/coriolisclient/tests/v1/test_licensing_appliances.py
new file mode 100644
index 0000000..394defe
--- /dev/null
+++ b/coriolisclient/tests/v1/test_licensing_appliances.py
@@ -0,0 +1,79 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolisclient.tests import test_base
+from coriolisclient.v1 import licensing
+from coriolisclient.v1 import licensing_appliances
+
+
+class LicensingAppliancesManagerTestCase(
+ test_base.CoriolisBaseTestCase):
+ """Test suite for the Coriolis v1 Licensing Appliances Manager."""
+
+ @mock.patch.object(licensing, 'LicensingClient')
+ def setUp(self, mock_LicensingClient):
+ mock_client = mock.Mock()
+ self.mock_licencing_client = mock.Mock()
+ mock_LicensingClient.return_value = self.mock_licencing_client
+ super(LicensingAppliancesManagerTestCase, self).setUp()
+ self.licence = licensing_appliances.LicensingAppliancesManager(
+ mock_client)
+
+ def test_list(self):
+ self.licence._licensing_cli.get.return_value = {
+ "appliance1": "mock_appliance1",
+ "appliance2": "mock_appliance2"
+ }
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.list()
+
+ self.assertEqual(
+ [mock_resource_class.return_value,
+ mock_resource_class.return_value],
+ result
+ )
+ self.mock_licencing_client.get.assert_called_once_with(
+ '/appliances', response_key='appliances')
+ mock_resource_class.assert_has_calls([
+ mock.call(self.licence, "appliance1", loaded=True),
+ mock.call(self.licence, "appliance2", loaded=True)
+ ])
+
+ def test_show(self):
+ self.licence._licensing_cli.get.return_value = mock.sentinel.data
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.show(mock.sentinel.appliance_id)
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.mock_licencing_client.get.assert_called_once_with(
+ '/appliances/%s' % mock.sentinel.appliance_id,
+ response_key='appliance')
+ mock_resource_class.assert_called_once_with(
+ self.licence, mock.sentinel.data, loaded=True
+ )
+
+ def test_create(self):
+ self.licence._licensing_cli.post.return_value = mock.sentinel.data
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.create()
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.mock_licencing_client.post.assert_called_once_with(
+ '/appliances', response_key='appliance')
+ mock_resource_class.assert_called_once_with(
+ self.licence, mock.sentinel.data, loaded=True
+ )
diff --git a/coriolisclient/tests/v1/test_licensing_reservations.py b/coriolisclient/tests/v1/test_licensing_reservations.py
new file mode 100644
index 0000000..bbf8cb1
--- /dev/null
+++ b/coriolisclient/tests/v1/test_licensing_reservations.py
@@ -0,0 +1,85 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolisclient.tests import test_base
+from coriolisclient.v1 import licensing
+from coriolisclient.v1 import licensing_reservations
+
+
+class LicensingReservationsManagerTestCase(
+ test_base.CoriolisBaseTestCase):
+ """Test suite for the Coriolis v1 Licensing Reservations Manager."""
+
+ @mock.patch.object(licensing, 'LicensingClient')
+ def setUp(self, mock_LicensingClient):
+ mock_client = mock.Mock()
+ self.mock_licencing_client = mock.Mock()
+ mock_LicensingClient.return_value = self.mock_licencing_client
+ super(LicensingReservationsManagerTestCase, self).setUp()
+ self.licence = licensing_reservations.LicensingReservationsManager(
+ mock_client)
+
+ def test_list(self):
+ self.licence._licensing_cli.get.return_value = {
+ "reservation1": "mock_reservation1",
+ "reservation2": "mock_reservation2"
+ }
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.list(mock.sentinel.appliance_id)
+
+ self.assertEqual(
+ [mock_resource_class.return_value,
+ mock_resource_class.return_value],
+ result
+ )
+ self.mock_licencing_client.get.assert_called_once_with(
+ '/appliances/%s/reservations' % mock.sentinel.appliance_id,
+ response_key='reservations')
+ mock_resource_class.assert_has_calls([
+ mock.call(self.licence, "reservation1", loaded=True),
+ mock.call(self.licence, "reservation2", loaded=True)
+ ])
+
+ def test_show(self):
+ self.licence._licensing_cli.get.return_value = mock.sentinel.data
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.show(
+ mock.sentinel.appliance_id, mock.sentinel.reservation_id)
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.mock_licencing_client.get.assert_called_once_with(
+ '/appliances/%s/reservations/%s' % (mock.sentinel.appliance_id,
+ mock.sentinel.reservation_id),
+ response_key='reservation')
+ mock_resource_class.assert_called_once_with(
+ self.licence, mock.sentinel.data, loaded=True
+ )
+
+ def test_refresh(self):
+ self.licence._licensing_cli.post.return_value = mock.sentinel.data
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.refresh(
+ mock.sentinel.appliance_id, mock.sentinel.reservation_id)
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.mock_licencing_client.post.assert_called_once_with(
+ '/appliances/%s/reservations/%s/refresh' %
+ (mock.sentinel.appliance_id, mock.sentinel.reservation_id),
+ response_key='reservation')
+ mock_resource_class.assert_called_once_with(
+ self.licence, mock.sentinel.data, loaded=True
+ )
diff --git a/coriolisclient/tests/v1/test_licensing_server.py b/coriolisclient/tests/v1/test_licensing_server.py
new file mode 100644
index 0000000..721f49f
--- /dev/null
+++ b/coriolisclient/tests/v1/test_licensing_server.py
@@ -0,0 +1,39 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolisclient.tests import test_base
+from coriolisclient.v1 import licensing
+from coriolisclient.v1 import licensing_server
+
+
+class LicensingServerManagerTestCase(
+ test_base.CoriolisBaseTestCase):
+ """Test suite for the Coriolis v1 Licensing Server Manager."""
+
+ @mock.patch.object(licensing, 'LicensingClient')
+ def setUp(self, mock_LicensingClient):
+ mock_client = mock.Mock()
+ self.mock_licencing_client = mock.Mock()
+ mock_LicensingClient.return_value = self.mock_licencing_client
+ super(LicensingServerManagerTestCase, self).setUp()
+ self.licence = licensing_server.LicensingServerManager(
+ mock_client)
+
+ def test_status(self):
+ mock_resource_class = mock.Mock()
+ self.licence.resource_class = mock_resource_class
+
+ result = self.licence.status()
+
+ self.assertEqual(
+ mock_resource_class.return_value,
+ result
+ )
+ self.mock_licencing_client.get.assert_called_once_with(
+ '/status',
+ response_key='status')
+ mock_resource_class.assert_called_once_with(
+ self.licence, self.mock_licencing_client.get.return_value,
+ loaded=True)
diff --git a/coriolisclient/tests/v1/test_logging.py b/coriolisclient/tests/v1/test_logging.py
new file mode 100644
index 0000000..a2b449e
--- /dev/null
+++ b/coriolisclient/tests/v1/test_logging.py
@@ -0,0 +1,282 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import copy
+import datetime
+import ddt
+import requests
+import tempfile
+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()
+ mock_client.verify = True
+ super(LoggingClientTestCase, self).setUp()
+ self.logger = logging.LoggingClient(mock_client)
+ self.datetime = copy.deepcopy(datetime.datetime)
+
+ @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),
+ ("1", 1, False),
+ ("1234567890123456789", None, True),
+ ("abc", None, True),
+ ("", None, True),
+ )
+ @ddt.unpack
+ def test_convert_period_to_timestamp(
+ self,
+ period,
+ expected_result,
+ raises
+ ):
+ if raises is False:
+ result = self.logger._convert_period_to_timestamp(period)
+ self.assertEqual(
+ expected_result,
+ result
+ )
+ else:
+ self.assertRaises(
+ exceptions.CoriolisException,
+ self.logger._convert_period_to_timestamp,
+ period
+ )
+
+ @mock.patch.object(datetime, 'datetime')
+ def test_convert_period_to_timestamp_period(
+ self,
+ mock_datetime
+ ):
+ mock_datetime.utcnow.return_value = self.datetime.fromtimestamp(100000)
+ result = self.logger._convert_period_to_timestamp("1d")
+ self.assertEqual(
+ 13600,
+ result
+ )
+
+ @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
+ ):
+ 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 tempfile.NamedTemporaryFile() as fd:
+ self.logger.download_logs(
+ mock.sentinel.app,
+ fd.name,
+ start_time=mock.sentinel.start_time,
+ end_time=mock.sentinel.end_time
+ )
+
+ result = fd.read()
+ self.assertEqual(
+ result,
+ b'test_chunk1test_chunk2'
+ )
+ mock_get.assert_called_once_with(
+ mock_construct_url.return_value,
+ headers=self.logger._auth_headers,
+ stream=True,
+ verify=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,
+ )
diff --git a/coriolisclient/v1/licensing_appliances.py b/coriolisclient/v1/licensing_appliances.py
index 0121107..7cfc208 100644
--- a/coriolisclient/v1/licensing_appliances.py
+++ b/coriolisclient/v1/licensing_appliances.py
@@ -41,5 +41,5 @@ def show(self, appliance_id):
def create(self):
url = '/appliances'
- data = self._licensing_cli.post(url, response_key='apppliance')
+ data = self._licensing_cli.post(url, response_key='appliance')
return self.resource_class(self, data, loaded=True)
diff --git a/coriolisclient/v1/logging.py b/coriolisclient/v1/logging.py
index e69fe52..2e841c2 100644
--- a/coriolisclient/v1/logging.py
+++ b/coriolisclient/v1/logging.py
@@ -137,7 +137,7 @@ def _convert_period_to_timestamp(self, period):
_ = datetime.datetime.fromtimestamp(int(period))
return int(period)
except (OverflowError, OSError):
- LOG.warning("Failed to initialize timestamp from period value: %d",
+ LOG.warning("Failed to initialize timestamp from period value: %s",
period)
except ValueError:
LOG.warning("Invalid value type for period: %s", period)
@@ -157,7 +157,7 @@ def _convert_period_to_timestamp(self, period):
if conv_unit is None:
raise exceptions.CoriolisException("invalid period %s" % period)
- args = {conv_unit: count}
+ args = {conv_unit: int(count)}
tm = datetime.datetime.utcnow() - datetime.timedelta(**args)
return int(tm.timestamp())