#!/usr/bin/python
#
# Copyright 2008 Google Inc. All Rights Reserved.
"""Test for host."""
# pylint: disable=missing-docstring
import sys
import unittest
import mock
import common
from autotest_lib.cli import cli_mock, host
from autotest_lib.client.common_lib import control_data
from autotest_lib.server import hosts
CLIENT = control_data.CONTROL_TYPE_NAMES.CLIENT
SERVER = control_data.CONTROL_TYPE_NAMES.SERVER
class host_ut(cli_mock.cli_unittest):
def test_parse_lock_options_both_set(self):
hh = host.host()
class opt(object):
lock = True
unlock = True
options = opt()
self.usage = "unused"
sys.exit.expect_call(1).and_raises(cli_mock.ExitException)
self.god.mock_io()
self.assertRaises(cli_mock.ExitException,
hh._parse_lock_options, options)
self.god.unmock_io()
def test_cleanup_labels_with_platform(self):
labels = ['l0', 'l1', 'l2', 'p0', 'l3']
hh = host.host()
self.assertEqual(['l0', 'l1', 'l2', 'l3'],
hh._cleanup_labels(labels, 'p0'))
def test_cleanup_labels_no_platform(self):
labels = ['l0', 'l1', 'l2', 'l3']
hh = host.host()
self.assertEqual(['l0', 'l1', 'l2', 'l3'],
hh._cleanup_labels(labels))
def test_cleanup_labels_with_non_avail_platform(self):
labels = ['l0', 'l1', 'l2', 'l3']
hh = host.host()
self.assertEqual(['l0', 'l1', 'l2', 'l3'],
hh._cleanup_labels(labels, 'p0'))
class host_list_unittest(cli_mock.cli_unittest):
def test_parse_host_not_required(self):
hl = host.host_list()
sys.argv = ['atest']
(options, leftover) = hl.parse()
self.assertEqual([], hl.hosts)
self.assertEqual([], leftover)
def test_parse_with_hosts(self):
hl = host.host_list()
mfile = cli_mock.create_file('host0\nhost3\nhost4\n')
sys.argv = ['atest', 'host1', '--mlist', mfile.name, 'host3']
(options, leftover) = hl.parse()
self.assertEqualNoOrder(['host0', 'host1','host3', 'host4'],
hl.hosts)
self.assertEqual(leftover, [])
mfile.clean()
def test_parse_with_labels(self):
hl = host.host_list()
sys.argv = ['atest', '--label', 'label0']
(options, leftover) = hl.parse()
self.assertEqual(['label0'], hl.labels)
self.assertEqual(leftover, [])
def test_parse_with_multi_labels(self):
hl = host.host_list()
sys.argv = ['atest', '--label', 'label0,label2']
(options, leftover) = hl.parse()
self.assertEqualNoOrder(['label0', 'label2'], hl.labels)
self.assertEqual(leftover, [])
def test_parse_with_escaped_commas_label(self):
hl = host.host_list()
sys.argv = ['atest', '--label', 'label\\,0']
(options, leftover) = hl.parse()
self.assertEqual(['label,0'], hl.labels)
self.assertEqual(leftover, [])
def test_parse_with_escaped_commas_multi_labels(self):
hl = host.host_list()
sys.argv = ['atest', '--label', 'label\\,0,label\\,2']
(options, leftover) = hl.parse()
self.assertEqualNoOrder(['label,0', 'label,2'], hl.labels)
self.assertEqual(leftover, [])
def test_parse_with_both(self):
hl = host.host_list()
mfile = cli_mock.create_file('host0\nhost3\nhost4\n')
sys.argv = ['atest', 'host1', '--mlist', mfile.name, 'host3',
'--label', 'label0']
(options, leftover) = hl.parse()
self.assertEqualNoOrder(['host0', 'host1','host3', 'host4'],
hl.hosts)
self.assertEqual(['label0'], hl.labels)
self.assertEqual(leftover, [])
mfile.clean()
def test_execute_list_all_no_labels(self):
self.run_cmd(argv=['atest', 'host', 'list'],
rpcs=[('get_hosts', {},
True,
[{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [],
u'invalid': False,
u'platform': None,
u'shard': None,
u'id': 1},
{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2}])],
out_words_ok=['host0', 'host1', 'Ready',
'plat1', 'False', 'True', 'None'])
def test_execute_list_all_with_labels(self):
self.run_cmd(argv=['atest', 'host', 'list'],
rpcs=[('get_hosts', {},
True,
[{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label0', u'label1'],
u'invalid': False,
u'platform': None,
u'shard': None,
u'id': 1},
{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 2}])],
out_words_ok=['host0', 'host1', 'Ready', 'plat1',
'label0', 'label1', 'label2', 'label3',
'False', 'True', 'None'])
def test_execute_list_filter_one_host(self):
self.run_cmd(argv=['atest', 'host', 'list', 'host1'],
rpcs=[('get_hosts', {'hostname__in': ['host1']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'True', 'None'],
out_words_no=['host0', 'host2',
'label1', 'False'])
def test_execute_list_filter_two_hosts(self):
mfile = cli_mock.create_file('host2')
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'--mlist', mfile.name],
# This is a bit fragile as the list order may change...
rpcs=[('get_hosts', {'hostname__in': ['host2', 'host1']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'True',
'host2', 'None'],
out_words_no=['host0', 'label1', 'False'])
mfile.clean()
def test_execute_list_filter_two_hosts_one_not_found(self):
mfile = cli_mock.create_file('host2')
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'--mlist', mfile.name],
# This is a bit fragile as the list order may change...
rpcs=[('get_hosts', {'hostname__in': ['host2', 'host1']},
True,
[{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['Ready', 'plat1',
'label3', 'True', 'None'],
out_words_no=['host1', 'False'],
err_words_ok=['host1'])
mfile.clean()
def test_execute_list_filter_two_hosts_none_found(self):
self.run_cmd(argv=['atest', 'host', 'list',
'host1', 'host2'],
# This is a bit fragile as the list order may change...
rpcs=[('get_hosts', {'hostname__in': ['host2', 'host1']},
True,
[])],
out_words_ok=[],
out_words_no=['Hostname', 'Status'],
err_words_ok=['Unknown', 'host1', 'host2'])
def test_execute_list_filter_label(self):
self.run_cmd(argv=['atest', 'host', 'list',
'-b', 'label3'],
rpcs=[('get_hosts', {'labels__name__in': ['label3']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'True',
'host2', 'None'],
out_words_no=['host0', 'label1', 'False'])
def test_execute_list_filter_multi_labels(self):
self.run_cmd(argv=['atest', 'host', 'list',
'-b', 'label3,label2'],
rpcs=[('get_hosts', {'multiple_labels': ['label2',
'label3']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat0'],
u'invalid': False,
u'shard': None,
u'platform': u'plat0',
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host3',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'label2', u'plat2'],
u'invalid': False,
u'shard': None,
u'platform': u'plat2',
u'id': 4}])],
out_words_ok=['host1', 'host3', 'Ready', 'plat0',
'label2', 'label3', 'plat2', 'None'],
out_words_no=['host2', 'False', 'plat1'])
def test_execute_list_filter_three_labels(self):
self.run_cmd(argv=['atest', 'host', 'list',
'-b', 'label3,label2'],
rpcs=[('get_hosts', {'multiple_labels': ['label2',
'label3']},
True,
[{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'label2', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 3}])],
out_words_ok=['host2', 'plat1',
'label2', 'label3', 'None'],
out_words_no=['host1', 'host3'])
def test_execute_list_filter_wild_labels(self):
self.run_cmd(argv=['atest', 'host', 'list',
'-b', 'label*'],
rpcs=[('get_hosts',
{'labels__name__startswith': 'label'},
True,
[{u'status': u'Ready',
u'hostname': u'host2',
u'locked': 1,
u'shard': None,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'label2', u'plat1'],
u'invalid': 0,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host2', 'plat1',
'label2', 'label3', 'None'],
out_words_no=['host1', 'host3'])
def test_execute_list_filter_multi_labels_no_results(self):
self.run_cmd(argv=['atest', 'host', 'list',
'-b', 'label3,label2, '],
rpcs=[('get_hosts', {'multiple_labels': ['label2',
'label3']},
True,
[])],
out_words_ok=[],
out_words_no=['host1', 'host2', 'host3',
'label2', 'label3'])
def test_execute_list_filter_label_and_hosts(self):
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'-b', 'label3', 'host2'],
rpcs=[('get_hosts', {'labels__name__in': ['label3'],
'hostname__in': ['host2', 'host1']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'labels': [u'label2', u'label3', u'plat1'],
u'lock_reason': u'',
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'True',
'host2', 'None'],
out_words_no=['host0', 'label1', 'False'])
def test_execute_list_filter_label_and_hosts_none(self):
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'-b', 'label3', 'host2'],
rpcs=[('get_hosts', {'labels__name__in': ['label3'],
'hostname__in': ['host2', 'host1']},
True,
[])],
out_words_ok=[],
out_words_no=['Hostname', 'Status'],
err_words_ok=['Unknown', 'host1', 'host2'])
def test_execute_list_filter_status(self):
self.run_cmd(argv=['atest', 'host', 'list',
'-s', 'Ready'],
rpcs=[('get_hosts', {'status__in': ['Ready']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'True',
'host2', 'None'],
out_words_no=['host0', 'label1', 'False'])
def test_execute_list_filter_status_and_hosts(self):
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'-s', 'Ready', 'host2'],
rpcs=[('get_hosts', {'status__in': ['Ready'],
'hostname__in': ['host2', 'host1']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'True',
'host2', 'None'],
out_words_no=['host0', 'label1', 'False'])
def test_execute_list_filter_status_and_hosts_none(self):
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'--status', 'Repair',
'host2'],
rpcs=[('get_hosts', {'status__in': ['Repair'],
'hostname__in': ['host2', 'host1']},
True,
[])],
out_words_ok=[],
out_words_no=['Hostname', 'Status'],
err_words_ok=['Unknown', 'host2'])
def test_execute_list_filter_statuses_and_hosts_none(self):
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'--status', 'Repair',
'host2'],
rpcs=[('get_hosts', {'status__in': ['Repair'],
'hostname__in': ['host2', 'host1']},
True,
[])],
out_words_ok=[],
out_words_no=['Hostname', 'Status'],
err_words_ok=['Unknown', 'host2'])
def test_execute_list_filter_locked(self):
self.run_cmd(argv=['atest', 'host', 'list', 'host1',
'--locked', 'host2'],
rpcs=[('get_hosts', {'locked': True,
'hostname__in': ['host2', 'host1']},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'locked_by': 'user0',
u'lock_reason': u'',
u'lock_time': u'2008-07-23 12:54:15',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'shard': None,
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host2',
u'locked': True,
u'locked_by': 'user0',
u'lock_reason': u'',
u'lock_time': u'2008-07-23 12:54:15',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'True',
'host2', 'None'],
out_words_no=['host0', 'label1', 'False'])
def test_execute_list_filter_unlocked(self):
self.run_cmd(argv=['atest', 'host', 'list',
'--unlocked'],
rpcs=[('get_hosts', {'locked': False},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label2', u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 2},
{u'status': u'Ready',
u'hostname': u'host2',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3}])],
out_words_ok=['host1', 'Ready', 'plat1',
'label2', 'label3', 'False',
'host2', 'None'],
out_words_no=['host0', 'label1', 'True'])
class host_stat_unittest(cli_mock.cli_unittest):
def test_execute_stat_two_hosts(self):
# The order of RPCs between host1 and host0 could change...
self.run_cmd(argv=['atest', 'host', 'stat', 'host0', 'host1'],
rpcs=[('get_hosts', {'hostname': 'host1'},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'lock_time': u'2008-07-23 12:54:15',
u'locked_by': 'user0',
u'lock_reason': u'',
u'protection': 'No protection',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'shard': None,
u'platform': u'plat1',
u'id': 3,
u'attributes': {}}]),
('get_hosts', {'hostname': 'host0'},
True,
[{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'protection': u'No protection',
u'labels': [u'label0', u'plat0'],
u'invalid': False,
u'shard': None,
u'platform': u'plat0',
u'id': 2,
u'attributes': {}}]),
('get_acl_groups', {'hosts__hostname': 'host1'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user2', u'debug_user', u'user0']}]),
('get_labels', {'host__hostname': 'host1'},
True,
[{u'id': 2,
u'platform': 1,
u'name': u'jme',
u'invalid': False,
u'kernel_config': u''}]),
('get_acl_groups', {'hosts__hostname': 'host0'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user0', u'debug_user']},
{u'description': u'myacl0',
u'hosts': [u'host0'],
u'id': 2,
u'name': u'acl0',
u'users': [u'user0']}]),
('get_labels', {'host__hostname': 'host0'},
True,
[{u'id': 4,
u'platform': 0,
u'name': u'label0',
u'invalid': False,
u'kernel_config': u''},
{u'id': 5,
u'platform': 1,
u'name': u'plat0',
u'invalid': False,
u'kernel_config': u''}])],
out_words_ok=['host0', 'host1', 'plat0', 'plat1',
'Everyone', 'acl0', 'label0'])
def test_execute_stat_one_bad_host_verbose(self):
self.run_cmd(argv=['atest', 'host', 'stat', 'host0',
'host1', '-v'],
rpcs=[('get_hosts', {'hostname': 'host1'},
True,
[]),
('get_hosts', {'hostname': 'host0'},
True,
[{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'protection': u'No protection',
u'labels': [u'label0', u'plat0'],
u'invalid': False,
u'platform': u'plat0',
u'id': 2,
u'attributes': {}}]),
('get_acl_groups', {'hosts__hostname': 'host0'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user0', u'debug_user']},
{u'description': u'myacl0',
u'hosts': [u'host0'],
u'id': 2,
u'name': u'acl0',
u'users': [u'user0']}]),
('get_labels', {'host__hostname': 'host0'},
True,
[{u'id': 4,
u'platform': 0,
u'name': u'label0',
u'invalid': False,
u'kernel_config': u''},
{u'id': 5,
u'platform': 1,
u'name': u'plat0',
u'invalid': False,
u'kernel_config': u''}])],
out_words_ok=['host0', 'plat0',
'Everyone', 'acl0', 'label0'],
out_words_no=['host1'],
err_words_ok=['host1', 'Unknown host'],
err_words_no=['host0'])
def test_execute_stat_one_bad_host(self):
self.run_cmd(argv=['atest', 'host', 'stat', 'host0', 'host1'],
rpcs=[('get_hosts', {'hostname': 'host1'},
True,
[]),
('get_hosts', {'hostname': 'host0'},
True,
[{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'protection': u'No protection',
u'labels': [u'label0', u'plat0'],
u'invalid': False,
u'platform': u'plat0',
u'id': 2,
u'attributes': {}}]),
('get_acl_groups', {'hosts__hostname': 'host0'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user0', u'debug_user']},
{u'description': u'myacl0',
u'hosts': [u'host0'],
u'id': 2,
u'name': u'acl0',
u'users': [u'user0']}]),
('get_labels', {'host__hostname': 'host0'},
True,
[{u'id': 4,
u'platform': 0,
u'name': u'label0',
u'invalid': False,
u'kernel_config': u''},
{u'id': 5,
u'platform': 1,
u'name': u'plat0',
u'invalid': False,
u'kernel_config': u''}])],
out_words_ok=['host0', 'plat0',
'Everyone', 'acl0', 'label0'],
out_words_no=['host1'],
err_words_ok=['host1', 'Unknown host'],
err_words_no=['host0'])
def test_execute_stat_wildcard(self):
# The order of RPCs between host1 and host0 could change...
self.run_cmd(argv=['atest', 'host', 'stat', 'ho*'],
rpcs=[('get_hosts', {'hostname__startswith': 'ho'},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'lock_time': u'2008-07-23 12:54:15',
u'locked_by': 'user0',
u'lock_reason': u'',
u'protection': 'No protection',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'id': 3,
u'attributes': {}},
{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'protection': u'No protection',
u'labels': [u'label0', u'plat0'],
u'invalid': False,
u'platform': u'plat0',
u'id': 2,
u'attributes': {}}]),
('get_acl_groups', {'hosts__hostname': 'host1'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user2', u'debug_user', u'user0']}]),
('get_labels', {'host__hostname': 'host1'},
True,
[{u'id': 2,
u'platform': 1,
u'name': u'jme',
u'invalid': False,
u'kernel_config': u''}]),
('get_acl_groups', {'hosts__hostname': 'host0'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user0', u'debug_user']},
{u'description': u'myacl0',
u'hosts': [u'host0'],
u'id': 2,
u'name': u'acl0',
u'users': [u'user0']}]),
('get_labels', {'host__hostname': 'host0'},
True,
[{u'id': 4,
u'platform': 0,
u'name': u'label0',
u'invalid': False,
u'kernel_config': u''},
{u'id': 5,
u'platform': 1,
u'name': u'plat0',
u'invalid': False,
u'kernel_config': u''}])],
out_words_ok=['host0', 'host1', 'plat0', 'plat1',
'Everyone', 'acl0', 'label0'])
def test_execute_stat_wildcard_and_host(self):
# The order of RPCs between host1 and host0 could change...
self.run_cmd(argv=['atest', 'host', 'stat', 'ho*', 'newhost0'],
rpcs=[('get_hosts', {'hostname': 'newhost0'},
True,
[{u'status': u'Ready',
u'hostname': u'newhost0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'lock_reason': u'',
u'protection': u'No protection',
u'labels': [u'label0', u'plat0'],
u'invalid': False,
u'platform': u'plat0',
u'id': 5,
u'attributes': {}}]),
('get_hosts', {'hostname__startswith': 'ho'},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'lock_time': u'2008-07-23 12:54:15',
u'locked_by': 'user0',
u'lock_reason': u'',
u'protection': 'No protection',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'id': 3,
u'attributes': {}},
{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_reason': u'',
u'protection': 'No protection',
u'lock_time': u'2008-07-23 12:54:15',
u'labels': [u'label0', u'plat0'],
u'invalid': False,
u'platform': u'plat0',
u'id': 2,
u'attributes': {}}]),
('get_acl_groups', {'hosts__hostname': 'newhost0'},
True,
[{u'description': u'',
u'hosts': [u'newhost0', 'host1'],
u'id': 42,
u'name': u'my_acl',
u'users': [u'user0', u'debug_user']},
{u'description': u'my favorite acl',
u'hosts': [u'newhost0'],
u'id': 2,
u'name': u'acl10',
u'users': [u'user0']}]),
('get_labels', {'host__hostname': 'newhost0'},
True,
[{u'id': 4,
u'platform': 0,
u'name': u'label0',
u'invalid': False,
u'kernel_config': u''},
{u'id': 5,
u'platform': 1,
u'name': u'plat0',
u'invalid': False,
u'kernel_config': u''}]),
('get_acl_groups', {'hosts__hostname': 'host1'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user2', u'debug_user', u'user0']}]),
('get_labels', {'host__hostname': 'host1'},
True,
[{u'id': 2,
u'platform': 1,
u'name': u'jme',
u'invalid': False,
u'kernel_config': u''}]),
('get_acl_groups', {'hosts__hostname': 'host0'},
True,
[{u'description': u'',
u'hosts': [u'host0', u'host1'],
u'id': 1,
u'name': u'Everyone',
u'users': [u'user0', u'debug_user']},
{u'description': u'myacl0',
u'hosts': [u'host0'],
u'id': 2,
u'name': u'acl0',
u'users': [u'user0']}]),
('get_labels', {'host__hostname': 'host0'},
True,
[{u'id': 4,
u'platform': 0,
u'name': u'label0',
u'invalid': False,
u'kernel_config': u''},
{u'id': 5,
u'platform': 1,
u'name': u'plat0',
u'invalid': False,
u'kernel_config': u''}])],
out_words_ok=['host0', 'host1', 'newhost0',
'plat0', 'plat1',
'Everyone', 'acl10', 'label0'])
class host_jobs_unittest(cli_mock.cli_unittest):
def test_execute_jobs_one_host(self):
self.run_cmd(argv=['atest', 'host', 'jobs', 'host0'],
rpcs=[('get_host_queue_entries',
{'host__hostname': 'host0', 'query_limit': 20,
'sort_by': ['-job__id']},
True,
[{u'status': u'Failed',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host0',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': u'meta0',
u'job': {u'control_file':
(u"def step_init():\n"
"\tjob.next_step([step_test])\n"
"def step_test():\n"
"\tjob.run_test('kernbench')\n\n"),
u'name': u'kernel-smp-2.6.xyz.x86_64',
u'control_type': CLIENT,
u'synchronizing': None,
u'priority': u'Low',
u'owner': u'user0',
u'created_on': u'2008-01-09 10:45:12',
u'synch_count': None,
u'id': 216},
u'active': 0,
u'id': 2981},
{u'status': u'Aborted',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host0',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': None,
u'job': {u'control_file':
u"job.run_test('sleeptest')\n\n",
u'name': u'testjob',
u'control_type': CLIENT,
u'synchronizing': 0,
u'priority': u'Low',
u'owner': u'user1',
u'created_on': u'2008-01-17 15:04:53',
u'synch_count': None,
u'id': 289},
u'active': 0,
u'id': 3167}])],
out_words_ok=['216', 'user0', 'Failed',
'kernel-smp-2.6.xyz.x86_64', 'Aborted',
'289', 'user1', 'Aborted',
'testjob'])
def test_execute_jobs_wildcard(self):
self.run_cmd(argv=['atest', 'host', 'jobs', 'ho*'],
rpcs=[('get_hosts', {'hostname__startswith': 'ho'},
True,
[{u'status': u'Ready',
u'hostname': u'host1',
u'locked': True,
u'lock_time': u'2008-07-23 12:54:15',
u'locked_by': 'user0',
u'labels': [u'label3', u'plat1'],
u'invalid': False,
u'platform': u'plat1',
u'id': 3},
{u'status': u'Ready',
u'hostname': u'host0',
u'locked': False,
u'locked_by': 'user0',
u'lock_time': u'2008-07-23 12:54:15',
u'labels': [u'label0', u'plat0'],
u'invalid': False,
u'platform': u'plat0',
u'id': 2}]),
('get_host_queue_entries',
{'host__hostname': 'host1', 'query_limit': 20,
'sort_by': ['-job__id']},
True,
[{u'status': u'Failed',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host1',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': u'meta0',
u'job': {u'control_file':
(u"def step_init():\n"
"\tjob.next_step([step_test])\n"
"def step_test():\n"
"\tjob.run_test('kernbench')\n\n"),
u'name': u'kernel-smp-2.6.xyz.x86_64',
u'control_type': CLIENT,
u'synchronizing': None,
u'priority': u'Low',
u'owner': u'user0',
u'created_on': u'2008-01-09 10:45:12',
u'synch_count': None,
u'id': 216},
u'active': 0,
u'id': 2981},
{u'status': u'Aborted',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host1',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': None,
u'job': {u'control_file':
u"job.run_test('sleeptest')\n\n",
u'name': u'testjob',
u'control_type': CLIENT,
u'synchronizing': 0,
u'priority': u'Low',
u'owner': u'user1',
u'created_on': u'2008-01-17 15:04:53',
u'synch_count': None,
u'id': 289},
u'active': 0,
u'id': 3167}]),
('get_host_queue_entries',
{'host__hostname': 'host0', 'query_limit': 20,
'sort_by': ['-job__id']},
True,
[{u'status': u'Failed',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host0',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': u'meta0',
u'job': {u'control_file':
(u"def step_init():\n"
"\tjob.next_step([step_test])\n"
"def step_test():\n"
"\tjob.run_test('kernbench')\n\n"),
u'name': u'kernel-smp-2.6.xyz.x86_64',
u'control_type': CLIENT,
u'synchronizing': None,
u'priority': u'Low',
u'owner': u'user0',
u'created_on': u'2008-01-09 10:45:12',
u'synch_count': None,
u'id': 216},
u'active': 0,
u'id': 2981},
{u'status': u'Aborted',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host0',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': None,
u'job': {u'control_file':
u"job.run_test('sleeptest')\n\n",
u'name': u'testjob',
u'control_type': CLIENT,
u'synchronizing': 0,
u'priority': u'Low',
u'owner': u'user1',
u'created_on': u'2008-01-17 15:04:53',
u'synch_count': None,
u'id': 289},
u'active': 0,
u'id': 3167}])],
out_words_ok=['216', 'user0', 'Failed',
'kernel-smp-2.6.xyz.x86_64', 'Aborted',
'289', 'user1', 'Aborted',
'testjob'])
def test_execute_jobs_one_host_limit(self):
self.run_cmd(argv=['atest', 'host', 'jobs', 'host0', '-q', '10'],
rpcs=[('get_host_queue_entries',
{'host__hostname': 'host0', 'query_limit': 10,
'sort_by': ['-job__id']},
True,
[{u'status': u'Failed',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host0',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': u'meta0',
u'job': {u'control_file':
(u"def step_init():\n"
"\tjob.next_step([step_test])\n"
"def step_test():\n"
"\tjob.run_test('kernbench')\n\n"),
u'name': u'kernel-smp-2.6.xyz.x86_64',
u'control_type': CLIENT,
u'synchronizing': None,
u'priority': u'Low',
u'owner': u'user0',
u'created_on': u'2008-01-09 10:45:12',
u'synch_count': None,
u'id': 216},
u'active': 0,
u'id': 2981},
{u'status': u'Aborted',
u'complete': 1,
u'host': {u'status': u'Ready',
u'locked': True,
u'locked_by': 'user0',
u'hostname': u'host0',
u'invalid': False,
u'id': 3232},
u'priority': 0,
u'meta_host': None,
u'job': {u'control_file':
u"job.run_test('sleeptest')\n\n",
u'name': u'testjob',
u'control_type': CLIENT,
u'synchronizing': 0,
u'priority': u'Low',
u'owner': u'user1',
u'created_on': u'2008-01-17 15:04:53',
u'synch_count': None,
u'id': 289},
u'active': 0,
u'id': 3167}])],
out_words_ok=['216', 'user0', 'Failed',
'kernel-smp-2.6.xyz.x86_64', 'Aborted',
'289', 'user1', 'Aborted',
'testjob'])
class host_mod_create_tests(object):
def _gen_attributes_rpcs(self, host, attributes):
"""Generate RPCs expected to add attributes to host.
@param host: hostname
@param attributes: dict of attributes
@return: list of rpcs to expect
"""
rpcs = []
for attr, val in attributes.iteritems():
rpcs.append(('set_host_attribute',
{
'hostname': host,
'attribute': attr,
'value': val,
},
True, None))
return rpcs
def _gen_labels_rpcs(self, labels, platform=False, host_id=None):
"""Generate RPCS expected to add labels.
@param labels: list of label names
@param platform: labels are platform labels
@param host_id: Host id old labels will be deleted from (if host exists)
"""
rpcs = []
if host_id:
rpcs.append(('get_labels', {'host': host_id}, True, []))
for label in labels:
rpcs += [
('get_labels', {'name': label}, True, []),
('add_label', {'name': label}, True, None)
]
if platform:
rpcs[-1][1]['platform'] = True
return rpcs
def _gen_acls_rpcs(self, hosts, acls, host_ids=[]):
"""Generate RPCs expected to add acls.
@param hosts: list of hostnames
@param acls: list of acl names
@param host_ids: List of host_ids if hosts already exist
"""
rpcs = []
for host_id in host_ids:
rpcs.append(('get_acl_groups', {'hosts': host_id}, True, []))
for acl in acls:
rpcs.append(('get_acl_groups', {'name': acl}, True, []))
rpcs.append(('add_acl_group', {'name': acl}, True, None))
for acl in acls:
rpcs.append((
'acl_group_add_hosts',
{
'hosts': hosts,
'id': acl,
},
True,
None,
))
return rpcs
def test_lock_one_host(self):
"""Test locking host / creating host locked."""
lock_reason = 'Because'
rpcs, out = self._gen_expectations(locked=True, lock_reason=lock_reason)
self.run_cmd(argv=self._command_single + ['--lock', '--lock_reason',
lock_reason],
rpcs=rpcs, out_words_ok=out)
def test_unlock_multiple_hosts(self):
"""Test unlocking host / creating host unlocked."""
rpcs, out = self._gen_expectations(hosts=self._hosts, locked=False)
self.run_cmd(argv=self._command_multiple + ['--unlock'], rpcs=rpcs,
out_words_ok=out)
def test_machine_list(self):
"""Test action an machines from machine list file."""
mfile = cli_mock.create_file(','.join(self._hosts))
rpcs, out = self._gen_expectations(hosts=self._hosts, locked=False)
try:
self.run_cmd(argv=self._command_multiple + ['--unlock'], rpcs=rpcs,
out_words_ok=out)
finally:
mfile.clean()
def test_single_attributes(self):
"""Test applying one attribute to one host."""
attrs = {'foo': 'bar'}
rpcs, out = self._gen_expectations(attributes=attrs)
self.run_cmd(self._command_single + ['--attribute', 'foo=bar'],
rpcs=rpcs, out_words_ok=out)
def test_attributes_comma(self):
"""Test setting an attribute with a comma in the value."""
attrs = {'foo': 'bar,zip'}
rpcs, out = self._gen_expectations(attributes=attrs)
self.run_cmd(self._command_single + ['--attribute', 'foo=bar,zip'],
rpcs=rpcs, out_words_ok=out)
def test_multiple_attributes_comma(self):
"""Test setting attributes when one of the values contains a comma."""
attrs = {'foo': 'bar,zip', 'zang': 'poodle'}
rpcs, out = self._gen_expectations(attributes=attrs)
self.run_cmd(self._command_single + ['--attribute', 'foo=bar,zip',
'--attribute', 'zang=poodle'],
rpcs=rpcs, out_words_ok=out)
def test_multiple_attributes_multiple_hosts(self):
"""Test applying multiple attributes to multiple hosts."""
attrs = {'foo': 'bar', 'baz': 'zip'}
rpcs, out = self._gen_expectations(hosts=self._hosts, attributes=attrs)
self.run_cmd(self._command_multiple + ['--attribute', 'foo=bar',
'--attribute', 'baz=zip'],
rpcs=rpcs, out_words_ok=out)
def test_platform(self):
"""Test applying platform label."""
rpcs, out = self._gen_expectations(platform='some_platform')
self.run_cmd(argv=self._command_single + ['--platform',
'some_platform'],
rpcs=rpcs, out_words_ok=out)
def test_labels(self):
"""Test applying labels."""
labels = ['label0', 'label1']
rpcs, out = self._gen_expectations(labels=labels)
self.run_cmd(argv=self._command_single + ['--labels', ','.join(labels)],
rpcs=rpcs, out_words_ok=out)
def test_labels_from_file(self):
"""Test applying labels from file."""
labels = ['label0', 'label1']
rpcs, out = self._gen_expectations(labels=labels)
labelsf = cli_mock.create_file(','.join(labels))
try:
self.run_cmd(argv=self._command_single + ['--blist', labelsf.name],
rpcs=rpcs, out_words_ok=out)
finally:
labelsf.clean()
def test_acls(self):
"""Test applying acls."""
acls = ['acl0', 'acl1']
rpcs, out = self._gen_expectations(acls=acls)
self.run_cmd(argv=self._command_single + ['--acls', ','.join(acls)],
rpcs=rpcs, out_words_ok=out)
def test_acls_from_file(self):
"""Test applying acls from file."""
acls = ['acl0', 'acl1']
rpcs, out = self._gen_expectations(acls=acls)
aclsf = cli_mock.create_file(','.join(acls))
try:
self.run_cmd(argv=self._command_single + ['-A', aclsf.name],
rpcs=rpcs, out_words_ok=out)
finally:
aclsf.clean()
def test_protection(self):
"""Test applying host protection."""
protection = 'Do not repair'
rpcs, out = self._gen_expectations(protection=protection)
self.run_cmd(argv=self._command_single + ['--protection', protection],
rpcs=rpcs,out_words_ok=out)
def test_protection_invalid(self):
"""Test invalid protection causes failure."""
protection = 'Invalid protection'
rpcs, out = self._gen_expectations(hosts=[])
self.run_cmd(argv=self._command_single + ['--protection', protection],
exit_code=2, err_words_ok=['invalid', 'choice'] +
protection.split())
def test_complex(self):
"""Test applying multiple modifications / creating a complex host."""
lock_reason = 'Because I said so.'
platform = 'some_platform'
labels = ['label0', 'label1']
acls = ['acl0', 'acl1']
protection = 'Do not verify'
labelsf = cli_mock.create_file(labels[1])
aclsf = cli_mock.create_file(acls[1])
cmd_args = ['-l', '-r', lock_reason, '-t', platform, '-b', labels[0],
'-B', labelsf.name, '-a', acls[0], '-A', aclsf.name, '-p',
protection]
rpcs, out = self._gen_expectations(locked=True, lock_reason=lock_reason,
acls=acls, labels=labels,
platform=platform,
protection=protection)
try:
self.run_cmd(argv=self._command_single + cmd_args, rpcs=rpcs,
out_words_ok=out)
finally:
labelsf.clean()
aclsf.clean()
class host_mod_unittest(host_mod_create_tests, cli_mock.cli_unittest):
"""Tests specific to the mod action and expectation generator for shared
tests.
"""
_hosts = ['localhost', '127.0.0.1']
_host_ids = [1, 2]
_command_base = ['atest', 'host', 'mod']
_command_single = _command_base + [_hosts[0]]
_command_multiple = _command_base + _hosts
def _gen_expectations(self, hosts=['localhost'], locked=None,
lock_reason='', force_lock=False, protection=None,
acls=[], labels=[], platform=None, attributes={}):
rpcs = []
out = set()
hosts = hosts[:]
hosts.reverse()
# Genarate result for get_hosts command to include all known hosts
host_dicts = []
for h, h_id in zip(self._hosts, self._host_ids):
host_dicts.append({'hostname': h, 'id': h_id})
rpcs.append(('get_hosts', {'hostname__in': hosts}, True, host_dicts))
# Expect actions only for known hosts
host_ids = []
for host in hosts:
if host not in self._hosts:
continue
host_id = self._host_ids[self._hosts.index(host)]
host_ids.append(host_id)
modify_args = {'id': host}
if locked is not None:
out.add('Locked' if locked else 'Unlocked')
modify_args['locked'] = locked
modify_args['lock_reason'] = lock_reason
if force_lock:
modify_args['force_modify_locking'] = True
if protection:
modify_args['protection'] = protection
if len(modify_args.keys()) > 1:
out.add(host)
rpcs.append(('modify_host', modify_args, True, None))
if labels:
rpcs += self._gen_labels_rpcs(labels, host_id=host_id)
rpcs.append(('host_add_labels', {'id': host, 'labels': labels},
True, None))
if platform:
rpcs += self._gen_labels_rpcs([platform], platform=True,
host_id=host_id)
rpcs.append(('host_add_labels', {'id': host,
'labels': [platform]},
True, None))
rpcs += self._gen_attributes_rpcs(host, attributes)
if acls:
rpcs += self._gen_acls_rpcs(hosts, acls, host_ids=host_ids)
return rpcs, list(out)
def test_mod_force_lock_one_host(self):
"""Test mod with forced locking."""
lock_reason = 'Because'
rpcs, out = self._gen_expectations(locked=True, force_lock=True,
lock_reason=lock_reason)
self.run_cmd(argv=self._command_single + [
'--lock', '--force_modify_locking', '--lock_reason',
lock_reason],
rpcs=rpcs, out_words_ok=out)
def test_mod_force_unlock_one_host(self):
"""Test mod forced unlocking."""
rpcs, out = self._gen_expectations(locked=False, force_lock=True)
self.run_cmd(argv=self._command_single + ['--unlock',
'--force_modify_locking'],
rpcs=rpcs, out_words_ok=out)
def test_mod_fail_unknown_host(self):
"""Test mod fails with unknown host."""
rpcs, out = self._gen_expectations(hosts=['nope'], locked=True)
self.run_cmd(argv=self._command_base + ['nope', '--lock'],
rpcs=rpcs, err_words_ok=['Cannot', 'modify', 'nope'])
class host_create_unittest(host_mod_create_tests, cli_mock.cli_unittest):
"""Test specific to create action and expectation generator for shared
tests.
"""
_hosts = ['localhost', '127.0.0.1']
_command_base = ['atest', 'host', 'create']
_command_single = _command_base + [_hosts[0]]
_command_multiple = _command_base + _hosts
def setUp(self):
"""Mock out the create_host method.
"""
super(host_create_unittest, self).setUp()
self._orig_create_host = hosts.create_host
def tearDown(self):
"""Undo mock.
"""
super(host_create_unittest, self).tearDown()
hosts.create_host = self._orig_create_host
def _mock_host(self, platform=None, labels=[]):
"""Update the return values of the mocked host object.
@param platform: return value of Host.get_platform()
@param labels: return value of Host.get_labels()
"""
mock_host = mock.MagicMock()
mock_host.get_platform.return_value = platform
mock_host.get_labels.return_value = labels
hosts.create_host = mock.MagicMock()
hosts.create_host.return_value = mock_host
def _gen_expectations(self, hosts=['localhost'], locked=False,
lock_reason=None, platform=None,
discovered_platform=None, labels=[],
discovered_labels=[], acls=[], protection=None,
attributes={}):
"""Build a list of expected RPC calls based on values to host command.
@param hosts: list of hostname being created (default ['localhost'])
@param locked: end state of host (bool)
@param lock_reason: reason for host to be locked
@param platform: platform label
@param discovered_platform: platform discovered automatically by host
@param labels: list of host labels (excluding platform)
@param discovered_labels: list of labels discovered automatically
@param acls: list of host acls
@param protection: host protection level
@return: list of expect rpc calls (each call is (op, args, success,
result))
"""
hosts = hosts[:]
hosts.reverse() # No idea why
lock_reason = lock_reason or 'Forced lock on device creation'
acls = acls or []
rpcs = []
out = ['Added', 'host'] + hosts
# Mock platform and label detection results
self._mock_host(discovered_platform, discovered_labels)
for host in hosts:
add_args = {
'hostname': host,
'status': 'Ready',
'locked': True,
'lock_reason': lock_reason,
}
if protection:
add_args['protection'] = protection
rpcs.append(('add_host', add_args, True, None))
rpcs += self._gen_labels_rpcs(labels)
if labels:
rpcs.append(('host_add_labels', {'id': host, 'labels': labels},
True, None))
if platform:
rpcs += self._gen_labels_rpcs([platform], platform=True)
rpcs.append(('host_add_labels', {'id': host,
'labels': [platform]},
True, None))
rpcs += self._gen_attributes_rpcs(host, attributes)
rpcs += self._gen_acls_rpcs(hosts, acls)
if not locked:
for host in hosts:
rpcs.append((
'modify_host',
{
'id': host,
'locked': False,
'lock_reason': '',
},
True,
None,
))
return rpcs, out
def test_create_no_args(self):
"""Test simple creation with to arguments."""
rpcs, out = self._gen_expectations()
self.run_cmd(argv=self._command_single, rpcs=rpcs, out_words_ok=out)
def test_create_with_discovered_platform(self):
"""Test discovered platform is used when platform isn't specified."""
rpcs, out = self._gen_expectations(platform='some_platform',
discovered_platform='some_platform')
self.run_cmd(argv=self._command_single, rpcs=rpcs, out_words_ok=out)
def test_create_specified_platform_overrides_discovered_platform(self):
"""Test that the specified platform overrides the discovered platform.
"""
rpcs, out = self._gen_expectations(platform='some_platform',
discovered_platform='wrong_platform')
self.run_cmd(argv=self._command_single + ['--platform',
'some_platform'],
rpcs=rpcs, out_words_ok=out)
def test_create_discovered_labels(self):
"""Test applying automatically discovered labels."""
labels = ['label0', 'label1']
rpcs, out = self._gen_expectations(labels=labels,
discovered_labels=labels)
self.run_cmd(argv=self._command_single, rpcs=rpcs, out_words_ok=out)
def test_create_specified_discovered_labels_combine(self):
"""Test applying both discovered and specified labels."""
labels = ['label0', 'label1']
rpcs, out = self._gen_expectations(labels=labels,
discovered_labels=[labels[0]])
self.run_cmd(argv=self._command_single + ['--labels', labels[1]],
rpcs=rpcs, out_words_ok=out)
if __name__ == '__main__':
unittest.main()