services/nodemanager/tests/test_computenode_driver.py

   1 #!/usr/bin/env python
   2 # Copyright (C) The Arvados Authors. All rights reserved.
   3 #
   4 # SPDX-License-Identifier: AGPL-3.0
   5
   6 from __future__ import absolute_import, print_function
   7
   8 import unittest
   9
  10 import libcloud.common.types as cloud_types
  11 import mock
  12
  13 import arvnodeman.computenode.driver as driver_base
  14 from . import testutil
  15
  16 class ComputeNodeDriverTestCase(unittest.TestCase):
  17     def setUp(self):
  18         self.driver_mock = mock.MagicMock(name='driver_mock')
  19         driver_base.BaseComputeNodeDriver.SEARCH_CACHE = {}
  20
  21     def test_search_for_now_uses_public_method(self):
  22         image = testutil.cloud_object_mock(1)
  23         self.driver_mock().list_images.return_value = [image]
  24         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  25         self.assertIs(image, driver.search_for_now('id_1', 'list_images'))
  26         self.assertEqual(1, self.driver_mock().list_images.call_count)
  27
  28     def test_search_for_now_uses_private_method(self):
  29         net = testutil.cloud_object_mock(1)
  30         self.driver_mock().ex_list_networks.return_value = [net]
  31         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  32         self.assertIs(net, driver.search_for_now('id_1', 'ex_list_networks'))
  33         self.assertEqual(1, self.driver_mock().ex_list_networks.call_count)
  34
  35     def test_search_for_now_raises_ValueError_on_zero_results(self):
  36         self.driver_mock().list_images.return_value = []
  37         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  38         with self.assertRaises(ValueError) as test:
  39             driver.search_for_now('id_1', 'list_images')
  40
  41     def test_search_for_now_raises_ValueError_on_extra_results(self):
  42         image = testutil.cloud_object_mock(1)
  43         self.driver_mock().list_images.return_value = [image, image]
  44         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  45         with self.assertRaises(ValueError) as test:
  46             driver.search_for_now('id_1', 'list_images')
  47
  48     def test_search_for_now_does_not_cache_results(self):
  49         image1 = testutil.cloud_object_mock(1)
  50         image2 = testutil.cloud_object_mock(1)
  51         self.driver_mock().list_images.side_effect = [[image1], [image2]]
  52         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  53         self.assertIsNot(driver.search_for_now('id_1', 'list_images'),
  54                          driver.search_for_now('id_1', 'list_images'))
  55         self.assertEqual(2, self.driver_mock().list_images.call_count)
  56
  57     def test_search_for_returns_cached_results(self):
  58         image1 = testutil.cloud_object_mock(1)
  59         image2 = testutil.cloud_object_mock(1)
  60         self.driver_mock().list_images.side_effect = [[image1], [image2]]
  61         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  62         self.assertIs(driver.search_for('id_1', 'list_images'),
  63                       driver.search_for('id_1', 'list_images'))
  64         self.assertEqual(1, self.driver_mock().list_images.call_count)