How to use the flytekit.sdk.workflow.Input function in flytekit

def test_fixed_inputs():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Integer),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        }
    )
    lp = workflow_to_test.create_launch_plan(
        fixed_inputs={'required_input': 4}
    )
    assert len(lp.fixed_inputs.literals) == 1
    assert lp.fixed_inputs.literals['required_input'].scalar.primitive.integer == 4
    assert len(lp.default_inputs.parameters) == 1
    assert lp.default_inputs.parameters['default_input'].default.scalar.primitive.integer == 5

command=["sh", "-c", "echo $(( {{.Inputs.val}} * {{.Inputs.val}} )) | tee /var/outputs/out"],
)

sum = SdkRawContainerTask(
    input_data_dir="/var/flyte/inputs",
    output_data_dir="/var/flyte/outputs",
    inputs={"x": Types.Integer, "y": Types.Integer},
    outputs={"out": Types.Integer},
    image="alpine",
    command=["sh", "-c", "echo $(( {{.Inputs.x}} + {{.Inputs.y}} )) | tee /var/flyte/outputs/out"],
)


@workflow_class
class RawContainerWorkflow(object):
    val1 = Input(Types.Integer)
    val2 = Input(Types.Integer)
    sq1 = square(val=val1)
    sq2 = square(val=val2)
    sm = sum(x=sq1.outputs.out, y=sq2.outputs.out)
    sum_of_squares = Output(sm.outputs.out, sdk_type=Types.Integer)

def test_labels():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Integer),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        }
    )
    lp = workflow_to_test.create_launch_plan(
        fixed_inputs={'required_input': 5},
        schedule=_schedules.CronSchedule("* * ? * * *"),
        role='what',
        labels=_common_models.Labels({"my": "label"})
    )
    assert lp.labels.values == {"my": "label"}

def test_schedule_pointing_to_datetime():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Datetime),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        }
    )
    lp = workflow_to_test.create_launch_plan(
        schedule=_schedules.CronSchedule("* * ? * * *", kickoff_time_input_arg='required_input'),
        role='what'
    )
    assert lp.entity_metadata.schedule.kickoff_time_input_arg == 'required_input'
    assert lp.entity_metadata.schedule.cron_expression == "* * ? * * *"

def test_hard_coded_deprecated_role():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Integer),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        }
    )
    lp = workflow_to_test.create_launch_plan(role='override')
    assert lp.auth_role.assumable_iam_role == 'override'

def test_schedule():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Integer),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        }
    )
    lp = workflow_to_test.create_launch_plan(
        fixed_inputs={'required_input': 5},
        schedule=_schedules.CronSchedule("* * ? * * *"),
        role='what'
    )
    assert lp.entity_metadata.schedule.kickoff_time_input_arg is None
    assert lp.entity_metadata.schedule.cron_expression == "* * ? * * *"
    assert lp.is_scheduled

def nested_dynamic_wf_task(wf_params, task_input_num, out):
    wf_params.logging.info("Running inner task... yielding a code generated sub workflow")

    # Inner workflow
    input_a = Input(Types.Integer, help="Tell me something")
    node1 = sq_sub_task(in1=input_a)

    MyUnregisteredWorkflowInner = workflow(
        inputs={
            'a': input_a,
        },
        outputs={
            'ooo': Output(node1.outputs.out1, sdk_type=Types.Integer,
                          help='This is an integer output')
        },
        nodes={
            'node_one': node1,
        }
    )

    setattr(MyUnregisteredWorkflowInner, 'auto_assign_name', manual_assign_name)

def test_hard_coded_assumable_iam_role():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Integer),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        }
    )
    lp = workflow_to_test.create_launch_plan(assumable_iam_role='override')
    assert lp.auth_role.assumable_iam_role == 'override'

def test_schedule_pointing_to_datetime():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Datetime),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        }
    )
    lp = workflow_to_test.create_launch_plan(
        schedule=_schedules.CronSchedule("* * ? * * *", kickoff_time_input_arg='required_input'),
        role='what'
    )
    assert lp.entity_metadata.schedule.kickoff_time_input_arg == 'required_input'
    assert lp.entity_metadata.schedule.cron_expression == "* * ? * * *"

def __init__(self, sdk_type, help=None, **kwargs):
        """
        :param flytekit.common.types.base_sdk_types.FlyteSdkType sdk_type: This is the SDK type necessary to create an
            input to this workflow.
        :param Text help: An optional help string to describe the input to users.
        :param bool required: If set, default must be None
        :param T default: If this is not a required input, the value will default to this value.  Specify as a kwarg.
        """
        super(Input, self).__init__('', _type_helpers.python_std_to_sdk_type(sdk_type), help=help, **kwargs)