test_message_logger.py (5363B)
1 # Any copyright is dedicated to the Public Domain. 2 # http://creativecommons.org/publicdomain/zero/1.0/ 3 4 import io 5 import json 6 import time 7 import types 8 9 import mozunit 10 import pytest 11 from conftest import setup_args 12 from mozlog.formatters import JSONFormatter 13 from mozlog.handlers.base import StreamHandler 14 from mozlog.structuredlog import StructuredLogger 15 16 17 @pytest.fixture 18 def logger(): 19 logger = StructuredLogger("mochitest_message_logger") 20 21 buf = io.StringIO() 22 handler = StreamHandler(buf, JSONFormatter()) 23 logger.add_handler(handler) 24 return logger 25 26 27 @pytest.fixture 28 def get_message_logger(setup_test_harness, logger): 29 setup_test_harness(*setup_args) 30 runtests = pytest.importorskip("runtests") 31 32 def fake_message(self, action, **extra): 33 message = { 34 "action": action, 35 "time": time.time(), 36 } 37 if action in ("test_start", "test_end", "test_status"): 38 message["test"] = "test_foo.html" 39 40 if action == "test_end": 41 message["status"] = "PASS" 42 message["expected"] = "PASS" 43 44 elif action == "test_status": 45 message["subtest"] = "bar" 46 message["status"] = "PASS" 47 48 elif action == "log": 49 message["level"] = "INFO" 50 message["message"] = "foobar" 51 52 message.update(**extra) 53 return self.write(json.dumps(message)) 54 55 def inner(**kwargs): 56 ml = runtests.MessageLogger(logger, **kwargs) 57 58 # Create a convenience function for faking incoming log messages. 59 ml.fake_message = types.MethodType(fake_message, ml) 60 return ml 61 62 return inner 63 64 65 @pytest.fixture 66 def get_lines(logger): 67 buf = logger.handlers[0].stream 68 69 def inner(): 70 lines = buf.getvalue().splitlines() 71 buf.truncate(0) 72 # Python3 will not reposition the buffer position after 73 # truncate and will extend the buffer with null bytes. 74 # Force the buffer position to the start of the buffer 75 # to prevent null bytes from creeping in. 76 buf.seek(0) 77 return lines 78 79 return inner 80 81 82 @pytest.fixture 83 def assert_actions(get_lines): 84 def inner(expected): 85 if isinstance(expected, str): 86 expected = [expected] 87 88 lines = get_lines() 89 actions = [json.loads(l)["action"] for l in lines] 90 assert actions == expected 91 92 return inner 93 94 95 def test_buffering_on(get_message_logger, assert_actions): 96 ml = get_message_logger(buffering=True) 97 98 # no buffering initially (outside of test) 99 ml.fake_message("log") 100 assert_actions(["log"]) 101 102 # inside a test buffering is enabled, only 'test_start' logged 103 ml.fake_message("test_start") 104 ml.fake_message("test_status") 105 ml.fake_message("log") 106 assert_actions(["test_start"]) 107 108 # buffering turned off manually within a test 109 ml.fake_message("buffering_off") 110 ml.fake_message("test_status") 111 ml.fake_message("log") 112 assert_actions(["test_status", "log"]) 113 114 # buffering turned back on again 115 ml.fake_message("buffering_on") 116 ml.fake_message("test_status") 117 ml.fake_message("log") 118 assert_actions([]) 119 120 # test end, it failed! All previsouly buffered messages are now logged. 121 ml.fake_message("test_end", status="FAIL") 122 assert_actions([ 123 "log", # "Buffered messages logged" 124 "test_status", 125 "log", 126 "test_status", 127 "log", 128 "log", # "Buffered messages finished" 129 "test_end", 130 ]) 131 132 # enabling buffering outside of a test has no affect 133 ml.fake_message("buffering_on") 134 ml.fake_message("log") 135 ml.fake_message("test_status") 136 assert_actions(["log", "test_status"]) 137 138 139 def test_buffering_off(get_message_logger, assert_actions): 140 ml = get_message_logger(buffering=False) 141 142 ml.fake_message("test_start") 143 assert_actions(["test_start"]) 144 145 # messages logged no matter what the state 146 ml.fake_message("test_status") 147 ml.fake_message("buffering_off") 148 ml.fake_message("log") 149 assert_actions(["test_status", "log"]) 150 151 # even after a 'buffering_on' action 152 ml.fake_message("buffering_on") 153 ml.fake_message("test_status") 154 ml.fake_message("log") 155 assert_actions(["test_status", "log"]) 156 157 # no buffer to empty on test fail 158 ml.fake_message("test_end", status="FAIL") 159 assert_actions(["test_end"]) 160 161 162 @pytest.mark.parametrize( 163 "name,expected", 164 ( 165 ("/tests/test_foo.html", "test_foo.html"), 166 ("chrome://mochitests/content/a11y/test_foo.html", "test_foo.html"), 167 ("chrome://mochitests/content/browser/test_foo.html", "test_foo.html"), 168 ("chrome://mochitests/content/chrome/test_foo.html", "test_foo.html"), 169 ( 170 "https://example.org:443/tests/netwerk/test_foo.html", 171 "netwerk/test_foo.html", 172 ), 173 ("http://mochi.test:8888/tests/test_foo.html", "test_foo.html"), 174 ("http://mochi.test:8888/content/dom/browser/test_foo.html", None), 175 ), 176 ) 177 def test_test_names_fixed_to_be_relative(name, expected, get_message_logger, get_lines): 178 ml = get_message_logger(buffering=False) 179 ml.fake_message("test_start", test=name) 180 lines = get_lines() 181 182 if expected is None: 183 expected = name 184 assert json.loads(lines[0])["test"] == expected 185 186 187 if __name__ == "__main__": 188 mozunit.main()