testing/mozbase/mozlog/tests/test_structured.py@6474c204b198
testing/mozbase/mozlog/tests/test_structured.py
Wed, 31 Dec 2014 06:09:35 +0100
- author
- Michael Schloh von Bennewitz <michael@schloh.com>
- date
- Wed, 31 Dec 2014 06:09:35 +0100
- changeset 0
- 6474c204b198
- permissions
- -rw-r--r--
Cloned upstream origin tor-browser at tor-browser-31.3.0esr-4.5-1-build1
revision ID fc1c9ff7c1b2defdbc039f12214767608f46423f for hacking purpose.
1 import argparse
2 import os
3 import time
4 import unittest
5 import StringIO
6 import json
8 from mozlog.structured import (
9 commandline,
10 reader,
11 structuredlog,
12 )
15 class TestHandler(object):
16 def __init__(self):
17 self.last_item = None
19 def __call__(self, data):
20 self.last_item = data
23 class BaseStructuredTest(unittest.TestCase):
24 def setUp(self):
25 self.logger = structuredlog.StructuredLogger("test")
26 self.handler = TestHandler()
27 self.logger.add_handler(self.handler)
29 @property
30 def last_item(self):
31 return self.handler.last_item
33 def assert_log_equals(self, expected, actual=None):
34 if actual is None:
35 actual = self.last_item
37 all_expected = {"pid": os.getpid(),
38 "thread": "MainThread",
39 "source": "test"}
40 specials = set(["time"])
42 all_expected.update(expected)
43 for key, value in all_expected.iteritems():
44 self.assertEqual(actual[key], value)
46 self.assertEquals(set(all_expected.keys()) | specials, set(actual.keys()))
49 class TestStructuredLog(BaseStructuredTest):
50 def test_suite_start(self):
51 self.logger.suite_start(["test"])
52 self.assert_log_equals({"action": "suite_start",
53 "tests":["test"]})
55 def test_suite_end(self):
56 self.logger.suite_end()
57 self.assert_log_equals({"action": "suite_end"})
59 def test_start(self):
60 self.logger.test_start("test1")
61 self.assert_log_equals({"action": "test_start",
62 "test":"test1"})
64 self.logger.test_start(("test1", "==", "test1-ref"))
65 self.assert_log_equals({"action": "test_start",
66 "test":("test1", "==", "test1-ref")})
68 def test_status(self):
69 self.logger.test_status("test1", "subtest name", "fail", expected="FAIL", message="Test message")
70 self.assert_log_equals({"action": "test_status",
71 "subtest": "subtest name",
72 "status": "FAIL",
73 "message": "Test message",
74 "test":"test1"})
76 def test_status_1(self):
77 self.logger.test_status("test1", "subtest name", "fail")
78 self.assert_log_equals({"action": "test_status",
79 "subtest": "subtest name",
80 "status": "FAIL",
81 "expected": "PASS",
82 "test":"test1"})
84 def test_status_2(self):
85 self.assertRaises(ValueError, self.logger.test_status, "test1", "subtest name", "XXXUNKNOWNXXX")
87 def test_end(self):
88 self.logger.test_end("test1", "fail", message="Test message")
89 self.assert_log_equals({"action": "test_end",
90 "status": "FAIL",
91 "expected": "OK",
92 "message": "Test message",
93 "test":"test1"})
95 def test_end_1(self):
96 self.logger.test_end("test1", "PASS", expected="PASS", extra={"data":123})
97 self.assert_log_equals({"action": "test_end",
98 "status": "PASS",
99 "extra": {"data": 123},
100 "test":"test1"})
102 def test_end_2(self):
103 self.assertRaises(ValueError, self.logger.test_end, "test1", "XXXUNKNOWNXXX")
105 def test_process(self):
106 self.logger.process_output(1234, "test output")
107 self.assert_log_equals({"action": "process_output",
108 "process": 1234,
109 "data": "test output"})
111 def test_log(self):
112 for level in ["critical", "error", "warning", "info", "debug"]:
113 getattr(self.logger, level)("message")
114 self.assert_log_equals({"action": "log",
115 "level": level.upper(),
116 "message": "message"})
118 def test_logging_adapter(self):
119 import logging
120 logging.basicConfig(level="DEBUG")
121 old_level = logging.root.getEffectiveLevel()
122 logging.root.setLevel("DEBUG")
124 std_logger = logging.getLogger("test")
125 std_logger.setLevel("DEBUG")
127 logger = structuredlog.std_logging_adapter(std_logger)
129 try:
130 for level in ["critical", "error", "warning", "info", "debug"]:
131 getattr(logger, level)("message")
132 self.assert_log_equals({"action": "log",
133 "level": level.upper(),
134 "message": "message"})
135 finally:
136 logging.root.setLevel(old_level)
138 def test_add_remove_handlers(self):
139 handler = TestHandler()
140 self.logger.add_handler(handler)
141 self.logger.info("test1")
143 self.assert_log_equals({"action": "log",
144 "level": "INFO",
145 "message": "test1"})
147 self.assert_log_equals({"action": "log",
148 "level": "INFO",
149 "message": "test1"}, actual=handler.last_item)
151 self.logger.remove_handler(handler)
152 self.logger.info("test2")
154 self.assert_log_equals({"action": "log",
155 "level": "INFO",
156 "message": "test2"})
158 self.assert_log_equals({"action": "log",
159 "level": "INFO",
160 "message": "test1"}, actual=handler.last_item)
162 def test_wrapper(self):
163 file_like = structuredlog.StructuredLogFileLike(self.logger)
165 file_like.write("line 1")
167 self.assert_log_equals({"action": "log",
168 "level": "INFO",
169 "message": "line 1"})
171 file_like.write("line 2\n")
173 self.assert_log_equals({"action": "log",
174 "level": "INFO",
175 "message": "line 2"})
177 file_like.write("line 3\r")
179 self.assert_log_equals({"action": "log",
180 "level": "INFO",
181 "message": "line 3"})
183 file_like.write("line 4\r\n")
185 self.assert_log_equals({"action": "log",
186 "level": "INFO",
187 "message": "line 4"})
189 class TestCommandline(unittest.TestCase):
190 def test_setup_logging(self):
191 parser = argparse.ArgumentParser()
192 commandline.add_logging_group(parser)
193 args = parser.parse_args(["--log-raw=/tmp/foo"])
194 logger = commandline.setup_logging("test", args, {})
195 self.assertEqual(len(logger.handlers), 1)
197 class TestReader(unittest.TestCase):
198 def to_file_like(self, obj):
199 data_str = "\n".join(json.dumps(item) for item in obj)
200 return StringIO.StringIO(data_str)
202 def test_read(self):
203 data = [{"action": "action_0", "data": "data_0"},
204 {"action": "action_1", "data": "data_1"}]
206 f = self.to_file_like(data)
207 self.assertEquals(data, list(reader.read(f)))
209 def test_imap_log(self):
210 data = [{"action": "action_0", "data": "data_0"},
211 {"action": "action_1", "data": "data_1"}]
213 f = self.to_file_like(data)
215 def f_action_0(item):
216 return ("action_0", item["data"])
218 def f_action_1(item):
219 return ("action_1", item["data"])
221 res_iter = reader.imap_log(reader.read(f),
222 {"action_0": f_action_0,
223 "action_1": f_action_1})
224 self.assertEquals([("action_0", "data_0"), ("action_1", "data_1")],
225 list(res_iter))
227 def test_each_log(self):
228 data = [{"action": "action_0", "data": "data_0"},
229 {"action": "action_1", "data": "data_1"}]
231 f = self.to_file_like(data)
233 count = {"action_0":0,
234 "action_1":0}
236 def f_action_0(item):
237 count[item["action"]] += 1
239 def f_action_1(item):
240 count[item["action"]] += 2
242 reader.each_log(reader.read(f),
243 {"action_0": f_action_0,
244 "action_1": f_action_1})
246 self.assertEquals({"action_0":1, "action_1":2}, count)
248 def test_handler(self):
249 data = [{"action": "action_0", "data": "data_0"},
250 {"action": "action_1", "data": "data_1"}]
252 f = self.to_file_like(data)
254 test = self
255 class ReaderTestHandler(reader.LogHandler):
256 def __init__(self):
257 self.action_0_count = 0
258 self.action_1_count = 0
260 def action_0(self, item):
261 test.assertEquals(item["action"], "action_0")
262 self.action_0_count += 1
264 def action_1(self, item):
265 test.assertEquals(item["action"], "action_1")
266 self.action_1_count += 1
268 handler = ReaderTestHandler()
269 reader.handle_log(reader.read(f), handler)
271 self.assertEquals(handler.action_0_count, 1)
272 self.assertEquals(handler.action_1_count, 1)
274 if __name__ == "__main__":
275 unittest.main()
