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 import unittest
   6 import mock
   7 import datetime
   8 import httplib2
   9
  10 from arvados_cwl.util import *
  11 from arvados.errors import ApiError
  12
  13 class MockDateTime(datetime.datetime):
  14     @classmethod
  15     def utcnow(cls):
  16         return datetime.datetime(2018, 1, 1, 0, 0, 0, 0)
  17
  18 datetime.datetime = MockDateTime
  19
  20 class TestUtil(unittest.TestCase):
  21     def test_get_intermediate_collection_info(self):
  22         name = "one"
  23         current_container = {"uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
  24         intermediate_output_ttl = 120
  25
  26         info = get_intermediate_collection_info(name, current_container, intermediate_output_ttl)
  27
  28         self.assertEqual(info["name"], "Intermediate collection for step one")
  29         self.assertEqual(info["trash_at"], datetime.datetime(2018, 1, 1, 0, 2, 0, 0))
  30         self.assertEqual(info["properties"], {"type" : "intermediate", "container" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
  31
  32     def test_get_current_container_success(self):
  33         api = mock.MagicMock()
  34         api.containers().current().execute.return_value = {"uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
  35
  36         current_container = get_current_container(api)
  37
  38         self.assertEqual(current_container, {"uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
  39
  40     def test_get_current_container_error(self):
  41         api = mock.MagicMock()
  42         api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 300}), "")
  43         logger = mock.MagicMock()
  44
  45         self.assertRaises(ApiError, get_current_container(api, num_retries=0, logger=logger))