liyujie
2025-08-28 b3810562527858a3b3d98ffa6e9c9c5b0f4a9a8e
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
#!/usr/bin/env python2
#
# Copyright 2008 Google Inc. All Rights Reserved.
 
"""Tests for shard."""
 
import unittest
 
import common
from autotest_lib.cli import cli_mock
 
 
class shard_list_unittest(cli_mock.cli_unittest):
    values = [{'hostname': u'shard1', u'id': 1, 'labels': ['board:lumpy']},
              {'hostname': u'shard2', u'id': 3, 'labels': ['board:daisy']},
              {'hostname': u'shard3', u'id': 5, 'labels': ['board:stumpy']},
              {'hostname': u'shard4', u'id': 6, 'labels': ['board:link']}]
 
 
    def test_shard_list(self):
        self.run_cmd(argv=['atest', 'shard', 'list'],
                     rpcs=[('get_shards', {}, True, self.values)],
                     out_words_ok=['shard1', 'shard2', 'shard3', 'shard4'],
                     out_words_no=['plat0', 'plat1'])
 
 
class shard_create_unittest(cli_mock.cli_unittest):
    def test_execute_create_two_shards(self):
        self.run_cmd(argv=['atest', 'shard', 'create',
                           '-l', 'board:lumpy', 'shard0'],
                     rpcs=[('add_shard',
                            {'hostname': 'shard0', 'labels': 'board:lumpy'},
                            True, 42)],
                     out_words_ok=['Created', 'shard0'])
 
 
    def test_execute_create_two_shards_bad(self):
        self.run_cmd(argv=['atest', 'shard', 'create',
                           '-l', 'board:lumpy', 'shard0'],
                     rpcs=[('add_shard',
                            {'hostname': 'shard0', 'labels': 'board:lumpy'},
                            False,
                            '''ValidationError: {'name':
                            'This value must be unique (shard1)'}''')],
                     out_words_no=['shard0'],
                     err_words_ok=['shard0', 'ValidationError'])
 
 
class shard_add_boards_unittest(cli_mock.cli_unittest):
    def test_execute_add_boards_to_shard(self):
        self.run_cmd(argv=['atest', 'shard', 'add_boards',
                           '-l', 'board:lumpy', 'shard0'],
                     rpcs=[('add_board_to_shard',
                           {'hostname': 'shard0', 'labels':'board:lumpy'},
                            True, 42)],
                     out_words_ok=['Added boards', 'board:lumpy', 'shard0'])
 
 
    def test_execute_add_boards_to_shard_bad(self):
        self.run_cmd(argv=['atest', 'shard', 'add_boards',
                           '-l', 'board:lumpy', 'shard0'],
                     rpcs=[('add_board_to_shard',
                           {'hostname': 'shard0', 'labels':'board:lumpy'},
                            False,
                            '''RPCException: board:lumpy is already on shard
                            shard0''')],
                     out_words_no=['shard0'],
                     err_words_ok=['shard0', 'RPCException'])
 
 
    def test_execute_add_boards_to_shard_fail_when_shard_nonexist(self):
        self.run_cmd(argv=['atest', 'shard', 'add_boards',
                           '-l', 'board:lumpy', 'shard0'],
                     rpcs=[('add_board_to_shard',
                           {'hostname': 'shard0', 'labels':'board:lumpy'},
                            False,
                            '''DoesNotExist: Shard matching query does not
                            exist. Lookup parameters were {'hostname':
                            'shard0'}''')],
                     out_words_no=['shard0'],
                     err_words_ok=['shard0', 'DoesNotExist:'])
 
 
class shard_delete_unittest(cli_mock.cli_unittest):
    def test_execute_delete_shards(self):
        self.run_cmd(argv=['atest', 'shard', 'delete',
                           'shard0', 'shard1', '--no-confirmation'],
                     rpcs=[('delete_shard', {'hostname': 'shard0'}, True, None),
                           ('delete_shard', {'hostname': 'shard1'}, True, None)
                           ],
                     out_words_ok=['Deleted', 'shard0', 'shard1'])
 
 
if __name__ == '__main__':
    unittest.main()