# pylint: disable-msg=C0111 #!/usr/bin/python # # Copyright 2008 Google Inc. All Rights Reserved. """Tests for action_common.""" import unittest, sys, copy import common from autotest_lib.cli import cli_mock, action_common, rpc # # List action # class atest_list_unittest(cli_mock.cli_unittest): def test_check_for_wilcard_none(self): orig_filters = {'name__in': ['item0', 'item1']} orig_checks = {'name__in': ['item0', 'item1']} mytest = action_common.atest_list() filters = copy.deepcopy(orig_filters) checks = copy.deepcopy(orig_checks) mytest.check_for_wildcard(filters, checks) self.assertEqual(filters, orig_filters) self.assertEqual(checks, orig_checks) def test_check_for_wilcard_none_list(self): orig_filters = {'name__in': ['item0']} orig_checks = {'name__in': ['item0']} mytest = action_common.atest_list() filters = copy.deepcopy(orig_filters) checks = copy.deepcopy(orig_checks) mytest.check_for_wildcard(filters, checks) self.assertEqual(filters, orig_filters) self.assertEqual(checks, orig_checks) def test_check_for_wilcard_one_list(self): filters = {'something__in': ['item*']} checks = {'something__in': ['item*']} mytest = action_common.atest_list() mytest.check_for_wildcard(filters, checks) self.assertEqual(filters, {'something__startswith': 'item'}) self.assertEqual(checks, {'something__startswith': None}) def test_check_for_wilcard_one_string(self): filters = {'something__name': 'item*'} checks = {'something__name': 'item*'} mytest = action_common.atest_list() mytest.check_for_wildcard(filters, checks) self.assertEqual(filters, {'something__name__startswith': 'item'}) self.assertEqual(checks, {'something__name__startswith': None}) def test_check_for_wilcard_one_string_login(self): filters = {'something__login': 'item*'} checks = {'something__login': 'item*'} mytest = action_common.atest_list() mytest.check_for_wildcard(filters, checks) self.assertEqual(filters, {'something__login__startswith': 'item'}) self.assertEqual(checks, {'something__login__startswith': None}) def test_check_for_wilcard_two(self): orig_filters = {'something__in': ['item0*', 'item1*']} orig_checks = {'something__in': ['item0*', 'item1*']} mytest = action_common.atest_list() filters = copy.deepcopy(orig_filters) checks = copy.deepcopy(orig_checks) self.god.stub_function(sys, 'exit') sys.exit.expect_call(1).and_raises(cli_mock.ExitException) self.god.mock_io() self.assertRaises(cli_mock.ExitException, mytest.check_for_wildcard, filters, checks) (out, err) = self.god.unmock_io() self.god.check_playback() self.assertEqual(filters, orig_filters) self.assertEqual(checks, orig_checks) def _atest_list_execute(self, filters={}, check_results={}): values = [{u'id': 180, u'platform': 0, u'name': u'label0', u'invalid': 0, u'kernel_config': u''}, {u'id': 338, u'platform': 0, u'name': u'label1', u'invalid': 0, u'kernel_config': u''}] mytest = action_common.atest_list() mytest.afe = rpc.afe_comm() self.mock_rpcs([('get_labels', filters, True, values)]) self.god.mock_io() self.assertEqual(values, mytest.execute(op='get_labels', filters=filters, check_results=check_results)) (out, err) = self.god.unmock_io() self.god.check_playback() return (out, err) def test_atest_list_execute_no_filters(self): self._atest_list_execute() def test_atest_list_execute_filters_all_good(self): filters = {} check_results = {} filters['name__in'] = ['label0', 'label1'] check_results['name__in'] = 'name' (out, err) = self._atest_list_execute(filters, check_results) self.assertEqual(err, '') def test_atest_list_execute_filters_good_and_bad(self): filters = {} check_results = {} filters['name__in'] = ['label0', 'label1', 'label2'] check_results['name__in'] = 'name' (out, err) = self._atest_list_execute(filters, check_results) self.assertWords(err, ['Unknown', 'label2']) def test_atest_list_execute_items_good_and_bad_no_check(self): filters = {} check_results = {} filters['name__in'] = ['label0', 'label1', 'label2'] check_results['name__in'] = None (out, err) = self._atest_list_execute(filters, check_results) self.assertEqual(err, '') def test_atest_list_execute_filters_wildcard(self): filters = {} check_results = {} filters['name__in'] = ['label*'] check_results['name__in'] = 'name' values = [{u'id': 180, u'platform': False, u'name': u'label0', u'invalid': False, u'kernel_config': u''}, {u'id': 338, u'platform': False, u'name': u'label1', u'invalid': False, u'kernel_config': u''}] mytest = action_common.atest_list() mytest.afe = rpc.afe_comm() self.mock_rpcs([('get_labels', {'name__startswith': 'label'}, True, values)]) self.god.mock_io() self.assertEqual(values, mytest.execute(op='get_labels', filters=filters, check_results=check_results)) (out, err) = self.god.unmock_io() self.god.check_playback() self.assertEqual(err, '') # # Creation & Deletion of a topic (ACL, label, user) # class atest_create_or_delete_unittest(cli_mock.cli_unittest): def _create_cr_del(self, items): def _items(): return items crdel = action_common.atest_create_or_delete() crdel.afe = rpc.afe_comm() crdel.topic = crdel.usage_topic = 'label' crdel.op_action = 'add' crdel.get_items = _items crdel.data['platform'] = False crdel.data_item_key = 'name' crdel.no_confirmation = True return crdel def test_execute_create_one_topic(self): acr = self._create_cr_del(['label0']) self.mock_rpcs([('add_label', {'name': 'label0', 'platform': False}, True, 42)]) ret = acr.execute() self.god.check_playback() self.assert_(['label0'], ret) def test_execute_create_two_topics(self): acr = self._create_cr_del(['label0', 'label1']) self.mock_rpcs([('add_label', {'name': 'label0', 'platform': False}, True, 42), ('add_label', {'name': 'label1', 'platform': False}, True, 43)]) ret = acr.execute() self.god.check_playback() self.assertEqualNoOrder(['label0', 'label1'], ret) def test_execute_create_error(self): acr = self._create_cr_del(['label0']) self.mock_rpcs([('add_label', {'name': 'label0', 'platform': False}, False, '''ValidationError: {'name': 'This value must be unique (label0)'}''')]) ret = acr.execute() self.god.check_playback() self.assertEqualNoOrder([], ret) # # Adding or Removing users or hosts from a topic(ACL or label) # class atest_add_or_remove_unittest(cli_mock.cli_unittest): def _create_add_remove(self, items, users=None, hosts=None): def _items(): return [items] addrm = action_common.atest_add_or_remove() addrm.afe = rpc.afe_comm() if users: addrm.users = users if hosts: addrm.hosts = hosts addrm.topic = 'acl_group' addrm.msg_topic = 'ACL' addrm.op_action = 'add' addrm.msg_done = 'Added to' addrm.get_items = _items return addrm def test__add_remove_uh_to_topic(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, True, None)]) acl_addrm._add_remove_uh_to_topic('acl0', 'users') self.god.check_playback() def test__add_remove_uh_to_topic_raise(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1']) self.assertRaises(AttributeError, acl_addrm._add_remove_uh_to_topic, 'acl0', 'hosts') def test_execute_add_or_remove_uh_to_topic_acl_users(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, True, None)]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqualNoOrder(['acl0'], execute_result['users']) self.assertEqual([], execute_result['hosts']) def test_execute_add_or_remove_uh_to_topic_acl_users_hosts(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1'], hosts=['host0', 'host1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, True, None), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0', 'host1']}, True, None)]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqualNoOrder(['acl0'], execute_result['users']) self.assertEqualNoOrder(['acl0'], execute_result['hosts']) def test_execute_add_or_remove_uh_to_topic_acl_bad_users(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, False, 'DoesNotExist: The following users do not exist: ' 'user0, user1')]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual([], execute_result['users']) self.assertEqual([], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, err_words_ok=['DoesNotExist', 'acl_group_add_users', 'user0', 'user1'], err_words_no = ['acl_group_add_hosts']) def test_execute_add_or_remove_uh_to_topic_acl_bad_users_partial(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, False, 'DoesNotExist: The following users do not exist: ' 'user0'), ('acl_group_add_users', {'id': 'acl0', 'users': ['user1']}, True, None)]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual(['acl0'], execute_result['users']) self.assertEqual([], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, out_words_ok=["Added to ACL 'acl0'", 'user1'], err_words_ok=['DoesNotExist', 'acl_group_add_users', 'user0'], err_words_no = ['acl_group_add_hosts']) def test_execute_add_or_remove_uh_to_topic_acl_bad_u_partial_kill(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1']) acl_addrm.kill_on_failure = True self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, False, 'DoesNotExist: The following users do not exist: ' 'user0')]) sys.exit.expect_call(1).and_raises(cli_mock.ExitException) self.god.mock_io() self.assertRaises(cli_mock.ExitException, acl_addrm.execute) (out, err) = self.god.unmock_io() self.god.check_playback() self._check_output(out=out, err=err, err_words_ok=['DoesNotExist', 'acl_group_add_users', 'user0'], err_words_no = ['acl_group_add_hosts']) def test_execute_add_or_remove_uh_to_topic_acl_bad_users_good_hosts(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1'], hosts=['host0', 'host1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, False, 'DoesNotExist: The following users do not exist: ' 'user0, user1'), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0', 'host1']}, True, None)]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual([], execute_result['users']) self.assertEqual(['acl0'], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, out_words_ok=["Added to ACL 'acl0' hosts:", "host0", "host1"], err_words_ok=['DoesNotExist', 'acl_group_add_users', 'user0', 'user1'], err_words_no = ['acl_group_add_hosts']) def test_execute_add_or_remove_uh_to_topic_acl_good_users_bad_hosts(self): acl_addrm = self._create_add_remove('acl0 with space', users=['user0', 'user1'], hosts=['host0', 'host1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0 with space', 'users': ['user0', 'user1']}, True, None), ('acl_group_add_hosts', {'id': 'acl0 with space', 'hosts': ['host0', 'host1']}, False, 'DoesNotExist: The following hosts do not exist: ' 'host0, host1')]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual(['acl0 with space'], execute_result['users']) self.assertEqual([], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, out_words_ok=["Added to ACL 'acl0 with space' users:", "user0", "user1"], err_words_ok=['DoesNotExist', 'acl_group_add_hosts', 'host0', 'host1'], err_words_no = ['acl_group_add_users']) def test_exe_add_or_remove_uh_to_topic_acl_good_u_bad_hosts_partial(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1'], hosts=['host0', 'host1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, True, None), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0', 'host1']}, False, 'DoesNotExist: The following hosts do not exist: ' 'host1'), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0']}, True, None)]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual(['acl0'], execute_result['users']) self.assertEqual(['acl0'], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, out_words_ok=["Added to ACL 'acl0' users:", "user0", "user1", "host0"], err_words_ok=['DoesNotExist', 'acl_group_add_hosts', 'host1'], err_words_no = ['acl_group_add_users']) def test_execute_add_or_remove_uh_to_topic_acl_bad_users_bad_hosts(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1'], hosts=['host0', 'host1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, False, 'DoesNotExist: The following users do not exist: ' 'user0, user1'), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0', 'host1']}, False, 'DoesNotExist: The following hosts do not exist: ' 'host0, host1')]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual([], execute_result['users']) self.assertEqual([], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, err_words_ok=['DoesNotExist', 'acl_group_add_hosts', 'host0', 'host1', 'acl_group_add_users', 'user0', 'user1']) def test_execute_add_or_remove_uh_to_topic_acl_bad_u_bad_h_partial(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1'], hosts=['host0', 'host1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, False, 'DoesNotExist: The following users do not exist: ' 'user0'), ('acl_group_add_users', {'id': 'acl0', 'users': ['user1']}, True, None), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0', 'host1']}, False, 'DoesNotExist: The following hosts do not exist: ' 'host1'), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0']}, True, None)]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual(['acl0'], execute_result['users']) self.assertEqual(['acl0'], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, out_words_ok=["Added to ACL 'acl0' user:", "Added to ACL 'acl0' host:", 'user1', 'host0'], err_words_ok=['DoesNotExist', 'acl_group_add_hosts', 'host1', 'acl_group_add_users', 'user0']) def test_execute_add_or_remove_to_topic_bad_acl_uh(self): acl_addrm = self._create_add_remove('acl0', users=['user0', 'user1'], hosts=['host0', 'host1']) self.mock_rpcs([('acl_group_add_users', {'id': 'acl0', 'users': ['user0', 'user1']}, False, 'DoesNotExist: acl_group matching ' 'query does not exist.'), ('acl_group_add_hosts', {'id': 'acl0', 'hosts': ['host0', 'host1']}, False, 'DoesNotExist: acl_group matching ' 'query does not exist.')]) execute_result = acl_addrm.execute() self.god.check_playback() self.assertEqual([], execute_result['users']) self.assertEqual([], execute_result['hosts']) self.assertOutput(acl_addrm, execute_result, err_words_ok=['DoesNotExist', 'acl_group_add_hosts', 'acl_group_add_users']) if __name__ == '__main__': unittest.main()