"""

def mk_validator_testers(self, validator):

def p(i):

validator.validate(i)

def f(i):

self.assertRaises(bv.ValidationError, validator.validate, i)

return p, f # 'p(input)' if you expect it to pass, 'f(input)' if you expect it to fail.

def test_string_validator(self):

s = bv.String(min_length=1, max_length=5, pattern='[A-z]+')

# Not a string

self.assertRaises(bv.ValidationError, lambda: s.validate(1))

# Too short

self.assertRaises(bv.ValidationError, lambda: s.validate(''))

# Too long

self.assertRaises(bv.ValidationError, lambda: s.validate('a' * 6))

# Doesn't pass regex

self.assertRaises(bv.ValidationError, lambda: s.validate('#'))

# Passes

s.validate('a')

# Check that the validator is converting all strings to unicode

self.assertEqual(type(s.validate('a')), six.text_type)

def test_string_regex_anchoring(self):

p, f = self.mk_validator_testers(bv.String(pattern=r'abc|xyz'))

p('abc')

p('xyz')

f('_abc')

f('abc_')

f('_xyz')

f('xyz_')

def test_boolean_validator(self):

b = bv.Boolean()

b.validate(True)

b.validate(False)

self.assertRaises(bv.ValidationError, lambda: b.validate(1))

def test_integer_validator(self):

i = bv.UInt32(min_value=10, max_value=100)

# Not an integer

self.assertRaises(bv.ValidationError, lambda: i.validate(1.4))

# Too small

self.assertRaises(bv.ValidationError, lambda: i.validate(1))

# Too large

self.assertRaises(bv.ValidationError, lambda: i.validate(101))

# Passes

i.validate(50)

# min_value is less than the default for the type

self.assertRaises(AssertionError, lambda: bv.UInt32(min_value=-3))

# non-sensical min_value

self.assertRaises(AssertionError, lambda: bv.UInt32(min_value=1.3))

def test_float_validator(self):

f64 = bv.Float64()

# Too large for a float to represent

self.assertRaises(bv.ValidationError, lambda: f64.validate(10**310))

# inf and nan should be rejected

self.assertRaises(bv.ValidationError, lambda: f64.validate(float('nan')))

self.assertRaises(bv.ValidationError, lambda: f64.validate(float('inf')))

# Passes

f64.validate(1.1 * 10**300)

# Test a float64 with an additional bound

f64b = bv.Float64(min_value=0, max_value=100)

# Check bounds

self.assertRaises(bv.ValidationError, lambda: f64b.validate(1000))

self.assertRaises(bv.ValidationError, lambda: f64b.validate(-1))

# Test a float64 with an invalid bound

self.assertRaises(AssertionError, lambda: bv.Float64(min_value=0, max_value=10**330))

f32 = bv.Float32()

self.assertRaises(bv.ValidationError, lambda: f32.validate(3.5 * 10**38))

self.assertRaises(bv.ValidationError, lambda: f32.validate(-3.5 * 10**38))

# Passes

f32.validate(0)

def test_bytes_validator(self):

b = bv.Bytes(min_length=1, max_length=10)

# Not a valid binary type

self.assertRaises(bv.ValidationError, lambda: b.validate(u'asdf'))

# Too short

self.assertRaises(bv.ValidationError, lambda: b.validate(b''))

# Too long

self.assertRaises(bv.ValidationError, lambda: b.validate(b'\x00' * 11))

# Passes

b.validate(b'\x00')

def test_timestamp_validator(self):

class UTC(datetime.tzinfo):

def utcoffset(self, dt): # pylint: disable=unused-argument,useless-suppression

return datetime.timedelta(0)

def tzname(self, dt): # pylint: disable=unused-argument,useless-suppression

return 'UTC'

def dst(self, dt): # pylint: disable=unused-argument,useless-suppression

return datetime.timedelta(0)

class PST(datetime.tzinfo):

def utcoffset(self, dt): # pylint: disable=unused-argument,useless-suppression

return datetime.timedelta(-8)

def tzname(self, dt): # pylint: disable=unused-argument,useless-suppression

return 'PST'

def dst(self, dt): # pylint: disable=unused-argument,useless-suppression

return datetime.timedelta(0)

t = bv.Timestamp('%a, %d %b %Y %H:%M:%S +0000')

self.assertRaises(bv.ValidationError, lambda: t.validate('abcd'))

now = datetime.datetime.utcnow()

t.validate(now)

then = datetime.datetime(1776, 7, 4, 12, 0, 0)

t.validate(then)

new_then = json_decode(t, json_encode(t, then))

self.assertEqual(then, new_then)

# Accept a tzinfo only if it's UTC

t.validate(now.replace(tzinfo=UTC()))

# Do not accept a non-UTC tzinfo

self.assertRaises(bv.ValidationError,

lambda: t.validate(now.replace(tzinfo=PST())))

def test_list_validator(self):

l = bv.List(bv.String(), min_items=1, max_items=10)

# Not a valid list type

self.assertRaises(bv.ValidationError, lambda: l.validate('a'))

# Too short

self.assertRaises(bv.ValidationError, lambda: l.validate([]))

# Too long

self.assertRaises(bv.ValidationError, lambda: l.validate([1] * 11))

# Not a valid string type

self.assertRaises(bv.ValidationError, lambda: l.validate([1]))

# Passes

l.validate(['a'])

def test_map_validator(self):

m = bv.Map(bv.String(pattern="^foo.*"), bv.String(pattern=".*bar$"))

# applies validators of children

m.validate({"foo-one": "one-bar", "foo-two": "two-bar"})

# does not match regex

self.assertRaises(bv.ValidationError, lambda: m.validate({"one": "two"}))

# does not match declared types

self.assertRaises(bv.ValidationError, lambda: m.validate({1: 2}))

def test_nullable_validator(self):

n = bv.Nullable(bv.String())

# Absent case

n.validate(None)

# Fails string validation

self.assertRaises(bv.ValidationError, lambda: n.validate(123))

# Passes

n.validate('abc')

# Stacking nullables isn't supported by our JSON wire format

self.assertRaises(AssertionError,

lambda: bv.Nullable(bv.Nullable(bv.String())))

self.assertRaises(AssertionError,

lambda: bv.Nullable(bv.Void()))

def test_void_validator(self):

v = bv.Void()

# Passes: Only case that validates

v.validate(None)

# Fails validation

self.assertRaises(bv.ValidationError, lambda: v.validate(123))

def test_struct_validator(self):

class C(object):

_all_field_names_ = {'f'}

_all_fields_ = [('f', bv.String())]

f = None

s = bv.Struct(C)

self.assertRaises(bv.ValidationError, lambda: s.validate(object()))

def test_json_encoder(self):

self.assertEqual(json_encode(bv.Void(), None), json.dumps(None))

self.assertEqual(json_encode(bv.String(), 'abc'), json.dumps('abc'))

self.assertEqual(json_encode(bv.String(), u'\u2650'), json.dumps(u'\u2650'))

self.assertEqual(json_encode(bv.UInt32(), 123), json.dumps(123))

# Because a bool is a subclass of an int, ensure they aren't mistakenly

# encoded as a true/false in JSON when an integer is the data type.

self.assertEqual(json_encode(bv.UInt32(), True), json.dumps(1))

self.assertEqual(json_encode(bv.Boolean(), True), json.dumps(True))

f = '%a, %d %b %Y %H:%M:%S +0000'

now = datetime.datetime.utcnow()

self.assertEqual(json_encode(bv.Timestamp('%a, %d %b %Y %H:%M:%S +0000'), now),

json.dumps(now.strftime(f)))

b = b'\xff' * 5

self.assertEqual(json_encode(bv.Bytes(), b),

json.dumps(base64.b64encode(b).decode('ascii')))

self.assertEqual(json_encode(bv.Nullable(bv.String()), None), json.dumps(None))

self.assertEqual(json_encode(bv.Nullable(bv.String()), u'abc'), json.dumps('abc'))

def test_json_encoder_union(self):

# pylint: disable=attribute-defined-outside-init

class S(object):

_all_field_names_ = {'f'}

_all_fields_ = [('f', bv.String())]

class U(object):

# pylint: disable=no-member

_tagmap = {'a': bv.Int64(),

'b': bv.Void(),

'c': bv.Struct(S),

'd': bv.List(bv.Int64()),

'e': bv.Nullable(bv.Int64()),

'f': bv.Nullable(bv.Struct(S)),

'g': bv.Map(bv.String(), bv.String())}

_tag = None

def __init__(self, tag, value=None):

self._tag = tag

self._value = value

def get_a(self):

return self._a

def get_c(self):

return self._c

def get_d(self):

return self._d

U.b = U('b')

# Test primitive variant

u = U('a', 64)

self.assertEqual(json_encode(bv.Union(U), u, old_style=True),

json.dumps({'a': 64}))

# Test symbol variant

u = U('b')

self.assertEqual(json_encode(bv.Union(U), u, old_style=True),

json.dumps('b'))

# Test struct variant

c = S()

c.f = 'hello'

c._f_present = True

u = U('c', c)

self.assertEqual(json_encode(bv.Union(U), u, old_style=True),

json.dumps({'c': {'f': 'hello'}}))

# Test list variant

u = U('d', [1, 2, 3, 'a'])

# lists should be re-validated during serialization

self.assertRaises(bv.ValidationError, lambda: json_encode(bv.Union(U), u))

l = [1, 2, 3, 4]

u = U('d', [1, 2, 3, 4])

self.assertEqual(json_encode(bv.Union(U), u, old_style=True),

json.dumps({'d': l}))

# Test a nullable union

self.assertEqual(json_encode(bv.Nullable(bv.Union(U)), None),

json.dumps(None))

self.assertEqual(json_encode(bv.Nullable(bv.Union(U)), u, old_style=True),

json.dumps({'d': l}))

# Test nullable primitive variant

u = U('e', None)

self.assertEqual(json_encode(bv.Nullable(bv.Union(U)), u, old_style=True),

json.dumps('e'))

u = U('e', 64)

self.assertEqual(json_encode(bv.Nullable(bv.Union(U)), u, old_style=True),

json.dumps({'e': 64}))

# Test nullable composite variant

u = U('f', None)

self.assertEqual(json_encode(bv.Nullable(bv.Union(U)), u, old_style=True),

json.dumps('f'))

u = U('f', c)

self.assertEqual(json_encode(bv.Nullable(bv.Union(U)), u, old_style=True),

json.dumps({'f': {'f': 'hello'}}))

u = U('g', {'one': 2})

self.assertRaises(bv.ValidationError, lambda: json_encode(bv.Union(U), u))

m = {'one': 'two'}

u = U('g', m)

self.assertEqual(json_encode(bv.Union(U), u, old_style=True), json.dumps({'g': m}))

def test_json_encoder_error_messages(self):

# pylint: disable=attribute-defined-outside-init

class S3(object):

_all_field_names_ = {'j'}

_all_fields_ = [('j', bv.UInt64(max_value=10))]

class S2(object):

_all_field_names_ = {'i'}

_all_fields_ = [('i', bv.Struct(S3))]

class S(object):

_all_field_names_ = {'f'}

_all_fields_ = [('f', bv.Struct(S2))]

class U(object):

# pylint: disable=no-member

_tagmap = {'t': bv.Nullable(bv.Struct(S))}

_tag = None

_catch_all = None

def __init__(self, tag, value=None):

self._tag = tag

self._value = value

def get_t(self):

return self._t

s = S()

s.f = S2()

s._f_present = True

s.f.i = S3()

s.f._i_present = True

s.f.i._j_present = False

# Test that validation error references outer and inner struct

with self.assertRaises(bv.ValidationError):

try:

json_encode(bv.Struct(S), s)

except bv.ValidationError as e:

prefix = 'f.i: '

self.assertEqual(prefix, str(e)[:len(prefix)])

raise

u = U('t', s)

# Test that validation error references outer union and inner structs

with self.assertRaises(bv.ValidationError):

try:

json_encode(bv.Union(U), u)

except bv.ValidationError as e:

prefix = 't.f.i: '

self.assertEqual(prefix, str(e)[:len(prefix)])

raise

def test_json_decoder(self):

self.assertEqual(json_decode(bv.String(), json.dumps('abc')), 'abc')

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.String(), json.dumps(32)))

self.assertEqual(json_decode(bv.UInt32(), json.dumps(123)), 123)

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.UInt32(), json.dumps('hello')))

self.assertEqual(json_decode(bv.Boolean(), json.dumps(True)), True)

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Boolean(), json.dumps(1)))

f = '%a, %d %b %Y %H:%M:%S +0000'

now = datetime.datetime.utcnow().replace(microsecond=0)

self.assertEqual(json_decode(bv.Timestamp('%a, %d %b %Y %H:%M:%S +0000'),

json.dumps(now.strftime(f))),

now)

# Try decoding timestamp with bad type

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Timestamp('%a, %d %b %Y %H:%M:%S +0000'), '1'))

b = b'\xff' * 5

self.assertEqual(json_decode(bv.Bytes(),

json.dumps(base64.b64encode(b).decode('ascii'))),

b)

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Bytes(), json.dumps(1)))

self.assertEqual(json_decode(bv.Nullable(bv.String()), json.dumps(None)), None)

self.assertEqual(json_decode(bv.Nullable(bv.String()), json.dumps('abc')), 'abc')

self.assertEqual(json_decode(bv.Void(), json.dumps(None)), None)

# Check that void can take any input if strict is False.

self.assertEqual(json_decode(bv.Void(), json.dumps(12345), strict=False), None)

# Check that an error is raised if strict is True and there's a non-null value

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Void(), json.dumps(12345), strict=True))

def test_json_decoder_struct(self):

class S(object):

_all_field_names_ = {'f', 'g'}

_all_fields_ = [('f', bv.String()),

('g', bv.Nullable(bv.String()))]

_g = None

@property

def f(self):

return self._f

@f.setter

def f(self, val):

self._f = val # pylint: disable=attribute-defined-outside-init

@property

def g(self):

return self._g

@g.setter

def g(self, val):

self._g = val

# Required struct fields must be present

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Struct(S), json.dumps({})))

json_decode(bv.Struct(S), json.dumps({'f': 't'}))

# Struct fields can have null values for nullable fields

msg = json.dumps({'f': 't', 'g': None})

json_decode(bv.Struct(S), msg)

# Unknown struct fields raise error if strict

msg = json.dumps({'f': 't', 'z': 123})

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Struct(S), msg, strict=True))

json_decode(bv.Struct(S), msg, strict=False)

def test_json_decoder_union(self):

class S(object):

_all_field_names_ = {'f'}

_all_fields_ = [('f', bv.String())]

class U(object):

_tagmap = {'a': bv.Int64(),

'b': bv.Void(),

'c': bv.Struct(S),

'd': bv.List(bv.Int64()),

'e': bv.Nullable(bv.Int64()),

'f': bv.Nullable(bv.Struct(S)),

'g': bv.Void(),

'h': bv.Map(bv.String(), bv.String())}

_catch_all = 'g'

_tag = None

def __init__(self, tag, value=None):

self._tag = tag

self._value = value

def get_a(self):

return self._value

def get_c(self):

return self._value

def get_d(self):

return self._value

U.b = U('b')

# TODO: When run with Python 3, pylint thinks `u` is a `datetime`

# object. This results in spurious `no-member` errors, which we

# choose to ignore here (for now). This does not happen when

# running pylint with Python 2 (hence the useless-suppression).

# pylint: disable=no-member,useless-suppression

# Test primitive variant

u = json_decode(bv.Union(U), json.dumps({'a': 64}), old_style=True)

self.assertEqual(u.get_a(), 64)

# Test void variant

u = json_decode(bv.Union(U), json.dumps('b'))

self.assertEqual(u._tag, 'b')

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Union(U), json.dumps({'b': [1, 2]})))

u = json_decode(bv.Union(U), json.dumps({'b': [1, 2]}), strict=False, old_style=True)

self.assertEqual(u._tag, 'b')

# Test struct variant

u = json_decode(bv.Union(U), json.dumps({'c': {'f': 'hello'}}), old_style=True)

self.assertEqual(u.get_c().f, 'hello')

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Union(U), json.dumps({'c': [1, 2, 3]})))

# Test list variant

l = [1, 2, 3, 4]

u = json_decode(bv.Union(U), json.dumps({'d': l}), old_style=True)

self.assertEqual(u.get_d(), l)

# Test map variant

m = {'one': 'two', 'three': 'four'}

u = json_decode(bv.Union(U), json.dumps({'h': m}), old_style=True)

self.assertEqual(u.get_d(), m)

# Raises if unknown tag

self.assertRaises(bv.ValidationError, lambda: json_decode(bv.Union(U), json.dumps('z')))

# Unknown variant (strict=True)

self.assertRaises(bv.ValidationError,

lambda: json_decode(bv.Union(U), json.dumps({'z': 'test'})))

# Test catch all variant

u = json_decode(bv.Union(U), json.dumps({'z': 'test'}),

strict=False, old_style=True)

self.assertEqual(u._tag, 'g')

# Test nullable union

u = json_decode(bv.Nullable(bv.Union(U)), json.dumps(None),

strict=False, old_style=True)

self.assertEqual(u, None)

# Test nullable union member

u = json_decode(bv.Union(U), json.dumps('e'))

self.assertEqual(u._tag, 'e')

self.assertEqual(u._value, None)

u = json_decode(bv.Union(U), json.dumps({'e': 64}),

strict=False, old_style=True)

self.assertEqual(u._tag, 'e')

self.assertEqual(u._value, 64)

# Test nullable composite variant

u = json_decode(bv.Union(U), json.dumps('f'))

self.assertEqual(u._tag, 'f')

u = json_decode(bv.Union(U), json.dumps({'f': {'f': 'hello'}}),

strict=False, old_style=True)

self.assertEqual(type(u._value), S)

self.assertEqual(u._value.f, 'hello')

def test_json_decoder_error_messages(self):

class S3(object):

_all_field_names_ = {'j'}

_all_fields_ = [('j', bv.UInt64(max_value=10))]

class S2(object):

_all_field_names_ = {'i'}

_all_fields_ = [('i', bv.Struct(S3))]

class S(object):

_all_field_names_ = {'f'}

_all_fields_ = [('f', bv.Struct(S2))]

class U(object):

_tagmap = {'t': bv.Nullable(bv.Struct(S))}

_tag = None

_catch_all = None

def __init__(self, tag, value=None):

self._tag = tag

self._value = value

def get_t(self):

return self._value

# Test that validation error references outer and inner struct

with self.assertRaises(bv.ValidationError):

try:

json_decode(bv.Struct(S), json.dumps({'f': {'i': {}}}), strict=False)

except bv.ValidationError as e:

prefix = 'f.i: '

self.assertEqual(prefix, str(e)[:len(prefix)])

raise

# Test that validation error references outer union and inner structs

with self.assertRaises(bv.ValidationError):

try:

json_decode(bv.Union(U), json.dumps({'t': {'f': {'i': {}}}}),

strict=False, old_style=True)

except bv.ValidationError as e:

prefix = 't.f.i: '

self.assertEqual(prefix, str(e)[:len(prefix)])

def setUp(self):

# Sanity check: stone must be importable for the compiler to work

__import__('stone')

# Compile spec by calling out to stone

p = subprocess.Popen(

[sys.executable,

'-m',

'stone.cli',

'python_types',

'output',

'-'],

stdin=subprocess.PIPE,

stderr=subprocess.PIPE)

_, stderr = p.communicate(

input=(test_spec + test_ns2_spec).encode('utf-8'))

if p.wait() != 0:

raise AssertionError('Could not execute stone tool: %s' %

stderr.decode('utf-8'))

sys.path.append('output')

self.ns2 = __import__('ns2')

self.ns = __import__('ns')

self.sv = __import__('stone_validators')

self.ss = __import__('stone_serializers')

self.encode = self.ss.json_encode

self.compat_obj_encode = self.ss.json_compat_obj_encode

self.decode = self.ss.json_decode

self.compat_obj_decode = self.ss.json_compat_obj_decode

def test_docstring(self):

# Check that the docstrings from the spec have in some form made it

# into the Python docstrings for the generated objects.

self.assertIn('Sample struct doc.', self.ns.A.__doc__)

self.assertIn('Sample field doc.', self.ns.A.a.__doc__)

self.assertIn('Sample union doc.', self.ns.U.__doc__)

self.assertIn('Sample field doc.', self.ns.U.t0.__doc__)

# Test doc conversion of Python bool.

self.assertIn('``True``', self.ns.DocTest.b.__doc__)

# Test doc converts type reference to sphinx-friendly representation.

self.assertIn(':class:`D`', self.ns.DocTest.t.__doc__)

def test_aliases(self):

# The left is a validator, the right is the struct devs can use...

self.assertEqual(self.ns.AliasedS2, self.ns.S2)

def test_struct_decoding(self):

d = self.decode(self.sv.Struct(self.ns.D),

json.dumps({'a': 'A', 'b': 1, 'c': 'C', 'd': [], 'e': {}}))

self.assertIsInstance(d, self.ns.D)

self.assertEqual(d.a, 'A')

self.assertEqual(d.b, 1)

self.assertEqual(d.c, 'C')

self.assertEqual(d.d, [])

self.assertEqual(d.e, {})

# Test with missing value for nullable field

d = self.decode(self.sv.Struct(self.ns.D),

json.dumps({'a': 'A', 'b': 1, 'd': [], 'e': {}}))

self.assertEqual(d.a, 'A')

self.assertEqual(d.b, 1)

self.assertEqual(d.c, None)

self.assertEqual(d.d, [])

self.assertEqual(d.e, {})

# Test with missing value for field with default

d = self.decode(self.sv.Struct(self.ns.D),

json.dumps({'a': 'A', 'c': 'C', 'd': [1], 'e': {'one': 'two'}}))

self.assertEqual(d.a, 'A')

self.assertEqual(d.b, 10)

self.assertEqual(d.c, 'C')

self.assertEqual(d.d, [1])

self.assertEqual(d.e, {'one': 'two'})

# Test with explicitly null value for nullable field

d = self.decode(self.sv.Struct(self.ns.D),

json.dumps({'a': 'A', 'c': None, 'd': [1, 2], 'e': {'one': 'two'}}))

self.assertEqual(d.a, 'A')

self.assertEqual(d.c, None)

self.assertEqual(d.d, [1, 2])

self.assertEqual(d.e, {'one': 'two'})

# Test with None and numbers in list

d = self.decode(self.sv.Struct(self.ns.D),

json.dumps({'a': 'A',

'c': None,

'd': [None, 1],

'e': {'one': None, 'two': 'three'}}))

self.assertEqual(d.a, 'A')

self.assertEqual(d.c, None)

self.assertEqual(d.d, [None, 1])

self.assertEqual(d.e, {'one': None, 'two': 'three'})

# Test with explicitly null value for field with default

with self.assertRaises(self.sv.ValidationError) as cm:

self.decode(self.sv.Struct(self.ns.D),

json.dumps({'a': 'A', 'b': None}))

self.assertEqual("b: expected integer, got null", str(cm.exception))

def test_union_decoding_old(self):

v = self.decode(self.sv.Union(self.ns.V), json.dumps('t0'))

self.assertIsInstance(v, self.ns.V)

# Test verbose representation of a void union member

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'t0': None}), old_style=True)

self.assertIsInstance(v, self.ns.V)

# Test bad value for void union member

with self.assertRaises(self.sv.ValidationError) as cm:

self.decode(self.sv.Union(self.ns.V), json.dumps({'t0': 10}), old_style=True)

self.assertEqual("expected null, got integer", str(cm.exception))

# Test compact representation of a nullable union member with missing value

v = self.decode(self.sv.Union(self.ns.V), json.dumps('t2'))

self.assertIsInstance(v, self.ns.V)

# Test verbose representation of a nullable union member with missing value

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'t2': None}), old_style=True)

self.assertIsInstance(v, self.ns.V)

# Test verbose representation of a nullable union member with bad value

with self.assertRaises(self.sv.ValidationError) as cm:

self.decode(self.sv.Union(self.ns.V), json.dumps({'t2': 123}), old_style=True)

self.assertEqual("'123' expected to be a string, got integer", str(cm.exception))

def test_union_decoding(self):

v = self.decode(self.sv.Union(self.ns.V), json.dumps('t0'))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t0())

# Test verbose representation of a void union member

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't0'}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t0())

# Test extra verbose representation of a void union member

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't0', 't0': None}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t0())

# Test error: extra key

with self.assertRaises(self.sv.ValidationError) as cm:

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't0', 'unk': 123}))

self.assertEqual("unexpected key 'unk'", str(cm.exception))

# Test error: bad type

with self.assertRaises(self.sv.ValidationError) as cm:

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 123}))

self.assertEqual('tag must be string, got integer', str(cm.exception))

# Test primitive union member

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't1', 't1': 'hello'}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t1())

self.assertEqual(v.get_t1(), 'hello')

# Test catch-all

v = self.decode(

self.sv.Union(self.ns.V),

json.dumps({'.tag': 'z', 'z': 'hello'}),

strict=False)

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_other())

# Test catch-all is reject if strict

with self.assertRaises(self.sv.ValidationError):

self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 'z'}))

# Test explicitly using catch-all is reject if strict

with self.assertRaises(self.sv.ValidationError):

self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 'other'}))

# Test nullable primitive union member with null value

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't2', 't2': None}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t2())

self.assertEqual(v.get_t2(), None)

# Test nullable primitive union member that is missing

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't2'}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t2())

self.assertEqual(v.get_t2(), None)

# Test error: extra key

with self.assertRaises(self.sv.ValidationError) as cm:

self.decode(self.sv.Union(self.ns.V),

json.dumps({'.tag': 't2', 't2': None, 'unk': 123}))

self.assertEqual("unexpected key 'unk'", str(cm.exception))

# Test composite union member

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't3', 'f': 'hello'}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t3())

self.assertIsInstance(v.get_t3(), self.ns.S)

self.assertEqual(v.get_t3().f, 'hello')

# Test error: extra key

with self.assertRaises(self.sv.ValidationError) as cm:

self.decode(self.sv.Union(self.ns.V),

json.dumps({'.tag': 't3', 'f': 'hello', 'g': 'blah'}))

self.assertEqual("t3: unknown field 'g'", str(cm.exception))

# Test composite union member with unknown key, but strict is False

v = self.decode(self.sv.Union(self.ns.V),

json.dumps({'.tag': 't3', 'f': 'hello', 'g': 'blah'}),

strict=False)

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t3())

self.assertIsInstance(v.get_t3(), self.ns.S)

self.assertEqual(v.get_t3().f, 'hello')

# Test nullable composite union member

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't4', 'f': 'hello'}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t4())

self.assertIsInstance(v.get_t4(), self.ns.S)

self.assertEqual(v.get_t4().f, 'hello')

# Test nullable composite union member that's null

v = self.decode(self.sv.Union(self.ns.V), json.dumps({'.tag': 't4'}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t4())

self.assertIsNone(v.get_t4())

# Test stacked unions

v = self.decode(

self.sv.Union(self.ns.V),

json.dumps({'.tag': 't5', 't5': {'.tag': 't1', 't1': 'hello'}}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t5(), None)

self.assertIsInstance(v.get_t5(), self.ns.U)

u = v.get_t5()

self.assertEqual(u.get_t1(), 'hello')

# Test stacked unions with null

v = self.decode(

self.sv.Union(self.ns.V),

json.dumps({'.tag': 't6'}))

self.assertIsInstance(v, self.ns.V)

self.assertTrue(v.is_t6(), None)

self.assertEqual(v.get_t6(), None)

# Test member that enumerates subtypes

v = self.decode(

self.sv.Union(self.ns.V),

json.dumps({'.tag': 't7', 't7': {'.tag': 'file', 'name': 'test', 'size': 1024}}))

self.assertIsInstance(v, self.ns.V)

self.assertIsInstance(v.get_t7(), self.ns.File)

f = v.get_t7()

self.assertEqual(f.name, 'test')

self.assertEqual(f.size, 1024)