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)