#
# SPDX-License-Identifier: Apache-2.0
+from builtins import bytes
+
import unittest
import mock
import datetime
class TestUtil(unittest.TestCase):
def test_get_intermediate_collection_info(self):
+ name = "one"
current_container = {"uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
intermediate_output_ttl = 120
- info = get_intermediate_collection_info(current_container, intermediate_output_ttl)
+ info = get_intermediate_collection_info(name, current_container, intermediate_output_ttl)
- self.assertEqual(info["name"], "Intermediate collection")
+ self.assertEqual(info["name"], "Intermediate collection for step one")
self.assertEqual(info["trash_at"], datetime.datetime(2018, 1, 1, 0, 2, 0, 0))
self.assertEqual(info["properties"], {"type" : "intermediate", "container" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
def test_get_current_container_error(self):
api = mock.MagicMock()
- api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 300}), "")
+ api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 300}), bytes(b""))
+ logger = mock.MagicMock()
+
+ with self.assertRaises(ApiError):
+ get_current_container(api, num_retries=0, logger=logger)
+
+ def test_get_current_container_404_error(self):
+ api = mock.MagicMock()
+ api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 404}), bytes(b""))
logger = mock.MagicMock()
- self.assertRaises(ApiError, get_current_container(api, num_retries=0, logger=logger))
+ current_container = get_current_container(api, num_retries=0, logger=logger)
+ self.assertEqual(current_container, None)
\ No newline at end of file