summaryrefslogtreecommitdiffstats
path: root/testing/mochitest/tests/python/test_message_logger.py
diff options
context:
space:
mode:
Diffstat (limited to 'testing/mochitest/tests/python/test_message_logger.py')
-rw-r--r--testing/mochitest/tests/python/test_message_logger.py191
1 files changed, 191 insertions, 0 deletions
diff --git a/testing/mochitest/tests/python/test_message_logger.py b/testing/mochitest/tests/python/test_message_logger.py
new file mode 100644
index 0000000000..60bf6f9dc9
--- /dev/null
+++ b/testing/mochitest/tests/python/test_message_logger.py
@@ -0,0 +1,191 @@
+# Any copyright is dedicated to the Public Domain.
+# http://creativecommons.org/publicdomain/zero/1.0/
+
+import json
+import time
+import types
+
+import mozunit
+import pytest
+import six
+from conftest import setup_args
+from mozlog.formatters import JSONFormatter
+from mozlog.handlers.base import StreamHandler
+from mozlog.structuredlog import StructuredLogger
+from six import string_types
+
+
+@pytest.fixture
+def logger():
+ logger = StructuredLogger("mochitest_message_logger")
+
+ buf = six.StringIO()
+ handler = StreamHandler(buf, JSONFormatter())
+ logger.add_handler(handler)
+ return logger
+
+
+@pytest.fixture
+def get_message_logger(setup_test_harness, logger):
+ setup_test_harness(*setup_args)
+ runtests = pytest.importorskip("runtests")
+
+ def fake_message(self, action, **extra):
+ message = {
+ "action": action,
+ "time": time.time(),
+ }
+ if action in ("test_start", "test_end", "test_status"):
+ message["test"] = "test_foo.html"
+
+ if action == "test_end":
+ message["status"] = "PASS"
+ message["expected"] = "PASS"
+
+ elif action == "test_status":
+ message["subtest"] = "bar"
+ message["status"] = "PASS"
+
+ elif action == "log":
+ message["level"] = "INFO"
+ message["message"] = "foobar"
+
+ message.update(**extra)
+ return self.write(json.dumps(message))
+
+ def inner(**kwargs):
+ ml = runtests.MessageLogger(logger, **kwargs)
+
+ # Create a convenience function for faking incoming log messages.
+ ml.fake_message = types.MethodType(fake_message, ml)
+ return ml
+
+ return inner
+
+
+@pytest.fixture
+def get_lines(logger):
+ buf = logger.handlers[0].stream
+
+ def inner():
+ lines = buf.getvalue().splitlines()
+ buf.truncate(0)
+ # Python3 will not reposition the buffer position after
+ # truncate and will extend the buffer with null bytes.
+ # Force the buffer position to the start of the buffer
+ # to prevent null bytes from creeping in.
+ buf.seek(0)
+ return lines
+
+ return inner
+
+
+@pytest.fixture
+def assert_actions(get_lines):
+ def inner(expected):
+ if isinstance(expected, string_types):
+ expected = [expected]
+
+ lines = get_lines()
+ actions = [json.loads(l)["action"] for l in lines]
+ assert actions == expected
+
+ return inner
+
+
+def test_buffering_on(get_message_logger, assert_actions):
+ ml = get_message_logger(buffering=True)
+
+ # no buffering initially (outside of test)
+ ml.fake_message("log")
+ assert_actions(["log"])
+
+ # inside a test buffering is enabled, only 'test_start' logged
+ ml.fake_message("test_start")
+ ml.fake_message("test_status")
+ ml.fake_message("log")
+ assert_actions(["test_start"])
+
+ # buffering turned off manually within a test
+ ml.fake_message("buffering_off")
+ ml.fake_message("test_status")
+ ml.fake_message("log")
+ assert_actions(["test_status", "log"])
+
+ # buffering turned back on again
+ ml.fake_message("buffering_on")
+ ml.fake_message("test_status")
+ ml.fake_message("log")
+ assert_actions([])
+
+ # test end, it failed! All previsouly buffered messages are now logged.
+ ml.fake_message("test_end", status="FAIL")
+ assert_actions(
+ [
+ "log", # "Buffered messages logged"
+ "test_status",
+ "log",
+ "test_status",
+ "log",
+ "log", # "Buffered messages finished"
+ "test_end",
+ ]
+ )
+
+ # enabling buffering outside of a test has no affect
+ ml.fake_message("buffering_on")
+ ml.fake_message("log")
+ ml.fake_message("test_status")
+ assert_actions(["log", "test_status"])
+
+
+def test_buffering_off(get_message_logger, assert_actions):
+ ml = get_message_logger(buffering=False)
+
+ ml.fake_message("test_start")
+ assert_actions(["test_start"])
+
+ # messages logged no matter what the state
+ ml.fake_message("test_status")
+ ml.fake_message("buffering_off")
+ ml.fake_message("log")
+ assert_actions(["test_status", "log"])
+
+ # even after a 'buffering_on' action
+ ml.fake_message("buffering_on")
+ ml.fake_message("test_status")
+ ml.fake_message("log")
+ assert_actions(["test_status", "log"])
+
+ # no buffer to empty on test fail
+ ml.fake_message("test_end", status="FAIL")
+ assert_actions(["test_end"])
+
+
+@pytest.mark.parametrize(
+ "name,expected",
+ (
+ ("/tests/test_foo.html", "test_foo.html"),
+ ("chrome://mochitests/content/a11y/test_foo.html", "test_foo.html"),
+ ("chrome://mochitests/content/browser/test_foo.html", "test_foo.html"),
+ ("chrome://mochitests/content/chrome/test_foo.html", "test_foo.html"),
+ (
+ "https://example.org:443/tests/netwerk/test_foo.html",
+ "netwerk/test_foo.html",
+ ),
+ ("http://mochi.test:8888/tests/test_foo.html", "test_foo.html"),
+ ("http://mochi.test:8888/content/dom/browser/test_foo.html", None),
+ ),
+)
+def test_test_names_fixed_to_be_relative(name, expected, get_message_logger, get_lines):
+ ml = get_message_logger(buffering=False)
+ ml.fake_message("test_start", test=name)
+ lines = get_lines()
+
+ if expected is None:
+ expected = name
+ assert json.loads(lines[0])["test"] == expected
+
+
+if __name__ == "__main__":
+ mozunit.main()