xref: /aosp_15_r20/external/rappor/pipeline/combine_status_test.py (revision 2abb31345f6c95944768b5222a9a5ed3fc68cc00)
1*2abb3134SXin Li#!/usr/bin/python -S
2*2abb3134SXin Li"""
3*2abb3134SXin Licombine_status_test.py: Tests for combine_status.py
4*2abb3134SXin Li"""
5*2abb3134SXin Li
6*2abb3134SXin Liimport csv
7*2abb3134SXin Liimport cStringIO
8*2abb3134SXin Liimport unittest
9*2abb3134SXin Li
10*2abb3134SXin Liimport combine_status  # module under test
11*2abb3134SXin Li
12*2abb3134SXin Li
13*2abb3134SXin Li# TODO: Make these test more the header row.  They rely heavily on the file
14*2abb3134SXin Li# system!
15*2abb3134SXin Li
16*2abb3134SXin Liclass CombineStatusTest(unittest.TestCase):
17*2abb3134SXin Li
18*2abb3134SXin Li  def testCombineDistTaskStatus(self):
19*2abb3134SXin Li    stdin = cStringIO.StringIO('')
20*2abb3134SXin Li    out = cStringIO.StringIO()
21*2abb3134SXin Li    c_out = csv.writer(out)
22*2abb3134SXin Li
23*2abb3134SXin Li    combine_status.CombineDistTaskStatus(stdin, c_out, {})
24*2abb3134SXin Li    actual = out.getvalue()
25*2abb3134SXin Li    self.assert_(actual.startswith('job_id,params_file,'), actual)
26*2abb3134SXin Li
27*2abb3134SXin Li  def testCombineAssocTaskStatus(self):
28*2abb3134SXin Li    stdin = cStringIO.StringIO('')
29*2abb3134SXin Li    out = cStringIO.StringIO()
30*2abb3134SXin Li    c_out = csv.writer(out)
31*2abb3134SXin Li
32*2abb3134SXin Li    combine_status.CombineAssocTaskStatus(stdin, c_out)
33*2abb3134SXin Li    actual = out.getvalue()
34*2abb3134SXin Li    self.assert_(actual.startswith('job_id,metric,'), actual)
35*2abb3134SXin Li
36*2abb3134SXin Li
37*2abb3134SXin Liif __name__ == '__main__':
38*2abb3134SXin Li  unittest.main()
39