1#!/usr/bin/python
2
3import unittest
4
5import common
6from autotest_lib.client.common_lib.test_utils import mock
7from autotest_lib.tko import models
8from autotest_lib.tko.parsers import version_0
9
10
11class test_job_load_from_dir(unittest.TestCase):
12    keyval_return = {'job_queued': 1234567890,
13                     'job_started': 1234567891,
14                     'job_finished': 1234567892,
15                     'user': 'janet',
16                     'label': 'steeltown',
17                     'hostname': 'abc123'}
18
19
20    def setUp(self):
21        self.god = mock.mock_god()
22        self.god.stub_function(models.job, 'read_keyval')
23        self.god.stub_function(version_0.job, 'find_hostname')
24        self.god.stub_function(models.test, 'parse_host_keyval')
25
26
27    def tearDown(self):
28        self.god.unstub_all()
29
30
31    def _expect_host_keyval(self, hostname, platform=None):
32        return_dict = {}
33        if platform:
34            return_dict['platform'] = platform
35            return_dict['labels'] = (platform + ',other_label,' +
36                                     'model%%3A%s' % platform)
37        (models.test.parse_host_keyval.expect_call('.', hostname)
38                .and_return(return_dict))
39
40
41    def test_load_from_dir_simple(self):
42        models.job.read_keyval.expect_call('.').and_return(
43                dict(self.keyval_return))
44        self._expect_host_keyval('abc123', 'my_platform')
45        job = version_0.job.load_from_dir('.')
46        self.assertEqual('janet', job['user'])
47        self.assertEqual('steeltown', job['label'])
48        self.assertEqual('abc123', job['machine'])
49        self.assertEqual('my_platform', job['machine_group'])
50        self.god.check_playback()
51
52
53    def _setup_two_machines(self):
54        raw_keyval = dict(self.keyval_return)
55        raw_keyval['hostname'] = 'easyas,abc123'
56        models.job.read_keyval.expect_call('.').and_return(raw_keyval)
57
58
59    def test_load_from_dir_two_machines(self):
60        self._setup_two_machines()
61        version_0.job.find_hostname.expect_call('.').and_raises(
62                    version_0.NoHostnameError('find_hostname stubbed out'))
63        self._expect_host_keyval('easyas', 'platform')
64        self._expect_host_keyval('abc123', 'platform')
65
66        job = version_0.job.load_from_dir('.')
67        self.assertEqual('easyas,abc123', job['machine'])
68        self.assertEqual('platform', job['machine_group'])
69
70        self.god.check_playback()
71
72
73    def test_load_from_dir_two_machines_with_find_hostname(self):
74        self._setup_two_machines()
75        version_0.job.find_hostname.expect_call('.').and_return('foo')
76        self._expect_host_keyval('foo')
77
78        job = version_0.job.load_from_dir('.')
79        self.assertEqual('foo', job['machine'])
80
81        self.god.check_playback()
82
83
84    def test_load_from_dir_two_machines_different_platforms(self):
85        self._setup_two_machines()
86        version_0.job.find_hostname.expect_call('.').and_raises(
87                    version_0.NoHostnameError('find_hostname stubbed out'))
88        self._expect_host_keyval('easyas', 'platformZ')
89        self._expect_host_keyval('abc123', 'platformA')
90
91        job = version_0.job.load_from_dir('.')
92        self.assertEqual('easyas,abc123', job['machine'])
93        self.assertEqual('platformA,platformZ', job['machine_group'])
94
95        self.god.check_playback()
96
97    def test_load_from_dir_one_machine_group_name(self):
98        raw_keyval = dict(self.keyval_return)
99        raw_keyval['host_group_name'] = 'jackson five'
100        models.job.read_keyval.expect_call('.').and_return(raw_keyval)
101        self._expect_host_keyval('abc123')
102        job = version_0.job.load_from_dir('.')
103        self.assertEqual('janet', job['user'])
104        self.assertEqual('abc123', job['machine'])
105        self.god.check_playback()
106
107
108    def test_load_from_dir_multi_machine_group_name(self):
109        raw_keyval = dict(self.keyval_return)
110        raw_keyval['user'] = 'michael'
111        raw_keyval['hostname'] = 'abc123,dancingmachine'
112        raw_keyval['host_group_name'] = 'jackson five'
113        models.job.read_keyval.expect_call('.').and_return(raw_keyval)
114        self._expect_host_keyval('jackson five')
115        job = version_0.job.load_from_dir('.')
116        self.assertEqual('michael', job['user'])
117        # The host_group_name is used instead because machine appeared to be
118        # a comma separated list.
119        self.assertEqual('jackson five', job['machine'])
120        self.god.check_playback()
121
122
123    def test_load_from_dir_no_machine_group_name(self):
124        raw_keyval = dict(self.keyval_return)
125        del raw_keyval['hostname']
126        raw_keyval['host_group_name'] = 'jackson five'
127        models.job.read_keyval.expect_call('.').and_return(raw_keyval)
128        self._expect_host_keyval('jackson five')
129        job = version_0.job.load_from_dir('.')
130        # The host_group_name is used because there is no machine.
131        self.assertEqual('jackson five', job['machine'])
132        self.god.check_playback()
133
134
135class test_status_line(unittest.TestCase):
136    statuses = ["GOOD", "WARN", "FAIL", "ABORT"]
137
138
139    def test_handles_start(self):
140        line = version_0.status_line(0, "START", "----", "test",
141                                     "", {})
142        self.assertEquals(line.type, "START")
143        self.assertEquals(line.status, None)
144
145
146    def test_fails_info(self):
147        self.assertRaises(AssertionError,
148                          version_0.status_line, 0, "INFO", "----", "----",
149                          "", {})
150
151
152    def test_handles_status(self):
153        for stat in self.statuses:
154            line = version_0.status_line(0, stat, "----", "test",
155                                         "", {})
156            self.assertEquals(line.type, "STATUS")
157            self.assertEquals(line.status, stat)
158
159
160    def test_handles_endstatus(self):
161        for stat in self.statuses:
162            line = version_0.status_line(0, "END " + stat, "----",
163                                         "test", "", {})
164            self.assertEquals(line.type, "END")
165            self.assertEquals(line.status, stat)
166
167
168    def test_fails_on_bad_status(self):
169        for stat in self.statuses:
170            self.assertRaises(AssertionError,
171                              version_0.status_line, 0,
172                              "BAD " + stat, "----", "test",
173                              "", {})
174
175
176    def test_saves_all_fields(self):
177        line = version_0.status_line(5, "GOOD", "subdir_name",
178                                     "test_name", "my reason here",
179                                     {"key1": "value",
180                                      "key2": "another value",
181                                      "key3": "value3"})
182        self.assertEquals(line.indent, 5)
183        self.assertEquals(line.status, "GOOD")
184        self.assertEquals(line.subdir, "subdir_name")
185        self.assertEquals(line.testname, "test_name")
186        self.assertEquals(line.reason, "my reason here")
187        self.assertEquals(line.optional_fields,
188                          {"key1": "value", "key2": "another value",
189                           "key3": "value3"})
190
191
192    def test_parses_blank_subdir(self):
193        line = version_0.status_line(0, "GOOD", "----", "test",
194                                     "", {})
195        self.assertEquals(line.subdir, None)
196
197
198    def test_parses_blank_testname(self):
199        line = version_0.status_line(0, "GOOD", "subdir", "----",
200                                     "", {})
201        self.assertEquals(line.testname, None)
202
203
204    def test_parse_line_smoketest(self):
205        input_data = ("\t\t\tGOOD\t----\t----\t"
206                      "field1=val1\tfield2=val2\tTest Passed")
207        line = version_0.status_line.parse_line(input_data)
208        self.assertEquals(line.indent, 3)
209        self.assertEquals(line.type, "STATUS")
210        self.assertEquals(line.status, "GOOD")
211        self.assertEquals(line.subdir, None)
212        self.assertEquals(line.testname, None)
213        self.assertEquals(line.reason, "Test Passed")
214        self.assertEquals(line.optional_fields,
215                          {"field1": "val1", "field2": "val2"})
216
217    def test_parse_line_handles_newline(self):
218        input_data = ("\t\tGOOD\t----\t----\t"
219                      "field1=val1\tfield2=val2\tNo newline here!")
220        for suffix in ("", "\n"):
221            line = version_0.status_line.parse_line(input_data +
222                                                    suffix)
223            self.assertEquals(line.indent, 2)
224            self.assertEquals(line.type, "STATUS")
225            self.assertEquals(line.status, "GOOD")
226            self.assertEquals(line.subdir, None)
227            self.assertEquals(line.testname, None)
228            self.assertEquals(line.reason, "No newline here!")
229            self.assertEquals(line.optional_fields,
230                              {"field1": "val1",
231                               "field2": "val2"})
232
233
234    def test_parse_line_handles_embedded_new_lines(self):
235        input_data = ("\tEND FAIL\t----\ttest\tfield1=val1\tStatus\nwith\n"
236                      "embedded\nnew lines\n")
237
238        line = version_0.status_line.parse_line(input_data)
239        self.assertEquals(line.indent, 1)
240        self.assertEquals(line.type, "END")
241        self.assertEquals(line.status, "FAIL")
242        self.assertEquals(line.subdir, None)
243        self.assertEquals(line.testname, "test")
244        self.assertEquals(line.reason, "Status\nwith\nembedded\nnew lines")
245        self.assertEquals(line.optional_fields, {"field1": "val1"})
246
247
248    def test_parse_line_fails_on_untabbed_lines(self):
249        input_data = "   GOOD\trandom\tfields\tof text"
250        line = version_0.status_line.parse_line(input_data)
251        self.assertEquals(line, None)
252        line = version_0.status_line.parse_line(input_data.lstrip())
253        self.assertEquals(line.indent, 0)
254        self.assertEquals(line.type, "STATUS")
255        self.assertEquals(line.status, "GOOD")
256        self.assertEquals(line.subdir, "random")
257        self.assertEquals(line.testname, "fields")
258        self.assertEquals(line.reason, "of text")
259        self.assertEquals(line.optional_fields, {})
260
261
262    def test_parse_line_fails_on_incomplete_lines(self):
263        input_data = "\t\tGOOD\tfield\tsecond field"
264        complete_data = input_data + "\tneeded last field"
265        line = version_0.status_line.parse_line(input_data)
266        self.assertEquals(line, None)
267        line = version_0.status_line.parse_line(complete_data)
268        self.assertEquals(line.indent, 2)
269        self.assertEquals(line.type, "STATUS")
270        self.assertEquals(line.status, "GOOD")
271        self.assertEquals(line.subdir, "field")
272        self.assertEquals(line.testname, "second field")
273        self.assertEquals(line.reason, "needed last field")
274        self.assertEquals(line.optional_fields, {})
275
276
277    def test_parse_line_handles_tabs_in_reason(self):
278        input_data = ("\tEND FAIL\t----\ttest\tfield1=val1\tfield2=val2\tReason"
279                      " with\ta\tcouple\ttabs")
280
281        line = version_0.status_line.parse_line(input_data)
282        self.assertEquals(line.indent, 1)
283        self.assertEquals(line.type, "END")
284        self.assertEquals(line.status, "FAIL")
285        self.assertEquals(line.subdir, None)
286        self.assertEquals(line.testname, "test")
287        self.assertEquals(line.reason, "Reason with\ta\tcouple\ttabs")
288        self.assertEquals(line.optional_fields, {"field1": "val1",
289                                                 "field2": "val2"})
290
291
292if __name__ == "__main__":
293    unittest.main()
294