def test_lp_serialize():

    @task
    def t1(a: int) -> typing.NamedTuple('OutputsBC', t1_int_output=int, c=str):
        a = (a + 2)
        return (a, ('world-' + str(a)))

    @task
    def t2(a: str, b: str) -> str:
        return (b + a)

    @workflow
    def my_subwf(a: int) -> (str, str):
        (x, y) = t1(a=a)
        (u, v) = t1(a=x)
        return (y, v)
    lp = launch_plan.LaunchPlan.create('serialize_test1', my_subwf)
    lp_with_defaults = launch_plan.LaunchPlan.create('serialize_test2', my_subwf, default_inputs={'a': 3})
    serialization_settings = context_manager.SerializationSettings(project='proj', domain='dom', version='123', image_config=ImageConfig(Image(name='name', fqn='asdf/fdsa', tag='123')), env={})
    sdk_lp = get_serializable(serialization_settings, lp)
    assert (len(sdk_lp.default_inputs.parameters) == 0)
    assert (len(sdk_lp.fixed_inputs.literals) == 0)
    sdk_lp = get_serializable(serialization_settings, lp_with_defaults)
    assert (len(sdk_lp.default_inputs.parameters) == 1)
    assert (len(sdk_lp.fixed_inputs.literals) == 0)
    parameter_a = sdk_lp.default_inputs.parameters['a']
    parameter_a = Parameter.from_flyte_idl(parameter_a.to_flyte_idl())
    assert (parameter_a.default is not None)
def test_lp_serialize():

    @task
    def t1(a: int) -> typing.NamedTuple('OutputsBC', t1_int_output=int, c=str):
        a = (a + 2)
        return (a, ('world-' + str(a)))

    @task
    def t2(a: str, b: str) -> str:
        return (b + a)

    @workflow
    def my_subwf(a: int) -> (str, str):
        (x, y) = t1(a=a)
        (u, v) = t1(a=x)
        return (y, v)
    lp = launch_plan.LaunchPlan.create('serialize_test1', my_subwf)
    lp_with_defaults = launch_plan.LaunchPlan.create('serialize_test2', my_subwf, default_inputs={'a': 3})
    serialization_settings = context_manager.SerializationSettings(project='proj', domain='dom', version='123', image_config=ImageConfig(Image(name='name', fqn='asdf/fdsa', tag='123')), env={})
    sdk_lp = get_serializable(serialization_settings, lp)
    assert (len(sdk_lp.default_inputs.parameters) == 0)
    assert (len(sdk_lp.fixed_inputs.literals) == 0)
    sdk_lp = get_serializable(serialization_settings, lp_with_defaults)
    assert (len(sdk_lp.default_inputs.parameters) == 1)
    assert (len(sdk_lp.fixed_inputs.literals) == 0)
    parameter_a = sdk_lp.default_inputs.parameters['a']
    parameter_a = Parameter.from_flyte_idl(parameter_a.to_flyte_idl())
    assert (parameter_a.default is not None)