1import json
2import time
3from six.moves import cStringIO as StringIO
4
5import mock
6
7from mozlog import handlers, structuredlog
8
9from ..formatters import wptreport
10from ..formatters.wptscreenshot import WptscreenshotFormatter
11from ..formatters.wptreport import WptreportFormatter
12
13
14def test_wptreport_runtime(capfd):
15    # setup the logger
16    output = StringIO()
17    logger = structuredlog.StructuredLogger("test_a")
18    logger.add_handler(handlers.StreamHandler(output, WptreportFormatter()))
19
20    # output a bunch of stuff
21    logger.suite_start(["test-id-1"], run_info={})
22    logger.test_start("test-id-1")
23    time.sleep(0.125)
24    logger.test_end("test-id-1", "PASS")
25    logger.suite_end()
26
27    # check nothing got output to stdout/stderr
28    # (note that mozlog outputs exceptions during handling to stderr!)
29    captured = capfd.readouterr()
30    assert captured.out == ""
31    assert captured.err == ""
32
33    # check the actual output of the formatter
34    output.seek(0)
35    output_obj = json.load(output)
36    # be relatively lax in case of low resolution timers
37    # 62 is 0.125s = 125ms / 2 = 62ms (assuming int maths)
38    # this provides a margin of 62ms, sufficient for even DOS (55ms timer)
39    assert output_obj["results"][0]["duration"] >= 62
40
41
42def test_wptreport_run_info_optional(capfd):
43    """per the mozlog docs, run_info is optional; check we work without it"""
44    # setup the logger
45    output = StringIO()
46    logger = structuredlog.StructuredLogger("test_a")
47    logger.add_handler(handlers.StreamHandler(output, WptreportFormatter()))
48
49    # output a bunch of stuff
50    logger.suite_start(["test-id-1"])  # no run_info arg!
51    logger.test_start("test-id-1")
52    logger.test_end("test-id-1", "PASS")
53    logger.suite_end()
54
55    # check nothing got output to stdout/stderr
56    # (note that mozlog outputs exceptions during handling to stderr!)
57    captured = capfd.readouterr()
58    assert captured.out == ""
59    assert captured.err == ""
60
61    # check the actual output of the formatter
62    output.seek(0)
63    output_obj = json.load(output)
64    assert "run_info" not in output_obj or output_obj["run_info"] == {}
65
66
67def test_wptreport_lone_surrogate(capfd):
68    output = StringIO()
69    logger = structuredlog.StructuredLogger("test_a")
70    logger.add_handler(handlers.StreamHandler(output, WptreportFormatter()))
71
72    # output a bunch of stuff
73    logger.suite_start(["test-id-1"])  # no run_info arg!
74    logger.test_start("test-id-1")
75    logger.test_status("test-id-1",
76                       subtest=u"Name with surrogate\uD800",
77                       status="FAIL",
78                       message=u"\U0001F601 \uDE0A\uD83D")
79    logger.test_end("test-id-1",
80                    status="PASS",
81                    message=u"\uDE0A\uD83D \U0001F601")
82    logger.suite_end()
83
84    # check nothing got output to stdout/stderr
85    # (note that mozlog outputs exceptions during handling to stderr!)
86    captured = capfd.readouterr()
87    assert captured.out == ""
88    assert captured.err == ""
89
90    # check the actual output of the formatter
91    output.seek(0)
92    output_obj = json.load(output)
93    test = output_obj["results"][0]
94    assert test["message"] == u"U+de0aU+d83d \U0001F601"
95    subtest = test["subtests"][0]
96    assert subtest["name"] == u"Name with surrogateU+d800"
97    assert subtest["message"] == u"\U0001F601 U+de0aU+d83d"
98
99
100def test_wptreport_lone_surrogate_ucs2(capfd):
101    # Since UCS4 is a superset of UCS2 we can meaningfully test the UCS2 code on a
102    # UCS4 build, but not the reverse. However UCS2 is harder to handle and UCS4 is
103    # the commonest (and sensible) configuration, so that's OK.
104    output = StringIO()
105    logger = structuredlog.StructuredLogger("test_a")
106    logger.add_handler(handlers.StreamHandler(output, WptreportFormatter()))
107
108    with mock.patch.object(wptreport,
109                           'surrogate_replacement',
110                           wptreport.SurrogateReplacementUcs2()):
111        # output a bunch of stuff
112        logger.suite_start(["test-id-1"])  # no run_info arg!
113        logger.test_start("test-id-1")
114        logger.test_status("test-id-1",
115                           subtest=u"Name with surrogate\uD800",
116                           status="FAIL",
117                           message=u"\U0001F601 \uDE0A\uD83D \uD83D\uDE0A")
118        logger.test_end("test-id-1",
119                        status="PASS",
120                        message=u"\uDE0A\uD83D \uD83D\uDE0A \U0001F601")
121        logger.suite_end()
122
123    # check nothing got output to stdout/stderr
124    # (note that mozlog outputs exceptions during handling to stderr!)
125    captured = capfd.readouterr()
126    assert captured.out == ""
127    assert captured.err == ""
128
129    # check the actual output of the formatter
130    output.seek(0)
131    output_obj = json.load(output)
132    test = output_obj["results"][0]
133    assert test["message"] == u"U+de0aU+d83d \U0001f60a \U0001F601"
134    subtest = test["subtests"][0]
135    assert subtest["name"] == u"Name with surrogateU+d800"
136    assert subtest["message"] == u"\U0001F601 U+de0aU+d83d \U0001f60a"
137
138
139def test_wptreport_known_intermittent(capfd):
140    output = StringIO()
141    logger = structuredlog.StructuredLogger("test_a")
142    logger.add_handler(handlers.StreamHandler(output, WptreportFormatter()))
143
144    # output a bunch of stuff
145    logger.suite_start(["test-id-1"])  # no run_info arg!
146    logger.test_start("test-id-1")
147    logger.test_status("test-id-1",
148                       "a-subtest",
149                       status="FAIL",
150                       expected="PASS",
151                       known_intermittent=["FAIL"])
152    logger.test_end("test-id-1",
153                    status="OK",)
154    logger.suite_end()
155
156    # check nothing got output to stdout/stderr
157    # (note that mozlog outputs exceptions during handling to stderr!)
158    captured = capfd.readouterr()
159    assert captured.out == ""
160    assert captured.err == ""
161
162    # check the actual output of the formatter
163    output.seek(0)
164    output_obj = json.load(output)
165    test = output_obj["results"][0]
166    assert test["status"] == u"OK"
167    subtest = test["subtests"][0]
168    assert subtest["expected"] == u"PASS"
169    assert subtest["known_intermittent"] == [u'FAIL']
170
171
172def test_wptscreenshot_test_end(capfd):
173    formatter = WptscreenshotFormatter()
174
175    # Empty
176    data = {}
177    assert formatter.test_end(data) is None
178
179    # No items
180    data['extra'] = {"reftest_screenshots": []}
181    assert formatter.test_end(data) is None
182
183    # Invalid item
184    data['extra']['reftest_screenshots'] = ["no dict item"]
185    assert formatter.test_end(data) is None
186
187    # Random hash
188    data['extra']['reftest_screenshots'] = [{"hash": "HASH", "screenshot": "DATA"}]
189    assert 'data:image/png;base64,DATA\n' == formatter.test_end(data)
190
191    # Already cached hash
192    assert formatter.test_end(data) is None
193