14325: Merge branch 'master'
[arvados.git] / sdk / cwl / tests / test_util.py
1 # Copyright (C) The Arvados Authors. All rights reserved.
2 #
3 # SPDX-License-Identifier: Apache-2.0
4
5 from builtins import bytes
6
7 import unittest
8 import mock
9 import datetime
10 import httplib2
11
12 from arvados_cwl.util import *
13 from arvados.errors import ApiError
14
15 class MockDateTime(datetime.datetime):
16     @classmethod
17     def utcnow(cls):
18         return datetime.datetime(2018, 1, 1, 0, 0, 0, 0)
19
20 datetime.datetime = MockDateTime
21
22 class TestUtil(unittest.TestCase):
23     def test_get_intermediate_collection_info(self):
24         name = "one"
25         current_container = {"uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
26         intermediate_output_ttl = 120
27
28         info = get_intermediate_collection_info(name, current_container, intermediate_output_ttl)
29
30         self.assertEqual(info["name"], "Intermediate collection for step one")
31         self.assertEqual(info["trash_at"], datetime.datetime(2018, 1, 1, 0, 2, 0, 0))
32         self.assertEqual(info["properties"], {"type" : "intermediate", "container" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
33
34     def test_get_current_container_success(self):
35         api = mock.MagicMock()
36         api.containers().current().execute.return_value = {"uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
37
38         current_container = get_current_container(api)
39
40         self.assertEqual(current_container, {"uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
41
42     def test_get_current_container_error(self):
43         api = mock.MagicMock()
44         api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 300}), bytes(b""))
45         logger = mock.MagicMock()
46
47         with self.assertRaises(ApiError):
48             get_current_container(api, num_retries=0, logger=logger)
49
50     def test_get_current_container_404_error(self):
51         api = mock.MagicMock()
52         api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 404}), bytes(b""))
53         logger = mock.MagicMock()
54
55         current_container = get_current_container(api, num_retries=0, logger=logger)
56         self.assertEqual(current_container, None)