#!/usr/bin/env python
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
from __future__ import absolute_import, print_function
arv_node = testutil.arvados_node_mock(8)
driver = self.new_driver()
self.assertEqual('compute8.zzzzz.arvadosapi.com',
- driver.arvados_create_kwargs(arv_node)['name'])
+ driver.arvados_create_kwargs(testutil.MockSize(1), arv_node)['name'])
def test_default_hostname_from_new_arvados_node(self):
arv_node = testutil.arvados_node_mock(hostname=None)
driver = self.new_driver()
self.assertEqual('dynamic.compute.zzzzz.arvadosapi.com',
- driver.arvados_create_kwargs(arv_node)['name'])
+ driver.arvados_create_kwargs(testutil.MockSize(1), arv_node)['name'])
def check_node_tagged(self, cloud_node, expected_tags):
tag_mock = self.driver_mock().ex_create_tags
node.name = name
self.assertEqual(name, ec2.ComputeNodeDriver.node_fqdn(node))
- def test_cloud_exceptions(self):
- for error in [Exception("test exception"),
- IOError("test exception"),
- ssl.SSLError("test exception"),
- cloud_types.LibcloudError("test exception")]:
- self.assertTrue(ec2.ComputeNodeDriver.is_cloud_exception(error),
- "{} not flagged as cloud exception".format(error))
-
- def test_noncloud_exceptions(self):
- self.assertFalse(
- ec2.ComputeNodeDriver.is_cloud_exception(ValueError("test error")),
- "ValueError flagged as cloud exception")
+ def test_create_ebs_volume(self):
+ arv_node = testutil.arvados_node_mock()
+ driver = self.new_driver()
+ # libcloud/ec2 "disk" sizes are in GB, Arvados/SLURM "scratch" value is in MB
+ size = testutil.MockSize(1)
+ size.disk=5
+ size.scratch=20000
+ driver.create_node(size, arv_node)
+ create_method = self.driver_mock().create_node
+ self.assertTrue(create_method.called)
+ self.assertEqual([{
+ "DeviceName": "/dev/xvdt",
+ "Ebs": {
+ "DeleteOnTermination": True,
+ "VolumeSize": 16,
+ "VolumeType": "gp2"
+ }}],
+ create_method.call_args[1].get('ex_blockdevicemappings'))
+
+ def test_ebs_volume_not_needed(self):
+ arv_node = testutil.arvados_node_mock()
+ driver = self.new_driver()
+ # libcloud/ec2 "disk" sizes are in GB, Arvados/SLURM "scratch" value is in MB
+ size = testutil.MockSize(1)
+ size.disk=80
+ size.scratch=20000
+ driver.create_node(size, arv_node)
+ create_method = self.driver_mock().create_node
+ self.assertTrue(create_method.called)
+ self.assertIsNone(create_method.call_args[1].get('ex_blockdevicemappings'))
+
+ def test_ebs_volume_too_big(self):
+ arv_node = testutil.arvados_node_mock()
+ driver = self.new_driver()
+ # libcloud/ec2 "disk" sizes are in GB, Arvados/SLURM "scratch" value is in MB
+ size = testutil.MockSize(1)
+ size.disk=80
+ size.scratch=20000000
+ driver.create_node(size, arv_node)
+ create_method = self.driver_mock().create_node
+ self.assertTrue(create_method.called)
+ self.assertEqual([{
+ "DeviceName": "/dev/xvdt",
+ "Ebs": {
+ "DeleteOnTermination": True,
+ "VolumeSize": 16384,
+ "VolumeType": "gp2"
+ }}],
+ create_method.call_args[1].get('ex_blockdevicemappings'))