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 import arvnodeman.status as status
  15 import arvnodeman.config as config
  16 from . import testutil
  17
  18 class ComputeNodeDriverTestCase(unittest.TestCase):
  19     def setUp(self):
  20         self.driver_mock = mock.MagicMock(name='driver_mock')
  21         driver_base.BaseComputeNodeDriver.SEARCH_CACHE = {}
  22
  23     def test_search_for_now_uses_public_method(self):
  24         image = testutil.cloud_object_mock(1)
  25         self.driver_mock().list_images.return_value = [image]
  26         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  27         self.assertIs(image, driver.search_for_now('id_1', 'list_images'))
  28         self.assertEqual(1, self.driver_mock().list_images.call_count)
  29
  30     def test_search_for_now_uses_private_method(self):
  31         net = testutil.cloud_object_mock(1)
  32         self.driver_mock().ex_list_networks.return_value = [net]
  33         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  34         self.assertIs(net, driver.search_for_now('id_1', 'ex_list_networks'))
  35         self.assertEqual(1, self.driver_mock().ex_list_networks.call_count)
  36
  37     def test_search_for_now_raises_ValueError_on_zero_results(self):
  38         self.driver_mock().list_images.return_value = []
  39         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  40         with self.assertRaises(ValueError) as test:
  41             driver.search_for_now('id_1', 'list_images')
  42
  43     def test_search_for_now_raises_ValueError_on_extra_results(self):
  44         image = testutil.cloud_object_mock(1)
  45         self.driver_mock().list_images.return_value = [image, image]
  46         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  47         with self.assertRaises(ValueError) as test:
  48             driver.search_for_now('id_1', 'list_images')
  49
  50     def test_search_for_now_does_not_cache_results(self):
  51         image1 = testutil.cloud_object_mock(1)
  52         image2 = testutil.cloud_object_mock(1)
  53         self.driver_mock().list_images.side_effect = [[image1], [image2]]
  54         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  55         self.assertIsNot(driver.search_for_now('id_1', 'list_images'),
  56                          driver.search_for_now('id_1', 'list_images'))
  57         self.assertEqual(2, self.driver_mock().list_images.call_count)
  58
  59     def test_search_for_returns_cached_results(self):
  60         image1 = testutil.cloud_object_mock(1)
  61         image2 = testutil.cloud_object_mock(1)
  62         self.driver_mock().list_images.side_effect = [[image1], [image2]]
  63         driver = driver_base.BaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  64         self.assertIs(driver.search_for('id_1', 'list_images'),
  65                       driver.search_for('id_1', 'list_images'))
  66         self.assertEqual(1, self.driver_mock().list_images.call_count)
  67
  68
  69     class TestBaseComputeNodeDriver(driver_base.BaseComputeNodeDriver):
  70         def arvados_create_kwargs(self, size, arvados_node):
  71             return {'name': arvados_node}
  72
  73
  74     def test_create_node_only_cloud_errors_are_counted(self):
  75         status.tracker.update({'cloud_errors': 0})
  76         errors = [(config.CLOUD_ERRORS[0], True), (KeyError, False)]
  77         self.driver_mock().list_images.return_value = []
  78         driver = self.TestBaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  79         error_count = 0
  80         for an_error, is_cloud_error in errors:
  81             self.driver_mock().create_node.side_effect = an_error
  82             with self.assertRaises(an_error):
  83                 driver.create_node('1', 'id_1')
  84             if is_cloud_error:
  85                 error_count += 1
  86             self.assertEqual(error_count, status.tracker.get('cloud_errors'))
  87
  88     def test_list_nodes_only_cloud_errors_are_counted(self):
  89         status.tracker.update({'cloud_errors': 0})
  90         errors = [(config.CLOUD_ERRORS[0], True), (KeyError, False)]
  91         driver = self.TestBaseComputeNodeDriver({}, {}, {}, self.driver_mock)
  92         error_count = 0
  93         for an_error, is_cloud_error in errors:
  94             self.driver_mock().list_nodes.side_effect = an_error
  95             with self.assertRaises(an_error):
  96                 driver.list_nodes()
  97             if is_cloud_error:
  98                 error_count += 1
  99             self.assertEqual(error_count, status.tracker.get('cloud_errors'))
 100
 101     def test_destroy_node_only_cloud_errors_are_counted(self):
 102         status.tracker.update({'cloud_errors': 0})
 103         errors = [(config.CLOUD_ERRORS[0], True), (KeyError, False)]
 104         self.driver_mock().list_nodes.return_value = [testutil.MockSize(1)]
 105         driver = self.TestBaseComputeNodeDriver({}, {}, {}, self.driver_mock)
 106         error_count = 0
 107         for an_error, is_cloud_error in errors:
 108             self.driver_mock().destroy_node.side_effect = an_error
 109             with self.assertRaises(an_error):
 110                 driver.destroy_node(testutil.MockSize(1))
 111             if is_cloud_error:
 112                 error_count += 1
 113             self.assertEqual(error_count, status.tracker.get('cloud_errors'))