Differential D5928 Diff 29382 python_modules/dagster/dagster_tests/general_tests/test_serdes.py

Changeset View

Standalone View

python_modules/dagster/dagster_tests/general_tests/test_serdes.py

Show First 20 Lines • Show All 106 Lines • ▼ Show 20 Lines	def test_forward_compat_serdes_new_enum_field():

unpacked = _unpack_value(packed, whitelist_map=_TEST_WHITELIST_MAP)		unpacked = _unpack_value(packed, whitelist_map=_TEST_WHITELIST_MAP)

assert unpacked != corge		assert unpacked != corge
assert unpacked.name == corge.name		assert unpacked.name == corge.name
assert unpacked.value == corge.value		assert unpacked.value == corge.value


# This behavior isn't possible on 2.7 because of `inspect` limitations
@pytest.mark.skipif(sys.version_info < (3,), reason="This behavior isn't available on 2.7")
def test_backward_compat_serdes():		def test_backward_compat_serdes():
_TEST_WHITELIST_MAP = _initial_whitelist_map()		_TEST_WHITELIST_MAP = _initial_whitelist_map()

@_whitelist_for_serdes(whitelist_map=_TEST_WHITELIST_MAP)		@_whitelist_for_serdes(whitelist_map=_TEST_WHITELIST_MAP)
class Quux(namedtuple("_Quux", "foo bar baz")):		class Quux(namedtuple("_Quux", "foo bar baz")):
def __new__(cls, foo, bar, baz):		def __new__(cls, foo, bar, baz):
return super(Quux, cls).__new__(cls, foo, bar, baz) # pylint: disable=bad-super-call		return super(Quux, cls).__new__(cls, foo, bar, baz) # pylint: disable=bad-super-call

Show All 18 Lines


def serdes_test_class(klass):		def serdes_test_class(klass):
_TEST_WHITELIST_MAP = _initial_whitelist_map()		_TEST_WHITELIST_MAP = _initial_whitelist_map()

return _whitelist_for_serdes(whitelist_map=_TEST_WHITELIST_MAP)(klass)		return _whitelist_for_serdes(whitelist_map=_TEST_WHITELIST_MAP)(klass)


@pytest.mark.skipif(
sys.version_info.major < 3, reason="Serdes declaration time checks python 3 only"
)
def test_wrong_first_arg():		def test_wrong_first_arg():
with pytest.raises(SerdesClassUsageError) as exc_info:		with pytest.raises(SerdesClassUsageError) as exc_info:

@serdes_test_class		@serdes_test_class
class NotCls(namedtuple("NotCls", "field_one field_two")):		class NotCls(namedtuple("NotCls", "field_one field_two")):
def __new__(not_cls, field_two, field_one):		def __new__(not_cls, field_two, field_one):
return super(NotCls, not_cls).__new__(field_one, field_two)		return super(NotCls, not_cls).__new__(field_one, field_two)

assert (		assert (
str(exc_info.value)		str(exc_info.value)
== 'For namedtuple NotCls: First parameter must be _cls or cls. Got "not_cls".'		== 'For namedtuple NotCls: First parameter must be _cls or cls. Got "not_cls".'
)		)


@pytest.mark.skipif(
sys.version_info.major < 3, reason="Serdes declaration time checks python 3 only"
)
def test_incorrect_order():		def test_incorrect_order():
with pytest.raises(SerdesClassUsageError) as exc_info:		with pytest.raises(SerdesClassUsageError) as exc_info:

@serdes_test_class		@serdes_test_class
class WrongOrder(namedtuple("WrongOrder", "field_one field_two")):		class WrongOrder(namedtuple("WrongOrder", "field_one field_two")):
def __new__(cls, field_two, field_one):		def __new__(cls, field_two, field_one):
return super(WrongOrder, cls).__new__(field_one, field_two)		return super(WrongOrder, cls).__new__(field_one, field_two)

assert str(exc_info.value) == (		assert str(exc_info.value) == (
"For namedtuple WrongOrder: "		"For namedtuple WrongOrder: "
"Params to __new__ must match the order of field declaration "		"Params to __new__ must match the order of field declaration "
"in the namedtuple. Declared field number 1 in the namedtuple "		"in the namedtuple. Declared field number 1 in the namedtuple "
'is "field_one". Parameter 1 in __new__ method is "field_two".'		'is "field_one". Parameter 1 in __new__ method is "field_two".'
)		)


@pytest.mark.skipif(
sys.version_info.major < 3, reason="Serdes declaration time checks python 3 only"
)
def test_missing_one_parameter():		def test_missing_one_parameter():
with pytest.raises(SerdesClassUsageError) as exc_info:		with pytest.raises(SerdesClassUsageError) as exc_info:

@serdes_test_class		@serdes_test_class
class MissingFieldInNew(namedtuple("MissingFieldInNew", "field_one field_two field_three")):		class MissingFieldInNew(namedtuple("MissingFieldInNew", "field_one field_two field_three")):
def __new__(cls, field_one, field_two):		def __new__(cls, field_one, field_two):
return super(MissingFieldInNew, cls).__new__(field_one, field_two, None)		return super(MissingFieldInNew, cls).__new__(field_one, field_two, None)

assert str(exc_info.value) == (		assert str(exc_info.value) == (
"For namedtuple MissingFieldInNew: "		"For namedtuple MissingFieldInNew: "
"Missing parameters to __new__. You have declared fields in "		"Missing parameters to __new__. You have declared fields in "
"the named tuple that are not present as parameters to the "		"the named tuple that are not present as parameters to the "
"to the __new__ method. In order for both serdes serialization "		"to the __new__ method. In order for both serdes serialization "
"and pickling to work, these must match. Missing: ['field_three']"		"and pickling to work, these must match. Missing: ['field_three']"
)		)


@pytest.mark.skipif(
sys.version_info.major < 3, reason="Serdes declaration time checks python 3 only"
)
def test_missing_many_parameters():		def test_missing_many_parameters():
with pytest.raises(SerdesClassUsageError) as exc_info:		with pytest.raises(SerdesClassUsageError) as exc_info:

@serdes_test_class		@serdes_test_class
class MissingFieldsInNew(		class MissingFieldsInNew(
namedtuple("MissingFieldsInNew", "field_one field_two field_three, field_four")		namedtuple("MissingFieldsInNew", "field_one field_two field_three, field_four")
):		):
def __new__(cls, field_one, field_two):		def __new__(cls, field_one, field_two):
▲ Show 20 Lines • Show All 164 Lines • Show Last 20 Lines