1#!/usr/bin/python2
2
3import unittest
4import common
5from autotest_lib.frontend import setup_django_environment
6from autotest_lib.frontend import setup_test_environment
7from autotest_lib.frontend.tko import csv_encoder
8
9class CsvEncodingTest(unittest.TestCase):
10    def _make_request(self, method, columns=None):
11        request = dict(method=method)
12        if columns:
13            request['columns'] = columns
14        return request
15
16
17    def _make_group(self, header_indices, pass_count, complete_count,
18                    incomplete_count=0):
19        return dict(header_indices=header_indices, pass_count=pass_count,
20                    complete_count=complete_count,
21                    incomplete_count=incomplete_count)
22
23
24    def _encode_and_check_result(self, request, result, *expected_csv_rows):
25        encoder = csv_encoder.encoder(request, result)
26        response = encoder.encode()
27        csv_result = response.content
28        expected_csv = '\r\n'.join(expected_csv_rows) + '\r\n'
29        self.assertEquals(csv_result, expected_csv)
30
31
32    def test_spreadsheet_encoder(self):
33        request = self._make_request('get_status_counts')
34        response = {'header_values' :
35                        [[('row1',), ('row2',), ('comma,header',)],
36                         [('col1', 'sub1'), ('col1', 'sub2'),
37                          ('col2', 'sub1')]],
38                    'groups' : [self._make_group((0, 0), 1, 2),
39                                self._make_group((1, 2), 3, 4, 5)]}
40
41        self._encode_and_check_result(request, response,
42                                      ',col1/sub1,col1/sub2,col2/sub1',
43                                      'row1,1 / 2,,',
44                                      'row2,,,3 / 4 (5 incomplete)',
45                                      '"comma,header",,,')
46
47
48    def test_table_encoder(self):
49        request = self._make_request('get_test_views', [['col1', 'Column 1'],
50                                                        ['col2', 'Column 2']])
51        response = [{'col1' : 'foo', 'col2' : 'bar'},
52                    {'col1' : 'baz', 'col2' : 'asdf'}]
53        self._encode_and_check_result(request, response,
54                                      'Column 1,Column 2',
55                                      'foo,bar',
56                                      'baz,asdf')
57
58
59    def test_grouped_table_encoder(self):
60        request = self._make_request('get_group_counts',
61                                     [['col1', 'Column 1'],
62                                      ['group_count', 'Count in group']])
63        response = {'header_values' : 'unused',
64                    'groups' : [{'col1' : 'foo', 'group_count' : 1},
65                                {'col1' : 'baz', 'group_count' : 3}]}
66        self._encode_and_check_result(request, response,
67                                      'Column 1,Count in group',
68                                      'foo,1',
69                                      'baz,3')
70
71
72    def _status_count_dict(self, col1_value, pass_count, complete_count,
73                                  incomplete_count):
74        return dict(col1=col1_value, pass_count=pass_count,
75                    complete_count=complete_count,
76                    incomplete_count=incomplete_count)
77
78
79    def test_status_count_table_encoder(self):
80        request = self._make_request('get_status_counts',
81                                     [['col1', 'Column 1'],
82                                      ['_unused_', 'Test pass rate']])
83        response = {'header_values' : 'unused',
84                    'groups' : [self._status_count_dict('foo', 1, 2, 0),
85                                self._status_count_dict('baz', 4, 5, 6)]}
86        self._encode_and_check_result(request, response,
87                                      'Column 1,Test pass rate',
88                                      'foo,1 / 2',
89                                      'baz,4 / 5 (6 incomplete)')
90
91
92    def test_extra_info_spreadsheet_encoder(self):
93        request = self._make_request('get_latest_tests')
94
95
96        group1 = self._make_group((0, 0), 1, 1)
97        group2 = self._make_group((1, 0), 1, 1)
98
99        group1['extra_info'] = ['info1', 'info2']
100        group2['extra_info'] = ['', 'info3']
101
102        response = {'header_values' :
103                        [[('row1',), ('row2',)],
104                         [('col1',), ('col2',)]],
105                    'groups' : [group1, group2]}
106
107        self._encode_and_check_result(request, response,
108                                      ',col1,col2',
109                                      'row1,"1 / 1\ninfo1\ninfo2",',
110                                      'row2,"1 / 1\n\ninfo3",')
111
112
113    def test_unhandled_method(self):
114        request = self._make_request('foo')
115        self._encode_and_check_result(request, None,
116                                      'Unhandled method foo (this indicates a '
117                                      'bug)')
118
119
120if __name__ == '__main__':
121    unittest.main()
122