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 datetime
   9 import httplib2
  10
  11 from unittest import mock
  12
  13 from arvados_cwl.util import *
  14 from arvados.errors import ApiError
  15 from arvados_cwl.util import common_prefix
  16
  17 class MockDateTime(datetime.datetime):
  18     @classmethod
  19     def utcnow(cls):
  20         return datetime.datetime(2018, 1, 1, 0, 0, 0, 0)
  21
  22 datetime.datetime = MockDateTime
  23
  24 class TestUtil(unittest.TestCase):
  25     def test_get_intermediate_collection_info(self):
  26         name = "one"
  27         current_container = {"uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
  28         intermediate_output_ttl = 120
  29
  30         info = get_intermediate_collection_info(name, current_container, intermediate_output_ttl)
  31
  32         self.assertEqual(info["name"], "Intermediate collection for step one")
  33         self.assertEqual(info["trash_at"], datetime.datetime(2018, 1, 1, 0, 2, 0, 0))
  34         self.assertEqual(info["properties"], {"type" : "intermediate", "container_uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
  35
  36     def test_get_current_container_success(self):
  37         api = mock.MagicMock()
  38         api.containers().current().execute.return_value = {"uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
  39
  40         current_container = get_current_container(api)
  41
  42         self.assertEqual(current_container, {"uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
  43
  44     def test_get_current_container_error(self):
  45         api = mock.MagicMock()
  46         api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 300}), bytes(b""))
  47         logger = mock.MagicMock()
  48
  49         with self.assertRaises(ApiError):
  50             get_current_container(api, num_retries=0, logger=logger)
  51
  52     def test_get_current_container_404_error(self):
  53         api = mock.MagicMock()
  54         api.containers().current().execute.side_effect = ApiError(httplib2.Response({"status": 404}), bytes(b""))
  55         logger = mock.MagicMock()
  56
  57         current_container = get_current_container(api, num_retries=0, logger=logger)
  58         self.assertEqual(current_container, None)
  59
  60     def test_common_prefix(self):
  61         self.assertEqual(common_prefix("file:///foo/bar", ["file:///foo/bar/baz"]), "file:///foo/")
  62         self.assertEqual(common_prefix("file:///foo", ["file:///foo", "file:///foo/bar", "file:///foo/bar/"]), "file:///")
  63         self.assertEqual(common_prefix("file:///foo/", ["file:///foo/", "file:///foo/bar", "file:///foo/bar/"]), "file:///foo/")
  64         self.assertEqual(common_prefix("file:///foo/bar", ["file:///foo/bar", "file:///foo/baz", "file:///foo/quux/q2"]), "file:///foo/")
  65         self.assertEqual(common_prefix("file:///foo/bar/", ["file:///foo/bar/", "file:///foo/baz", "file:///foo/quux/q2"]), "file:///foo/")
  66         self.assertEqual(common_prefix("file:///foo/bar/splat", ["file:///foo/bar/splat", "file:///foo/baz", "file:///foo/quux/q2"]), "file:///foo/")
  67         self.assertEqual(common_prefix("file:///foo/bar/splat", ["file:///foo/bar/splat", "file:///nope", "file:///foo/quux/q2"]), "file:///")
  68         self.assertEqual(common_prefix("file:///blub/foo", ["file:///blub/foo", "file:///blub/foo/bar", "file:///blub/foo/bar/"]), "file:///blub/")
  69
  70         # sanity check, the subsequent code strips off the prefix so
  71         # just confirm the logic doesn't have a fencepost error
  72         prefix = "file:///"
  73         self.assertEqual("file:///foo/bar"[len(prefix):], "foo/bar")