#!/usr/bin/env python

from __future__ import absolute_import, division, print_function, unicode_literals

import base64

import datetime

import json

import shutil

import six

import subprocess

import sys

import unittest

import stone.backends.python_rsrc.stone_validators as bv

from stone.backends.python_rsrc.stone_serializers import (

CallerPermissionsInterface,

json_encode,

json_decode,

_strftime as stone_strftime,

)

class TestDropInModules(unittest.TestCase):

"""

Tests the stone_serializers and stone_validators modules.

"""

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):

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

# Not a valid list type

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

# Too short

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

# Too long

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

# Not a valid string type

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

# Passes

l1.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

@classmethod

def _is_tag_present(cls, tag, cp):

assert cp

if tag in cls._tagmap:

return True

return False

@classmethod

def _get_val_data_type(cls, tag, cp):

assert cp

return cls._tagmap[tag]

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))

l1 = [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': l1}))

# 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': l1}))

# 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

@classmethod

def _is_tag_present(cls, tag, cp):

assert cp

if tag in cls._tagmap:

return True

return False

@classmethod

def _get_val_data_type(cls, tag, cp):

assert cp

return cls._tagmap[tag]

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

@classmethod

def _is_tag_present(cls, tag, cp):

assert cp

if tag in cls._tagmap:

return True

return False

@classmethod

def _get_val_data_type(cls, tag, cp):

assert cp

return cls._tagmap[tag]

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

l1 = [1, 2, 3, 4]

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

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

# 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

@classmethod

def _is_tag_present(cls, tag, cp):

assert cp

if tag in cls._tagmap:

return True

return False

@classmethod

def _get_val_data_type(cls, tag, cp):

assert cp

return cls._tagmap[tag]

# 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)])

raise

test_spec = """\

namespace ns

import ns2

struct A

"Sample struct doc."

a String

"Sample field doc."

b Int64

struct B extends A

c Bytes

struct C extends B

d Float64

struct D

a String

b UInt64 = 10

c String?

d List(Int64?)

e Map(String, String?)

struct E

a String = "test"

b UInt64 = 10

c Int64?

struct DocTest

b Boolean

"If :val:`true` then..."

t String

"References :type:`D`."

union_closed U

"Sample union doc."

t0

"Sample field doc."

t1 String

t2

union UOpen extends U

t3

union_closed UExtend extends U

t3

union_closed UExtend2 extends U

t3

union_closed UExtendExtend extends UExtend

t4

union V

t0

t1 String

t2 String?

t3 S

t4 S?

t5 U

t6 U?

t7 Resource

t8 Resource?

t9 List(String)

t10 List(U)

t11 Map(String, Int32)

t12 Map(String, U)

struct S

f String

struct Resource

union_closed

file File

folder Folder

name String

struct File extends Resource

size UInt64

struct Folder extends Resource

"Regular folder"

# Differs from Resource because it's a catch-all

struct ResourceLax

union

file File2

folder Folder2

name String

struct File2 extends ResourceLax

size UInt64

struct Folder2 extends ResourceLax

"Regular folder"

struct ImportTestS extends ns2.BaseS

a String

union_closed ImportTestU extends ns2.BaseU

a UInt64

union_closed U2

a

b OptionalS

struct OptionalS

f1 String = "hello"

f2 UInt64 = 3

struct S2

f1 OptionalS

alias AliasedS2 = S2

alias AliasedString = String(max_length=10)

struct ContainsAlias

s AliasedString

struct S3

u ns2.BaseU = z

"""

test_ns2_spec = """\

namespace ns2

struct BaseS

"This is a test \u2650"

z Int64

union_closed BaseU

z

x String

alias AliasedBaseU = BaseU

"""

class TestGeneratedPython(unittest.TestCase):

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'}))