Bug 1340551 - [mozlog] Introduce concept of ContainerType in logtypes and allow nested containers, r=jgraham

Currently the List and Tuple DataTypes must specify what items they contain. But there's no way to specify
item types recursively, e.g List(Tuple(Int, Int)). Also the Dict type can't specify the item types it must
contain either. Dict is a special case because we may want to control both keys and values.

This patch formalizes a ContainerType (of which List, Tuple and Dict are subclasses).

MozReview-Commit-ID: Bouhy1DIAyD

--HG--
extra : rebase_source : e7b26f4411861fc3065b4b5b746f05172f70d455
This commit is contained in:
Andrew Halberstadt 2017-02-21 14:24:14 -05:00
parent f1ceb4a895
commit b5de92b51f
4 changed files with 165 additions and 25 deletions

View File

@ -2,6 +2,8 @@
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import inspect
convertor_registry = {}
missing = object()
no_default = object()
@ -123,6 +125,31 @@ class DataType(object):
(value, type(value).__name__, self.name, self.__class__.__name__))
class ContainerType(DataType):
"""A DataType that contains other DataTypes.
ContainerTypes must specify which other DataType they will contain. ContainerTypes
may contain other ContainerTypes.
Some examples:
List(Int, 'numbers')
Tuple((Unicode, Int, Any), 'things')
Dict(Unicode, 'config')
Dict({TestId: Status}, 'results')
Dict(List(Unicode), 'stuff')
"""
def __init__(self, item_type, name=None, **kwargs):
DataType.__init__(self, name, **kwargs)
self.item_type = self._format_item_type(item_type)
def _format_item_type(self, item_type):
if inspect.isclass(item_type):
return item_type(None)
return item_type
class Unicode(DataType):
def convert(self, data):
@ -163,19 +190,29 @@ class SubStatus(Status):
allowed = ["PASS", "FAIL", "ERROR", "TIMEOUT", "ASSERT", "NOTRUN", "SKIP"]
class Dict(DataType):
class Dict(ContainerType):
def _format_item_type(self, item_type):
superfmt = super(Dict, self)._format_item_type
if isinstance(item_type, dict):
if len(item_type) != 1:
raise ValueError("Dict item type specifier must contain a single entry.")
key_type, value_type = item_type.items()[0]
return superfmt(key_type), superfmt(value_type)
return Any(None), superfmt(item_type)
def convert(self, data):
return dict(data)
key_type, value_type = self.item_type
return {key_type.convert(k): value_type.convert(v) for k, v in dict(data).items()}
class List(DataType):
def __init__(self, name, item_type, default=no_default, optional=False):
DataType.__init__(self, name, default, optional)
self.item_type = item_type(None)
class List(ContainerType):
def convert(self, data):
# while dicts and strings _can_ be cast to lists, doing so is probably not intentional
if isinstance(data, (basestring, dict)):
raise ValueError("Expected list but got %s" % type(data))
return [self.item_type.convert(item) for item in data]
@ -191,14 +228,17 @@ class Any(DataType):
return data
class Tuple(DataType):
class Tuple(ContainerType):
def __init__(self, name, item_types, default=no_default, optional=False):
DataType.__init__(self, name, default, optional)
self.item_types = item_types
def _format_item_type(self, item_type):
superfmt = super(Tuple, self)._format_item_type
if isinstance(item_type, (tuple, list)):
return [superfmt(t) for t in item_type]
return (superfmt(item_type),)
def convert(self, data):
if len(data) != len(self.item_types):
raise ValueError("Expected %i items got %i" % (len(self.item_types), len(data)))
if len(data) != len(self.item_type):
raise ValueError("Expected %i items got %i" % (len(self.item_type), len(data)))
return tuple(item_type.convert(value)
for item_type, value in zip(self.item_types, data))
for item_type, value in zip(self.item_type, data))

View File

@ -256,11 +256,11 @@ class StructuredLogger(object):
self._state.suite_started = False
return True
@log_action(List("tests", Unicode),
Dict("run_info", default=None, optional=True),
Dict("version_info", default=None, optional=True),
Dict("device_info", default=None, optional=True),
Dict("extra", default=None, optional=True))
@log_action(List(Unicode, "tests"),
Dict(Any, "run_info", default=None, optional=True),
Dict(Any, "version_info", default=None, optional=True),
Dict(Any, "device_info", default=None, optional=True),
Dict(Any, "extra", default=None, optional=True))
def suite_start(self, data):
"""Log a suite_start message
@ -275,7 +275,7 @@ class StructuredLogger(object):
self._log_data("suite_start", data)
@log_action(Dict("extra", default=None, optional=True))
@log_action(Dict(Any, "extra", default=None, optional=True))
def suite_end(self, data):
"""Log a suite_end message"""
if not self._ensure_suite_state('suite_end', data):
@ -309,7 +309,7 @@ class StructuredLogger(object):
SubStatus("expected", default="PASS"),
Unicode("message", default=None, optional=True),
Unicode("stack", default=None, optional=True),
Dict("extra", default=None, optional=True))
Dict(Any, "extra", default=None, optional=True))
def test_status(self, data):
"""
Log a test_status message indicating a subtest result. Tests that
@ -340,7 +340,7 @@ class StructuredLogger(object):
Status("expected", default="OK"),
Unicode("message", default=None, optional=True),
Unicode("stack", default=None, optional=True),
Dict("extra", default=None, optional=True))
Dict(Any, "extra", default=None, optional=True))
def test_end(self, data):
"""
Log a test_end message indicating that a test completed. For tests
@ -389,7 +389,7 @@ class StructuredLogger(object):
Int("stackwalk_retcode", default=None, optional=True),
Unicode("stackwalk_stdout", default=None, optional=True),
Unicode("stackwalk_stderr", default=None, optional=True),
List("stackwalk_errors", Unicode, default=None))
List(Unicode, "stackwalk_errors", default=None))
def crash(self, data):
if data["stackwalk_errors"] is None:
data["stackwalk_errors"] = []
@ -397,7 +397,7 @@ class StructuredLogger(object):
self._log_data("crash", data)
@log_action(Unicode("primary", default=None),
List("secondary", Unicode, default=None))
List(Unicode, "secondary", default=None))
def valgrind_error(self, data):
self._log_data("valgrind_error", data)
@ -476,7 +476,7 @@ def _lint_func(level_name):
Unicode("hint", default=None, optional=True),
Unicode("source", default=None, optional=True),
Unicode("rule", default=None, optional=True),
Tuple("lineoffset", (Int, Int), default=None, optional=True),
Tuple((Int, Int), "lineoffset", default=None, optional=True),
Unicode("linter", default=None, optional=True))
def lint(self, data):
data["level"] = level_name

View File

@ -1,4 +1,5 @@
[DEFAULT]
subsuite = mozbase, os == "linux"
[test_logger.py]
[test_logtypes.py]
[test_structured.py]

View File

@ -0,0 +1,99 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import unittest
import mozunit
from mozlog.logtypes import (
Any,
Dict,
Int,
List,
Tuple,
Unicode,
)
class TestContainerTypes(unittest.TestCase):
def test_dict_type_basic(self):
d = Dict('name')
with self.assertRaises(ValueError):
d({'foo': 'bar'})
d = Dict(Any, 'name')
d({'foo': 'bar'}) # doesn't raise
def test_dict_type_with_dictionary_item_type(self):
d = Dict({Int: Int}, 'name')
with self.assertRaises(ValueError):
d({'foo': 1})
with self.assertRaises(ValueError):
d({1: 'foo'})
d({1: 2}) # doesn't raise
def test_dict_type_with_recursive_item_types(self):
d = Dict(Dict({Unicode: List(Int)}), 'name')
with self.assertRaises(ValueError):
d({'foo': 'bar'})
with self.assertRaises(ValueError):
d({'foo': {'bar': 'baz'}})
with self.assertRaises(ValueError):
d({'foo': {'bar': ['baz']}})
d({'foo': {'bar': [1]}}) # doesn't raise
def test_list_type_basic(self):
l = List('name')
with self.assertRaises(ValueError):
l(['foo'])
l = List(Any, 'name')
l(['foo', 1]) # doesn't raise
def test_list_type_with_recursive_item_types(self):
l = List(Dict(List(Tuple((Unicode, Int)))), 'name')
with self.assertRaises(ValueError):
l(['foo'])
with self.assertRaises(ValueError):
l([{'foo': 'bar'}])
with self.assertRaises(ValueError):
l([{'foo': ['bar']}])
l([{'foo': [('bar', 1)]}]) # doesn't raise
def test_tuple_type_basic(self):
t = Tuple('name')
with self.assertRaises(ValueError):
t((1,))
t = Tuple(Any, 'name')
t((1,)) # doesn't raise
def test_tuple_type_with_tuple_item_type(self):
t = Tuple((Unicode, Int))
with self.assertRaises(ValueError):
t(('foo', 'bar'))
t(('foo', 1)) # doesn't raise
def test_tuple_type_with_recursive_item_types(self):
t = Tuple((Dict(List(Any)), List(Dict(Any)), Unicode), 'name')
with self.assertRaises(ValueError):
t(({'foo': 'bar'}, [{'foo': 'bar'}], 'foo'))
with self.assertRaises(ValueError):
t(({'foo': ['bar']}, ['foo'], 'foo'))
t(({'foo': ['bar']}, [{'foo': 'bar'}], 'foo')) # doesn't raise
if __name__ == '__main__':
mozunit.main()