cli/action_common_unittest.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright 2008 Google Inc. All Rights Reserved.
   4
   5 """Tests for action_common."""
   6
   7 import unittest, os, sys, StringIO, copy
   8
   9 import common
  10 from autotest_lib.cli import cli_mock, topic_common, action_common, rpc
  11 from autotest_lib.frontend.afe.json_rpc import proxy
  12
  13 #
  14 # List action
  15 #
  16 class atest_list_unittest(cli_mock.cli_unittest):
  17     def test_check_for_wilcard_none(self):
  18         orig_filters = {'name__in': ['item0', 'item1']}
  19         orig_checks = {'name__in': ['item0', 'item1']}
  20         mytest = action_common.atest_list()
  21
  22         filters = copy.deepcopy(orig_filters)
  23         checks = copy.deepcopy(orig_checks)
  24         mytest.check_for_wildcard(filters, checks)
  25         self.assertEqual(filters, orig_filters)
  26         self.assertEqual(checks, orig_checks)
  27
  28
  29     def test_check_for_wilcard_none_list(self):
  30         orig_filters = {'name__in': ['item0']}
  31         orig_checks = {'name__in': ['item0']}
  32         mytest = action_common.atest_list()
  33
  34         filters = copy.deepcopy(orig_filters)
  35         checks = copy.deepcopy(orig_checks)
  36         mytest.check_for_wildcard(filters, checks)
  37         self.assertEqual(filters, orig_filters)
  38         self.assertEqual(checks, orig_checks)
  39
  40     def test_check_for_wilcard_one_list(self):
  41         filters = {'something__in': ['item*']}
  42         checks = {'something__in': ['item*']}
  43         mytest = action_common.atest_list()
  44
  45         mytest.check_for_wildcard(filters, checks)
  46         self.assertEqual(filters, {'something__startswith': 'item'})
  47         self.assertEqual(checks, {'something__startswith': None})
  48
  49
  50     def test_check_for_wilcard_one_string(self):
  51         filters = {'something__name': 'item*'}
  52         checks = {'something__name': 'item*'}
  53         mytest = action_common.atest_list()
  54
  55         mytest.check_for_wildcard(filters, checks)
  56         self.assertEqual(filters, {'something__name__startswith': 'item'})
  57         self.assertEqual(checks, {'something__name__startswith': None})
  58
  59
  60
  61     def test_check_for_wilcard_one_string_login(self):
  62         filters = {'something__login': 'item*'}
  63         checks = {'something__login': 'item*'}
  64         mytest = action_common.atest_list()
  65
  66         mytest.check_for_wildcard(filters, checks)
  67         self.assertEqual(filters, {'something__login__startswith': 'item'})
  68         self.assertEqual(checks, {'something__login__startswith': None})
  69
  70
  71     def test_check_for_wilcard_two(self):
  72         orig_filters = {'something__in': ['item0*', 'item1*']}
  73         orig_checks = {'something__in': ['item0*', 'item1*']}
  74         mytest = action_common.atest_list()
  75
  76         filters = copy.deepcopy(orig_filters)
  77         checks = copy.deepcopy(orig_checks)
  78         self.god.stub_function(sys, 'exit')
  79         sys.exit.expect_call(1).and_raises(cli_mock.ExitException)
  80         self.god.mock_io()
  81         self.assertRaises(cli_mock.ExitException,
  82                           mytest.check_for_wildcard, filters, checks)
  83         (out, err) = self.god.unmock_io()
  84         self.god.check_playback()
  85         self.assertEqual(filters, orig_filters)
  86         self.assertEqual(checks, orig_checks)
  87
  88
  89     def _atest_list_execute(self, filters={}, check_results={}):
  90         values = [{u'id': 180,
  91                    u'platform': 0,
  92                    u'name': u'label0',
  93                    u'invalid': 0,
  94                    u'kernel_config': u''},
  95                   {u'id': 338,
  96                    u'platform': 0,
  97                    u'name': u'label1',
  98                    u'invalid': 0,
  99                    u'kernel_config': u''}]
 100         mytest = action_common.atest_list()
 101         mytest.afe = rpc.afe_comm()
 102         self.mock_rpcs([('get_labels',
 103                          filters,
 104                          True,
 105                          values)])
 106         self.god.mock_io()
 107         self.assertEqual(values,
 108                          mytest.execute(op='get_labels',
 109                                         filters=filters,
 110                                         check_results=check_results))
 111         (out, err) = self.god.unmock_io()
 112         self.god.check_playback()
 113         return (out, err)
 114
 115
 116     def test_atest_list_execute_no_filters(self):
 117         self._atest_list_execute()
 118
 119
 120     def test_atest_list_execute_filters_all_good(self):
 121         filters = {}
 122         check_results = {}
 123         filters['name__in'] = ['label0', 'label1']
 124         check_results['name__in'] = 'name'
 125         (out, err) = self._atest_list_execute(filters, check_results)
 126         self.assertEqual(err, '')
 127
 128
 129     def test_atest_list_execute_filters_good_and_bad(self):
 130         filters = {}
 131         check_results = {}
 132         filters['name__in'] = ['label0', 'label1', 'label2']
 133         check_results['name__in'] = 'name'
 134         (out, err) = self._atest_list_execute(filters, check_results)
 135         self.assertWords(err, ['Unknown', 'label2'])
 136
 137
 138     def test_atest_list_execute_items_good_and_bad_no_check(self):
 139         filters = {}
 140         check_results = {}
 141         filters['name__in'] = ['label0', 'label1', 'label2']
 142         check_results['name__in'] = None
 143         (out, err) = self._atest_list_execute(filters, check_results)
 144         self.assertEqual(err, '')
 145
 146
 147     def test_atest_list_execute_filters_wildcard(self):
 148         filters = {}
 149         check_results = {}
 150         filters['name__in'] = ['label*']
 151         check_results['name__in'] = 'name'
 152         values = [{u'id': 180,
 153                    u'platform': False,
 154                    u'name': u'label0',
 155                    u'invalid': False,
 156                    u'kernel_config': u''},
 157                   {u'id': 338,
 158                    u'platform': False,
 159                    u'name': u'label1',
 160                    u'invalid': False,
 161                    u'kernel_config': u''}]
 162         mytest = action_common.atest_list()
 163         mytest.afe = rpc.afe_comm()
 164         self.mock_rpcs([('get_labels', {'name__startswith': 'label'},
 165                          True, values)])
 166         self.god.mock_io()
 167         self.assertEqual(values,
 168                          mytest.execute(op='get_labels',
 169                                         filters=filters,
 170                                         check_results=check_results))
 171         (out, err) = self.god.unmock_io()
 172         self.god.check_playback()
 173         self.assertEqual(err, '')
 174
 175
 176
 177 #
 178 # Creation & Deletion of a topic (ACL, label, user)
 179 #
 180 class atest_create_or_delete_unittest(cli_mock.cli_unittest):
 181     def _create_cr_del(self, items):
 182         def _items():
 183             return items
 184         crdel = action_common.atest_create_or_delete()
 185         crdel.afe = rpc.afe_comm()
 186
 187         crdel.topic =  crdel.usage_topic = 'label'
 188         crdel.op_action = 'add'
 189         crdel.get_items = _items
 190         crdel.data['platform'] = False
 191         crdel.data_item_key = 'name'
 192         return crdel
 193
 194
 195     def test_execute_create_one_topic(self):
 196         acr = self._create_cr_del(['label0'])
 197         self.mock_rpcs([('add_label',
 198                          {'name': 'label0', 'platform': False},
 199                          True, 42)])
 200         ret = acr.execute()
 201         self.god.check_playback()
 202         self.assert_(['label0'], ret)
 203
 204
 205     def test_execute_create_two_topics(self):
 206         acr = self._create_cr_del(['label0', 'label1'])
 207         self.mock_rpcs([('add_label',
 208                          {'name': 'label0', 'platform': False},
 209                          True, 42),
 210                         ('add_label',
 211                          {'name': 'label1', 'platform': False},
 212                          True, 43)])
 213         ret = acr.execute()
 214         self.god.check_playback()
 215         self.assertEqualNoOrder(['label0', 'label1'], ret)
 216
 217
 218     def test_execute_create_error(self):
 219         acr = self._create_cr_del(['label0'])
 220         self.mock_rpcs([('add_label',
 221                          {'name': 'label0', 'platform': False},
 222                          False,
 223                          '''ValidationError:
 224                          {'name': 'This value must be unique (label0)'}''')])
 225         ret = acr.execute()
 226         self.god.check_playback()
 227         self.assertEqualNoOrder([], ret)
 228
 229
 230
 231 #
 232 # Adding or Removing users or hosts from a topic(ACL or label)
 233 #
 234 class atest_add_or_remove_unittest(cli_mock.cli_unittest):
 235     def _create_add_remove(self, items, users=None, hosts=None):
 236         def _items():
 237             return [items]
 238         addrm = action_common.atest_add_or_remove()
 239         addrm.afe = rpc.afe_comm()
 240         if users:
 241             addrm.users = users
 242         if hosts:
 243             addrm.hosts = hosts
 244
 245         addrm.topic = 'acl_group'
 246         addrm.msg_topic = 'ACL'
 247         addrm.op_action = 'add'
 248         addrm.msg_done = 'Added to'
 249         addrm.get_items = _items
 250         return addrm
 251
 252
 253     def test__add_remove_uh_to_topic(self):
 254         acl_addrm = self._create_add_remove('acl0',
 255                                         users=['user0', 'user1'])
 256         self.mock_rpcs([('acl_group_add_users',
 257                          {'id': 'acl0',
 258                           'users': ['user0', 'user1']},
 259                          True,
 260                          None)])
 261         acl_addrm._add_remove_uh_to_topic('acl0', 'users')
 262         self.god.check_playback()
 263
 264
 265     def test__add_remove_uh_to_topic_raise(self):
 266         acl_addrm = self._create_add_remove('acl0',
 267                                         users=['user0', 'user1'])
 268         self.assertRaises(AttributeError,
 269                           acl_addrm._add_remove_uh_to_topic,
 270                           'acl0', 'hosts')
 271
 272
 273     def test_execute_add_or_remove_uh_to_topic_acl_users(self):
 274         acl_addrm = self._create_add_remove('acl0',
 275                                         users=['user0', 'user1'])
 276         self.mock_rpcs([('acl_group_add_users',
 277                          {'id': 'acl0',
 278                           'users': ['user0', 'user1']},
 279                          True,
 280                          None)])
 281         execute_result = acl_addrm.execute()
 282         self.god.check_playback()
 283         self.assertEqualNoOrder(['acl0'], execute_result['users'])
 284         self.assertEqual([], execute_result['hosts'])
 285
 286
 287
 288     def test_execute_add_or_remove_uh_to_topic_acl_users_hosts(self):
 289         acl_addrm = self._create_add_remove('acl0',
 290                                             users=['user0', 'user1'],
 291                                             hosts=['host0', 'host1'])
 292         self.mock_rpcs([('acl_group_add_users',
 293                          {'id': 'acl0',
 294                           'users': ['user0', 'user1']},
 295                          True,
 296                          None),
 297                         ('acl_group_add_hosts',
 298                          {'id': 'acl0',
 299                           'hosts': ['host0', 'host1']},
 300                          True,
 301                          None)])
 302         execute_result = acl_addrm.execute()
 303         self.god.check_playback()
 304         self.assertEqualNoOrder(['acl0'], execute_result['users'])
 305         self.assertEqualNoOrder(['acl0'], execute_result['hosts'])
 306
 307
 308     def test_execute_add_or_remove_uh_to_topic_acl_bad_users(self):
 309         acl_addrm = self._create_add_remove('acl0',
 310                                             users=['user0', 'user1'])
 311         self.mock_rpcs([('acl_group_add_users',
 312                          {'id': 'acl0',
 313                           'users': ['user0', 'user1']},
 314                          False,
 315                          'DoesNotExist: The following users do not exist: '
 316                          'user0, user1')])
 317         execute_result = acl_addrm.execute()
 318         self.god.check_playback()
 319         self.assertEqual([], execute_result['users'])
 320         self.assertEqual([], execute_result['hosts'])
 321         self.assertOutput(acl_addrm, execute_result,
 322                           err_words_ok=['DoesNotExist',
 323                                         'acl_group_add_users',
 324                                         'user0', 'user1'],
 325                           err_words_no = ['acl_group_add_hosts'])
 326
 327
 328     def test_execute_add_or_remove_uh_to_topic_acl_bad_users_partial(self):
 329         acl_addrm = self._create_add_remove('acl0',
 330                                             users=['user0', 'user1'])
 331         self.mock_rpcs([('acl_group_add_users',
 332                          {'id': 'acl0',
 333                           'users': ['user0', 'user1']},
 334                          False,
 335                          'DoesNotExist: The following users do not exist: '
 336                          'user0'),
 337                         ('acl_group_add_users',
 338                          {'id': 'acl0',
 339                           'users': ['user1']},
 340                          True,
 341                          None)])
 342         execute_result = acl_addrm.execute()
 343         self.god.check_playback()
 344         self.assertEqual(['acl0'], execute_result['users'])
 345         self.assertEqual([], execute_result['hosts'])
 346         self.assertOutput(acl_addrm, execute_result,
 347                           out_words_ok=["Added to ACL 'acl0'", 'user1'],
 348                           err_words_ok=['DoesNotExist',
 349                                         'acl_group_add_users',
 350                                         'user0'],
 351                           err_words_no = ['acl_group_add_hosts'])
 352
 353
 354     def test_execute_add_or_remove_uh_to_topic_acl_bad_u_partial_kill(self):
 355         acl_addrm = self._create_add_remove('acl0',
 356                                             users=['user0', 'user1'])
 357         acl_addrm.kill_on_failure = True
 358         self.mock_rpcs([('acl_group_add_users',
 359                          {'id': 'acl0',
 360                           'users': ['user0', 'user1']},
 361                          False,
 362                          'DoesNotExist: The following users do not exist: '
 363                          'user0')])
 364         sys.exit.expect_call(1).and_raises(cli_mock.ExitException)
 365         self.god.mock_io()
 366         self.assertRaises(cli_mock.ExitException, acl_addrm.execute)
 367         (out, err) = self.god.unmock_io()
 368         self.god.check_playback()
 369         self._check_output(out=out, err=err,
 370                           err_words_ok=['DoesNotExist',
 371                                         'acl_group_add_users',
 372                                         'user0'],
 373                           err_words_no = ['acl_group_add_hosts'])
 374
 375
 376     def test_execute_add_or_remove_uh_to_topic_acl_bad_users_good_hosts(self):
 377         acl_addrm = self._create_add_remove('acl0',
 378                                         users=['user0', 'user1'],
 379                                         hosts=['host0', 'host1'])
 380         self.mock_rpcs([('acl_group_add_users',
 381                          {'id': 'acl0',
 382                           'users': ['user0', 'user1']},
 383                          False,
 384                          'DoesNotExist: The following users do not exist: '
 385                          'user0, user1'),
 386                         ('acl_group_add_hosts',
 387                          {'id': 'acl0',
 388                           'hosts': ['host0', 'host1']},
 389                          True,
 390                          None)])
 391
 392         execute_result = acl_addrm.execute()
 393         self.god.check_playback()
 394         self.assertEqual([], execute_result['users'])
 395         self.assertEqual(['acl0'], execute_result['hosts'])
 396         self.assertOutput(acl_addrm, execute_result,
 397                           out_words_ok=["Added to ACL 'acl0' hosts:",
 398                                         "host0", "host1"],
 399                           err_words_ok=['DoesNotExist',
 400                                         'acl_group_add_users',
 401                                         'user0', 'user1'],
 402                           err_words_no = ['acl_group_add_hosts'])
 403
 404
 405     def test_execute_add_or_remove_uh_to_topic_acl_good_users_bad_hosts(self):
 406         acl_addrm = self._create_add_remove('acl0 with space',
 407                                         users=['user0', 'user1'],
 408                                         hosts=['host0', 'host1'])
 409         self.mock_rpcs([('acl_group_add_users',
 410                          {'id': 'acl0 with space',
 411                           'users': ['user0', 'user1']},
 412                          True,
 413                          None),
 414                         ('acl_group_add_hosts',
 415                          {'id': 'acl0 with space',
 416                           'hosts': ['host0', 'host1']},
 417                          False,
 418                          'DoesNotExist: The following hosts do not exist: '
 419                          'host0, host1')])
 420
 421         execute_result = acl_addrm.execute()
 422         self.god.check_playback()
 423         self.assertEqual(['acl0 with space'], execute_result['users'])
 424         self.assertEqual([], execute_result['hosts'])
 425         self.assertOutput(acl_addrm, execute_result,
 426                           out_words_ok=["Added to ACL 'acl0 with space' users:",
 427                                         "user0", "user1"],
 428                           err_words_ok=['DoesNotExist',
 429                                         'acl_group_add_hosts',
 430                                         'host0', 'host1'],
 431                           err_words_no = ['acl_group_add_users'])
 432
 433
 434     def test_exe_add_or_remove_uh_to_topic_acl_good_u_bad_hosts_partial(self):
 435         acl_addrm = self._create_add_remove('acl0',
 436                                         users=['user0', 'user1'],
 437                                         hosts=['host0', 'host1'])
 438         self.mock_rpcs([('acl_group_add_users',
 439                          {'id': 'acl0',
 440                           'users': ['user0', 'user1']},
 441                          True,
 442                          None),
 443                         ('acl_group_add_hosts',
 444                          {'id': 'acl0',
 445                           'hosts': ['host0', 'host1']},
 446                          False,
 447                          'DoesNotExist: The following hosts do not exist: '
 448                          'host1'),
 449                         ('acl_group_add_hosts',
 450                          {'id': 'acl0',
 451                           'hosts': ['host0']},
 452                          True,
 453                          None)])
 454
 455         execute_result = acl_addrm.execute()
 456         self.god.check_playback()
 457         self.assertEqual(['acl0'], execute_result['users'])
 458         self.assertEqual(['acl0'], execute_result['hosts'])
 459         self.assertOutput(acl_addrm, execute_result,
 460                           out_words_ok=["Added to ACL 'acl0' users:",
 461                                         "user0", "user1", "host0"],
 462                           err_words_ok=['DoesNotExist',
 463                                         'acl_group_add_hosts',
 464                                         'host1'],
 465                           err_words_no = ['acl_group_add_users'])
 466
 467
 468     def test_execute_add_or_remove_uh_to_topic_acl_bad_users_bad_hosts(self):
 469         acl_addrm = self._create_add_remove('acl0',
 470                                         users=['user0', 'user1'],
 471                                         hosts=['host0', 'host1'])
 472         self.mock_rpcs([('acl_group_add_users',
 473                          {'id': 'acl0',
 474                           'users': ['user0', 'user1']},
 475                          False,
 476                          'DoesNotExist: The following users do not exist: '
 477                          'user0, user1'),
 478                         ('acl_group_add_hosts',
 479                          {'id': 'acl0',
 480                           'hosts': ['host0', 'host1']},
 481                          False,
 482                          'DoesNotExist: The following hosts do not exist: '
 483                          'host0, host1')])
 484
 485
 486         execute_result = acl_addrm.execute()
 487         self.god.check_playback()
 488         self.assertEqual([], execute_result['users'])
 489         self.assertEqual([], execute_result['hosts'])
 490         self.assertOutput(acl_addrm, execute_result,
 491                           err_words_ok=['DoesNotExist',
 492                                         'acl_group_add_hosts',
 493                                         'host0', 'host1',
 494                                         'acl_group_add_users',
 495                                         'user0', 'user1'])
 496
 497
 498     def test_execute_add_or_remove_uh_to_topic_acl_bad_u_bad_h_partial(self):
 499         acl_addrm = self._create_add_remove('acl0',
 500                                         users=['user0', 'user1'],
 501                                         hosts=['host0', 'host1'])
 502         self.mock_rpcs([('acl_group_add_users',
 503                          {'id': 'acl0',
 504                           'users': ['user0', 'user1']},
 505                          False,
 506                          'DoesNotExist: The following users do not exist: '
 507                          'user0'),
 508                         ('acl_group_add_users',
 509                          {'id': 'acl0',
 510                           'users': ['user1']},
 511                          True,
 512                          None),
 513                         ('acl_group_add_hosts',
 514                          {'id': 'acl0',
 515                           'hosts': ['host0', 'host1']},
 516                          False,
 517                          'DoesNotExist: The following hosts do not exist: '
 518                          'host1'),
 519                         ('acl_group_add_hosts',
 520                          {'id': 'acl0',
 521                           'hosts': ['host0']},
 522                          True,
 523                          None)])
 524         execute_result = acl_addrm.execute()
 525         self.god.check_playback()
 526         self.assertEqual(['acl0'], execute_result['users'])
 527         self.assertEqual(['acl0'], execute_result['hosts'])
 528         self.assertOutput(acl_addrm, execute_result,
 529                           out_words_ok=["Added to ACL 'acl0' user:",
 530                                         "Added to ACL 'acl0' host:",
 531                                         'user1', 'host0'],
 532                           err_words_ok=['DoesNotExist',
 533                                         'acl_group_add_hosts',
 534                                         'host1',
 535                                         'acl_group_add_users',
 536                                         'user0'])
 537
 538
 539     def test_execute_add_or_remove_to_topic_bad_acl_uh(self):
 540         acl_addrm = self._create_add_remove('acl0',
 541                                         users=['user0', 'user1'],
 542                                         hosts=['host0', 'host1'])
 543         self.mock_rpcs([('acl_group_add_users',
 544                          {'id': 'acl0',
 545                           'users': ['user0', 'user1']},
 546                          False,
 547                          'DoesNotExist: acl_group matching '
 548                          'query does not exist.'),
 549                         ('acl_group_add_hosts',
 550                          {'id': 'acl0',
 551                           'hosts': ['host0', 'host1']},
 552                          False,
 553                          'DoesNotExist: acl_group matching '
 554                          'query does not exist.')])
 555         execute_result = acl_addrm.execute()
 556         self.god.check_playback()
 557         self.assertEqual([], execute_result['users'])
 558         self.assertEqual([], execute_result['hosts'])
 559         self.assertOutput(acl_addrm, execute_result,
 560                           err_words_ok=['DoesNotExist',
 561                                         'acl_group_add_hosts',
 562                                         'acl_group_add_users'])
 563
 564
 565 if __name__ == '__main__':
 566     unittest.main()