How to use the flytekit.common.exceptions.user.FlyteAssertion function in flytekit

def test_launch_plan_node():
    workflow_to_test = _workflow.workflow(
        {},
        inputs={
            'required_input': _workflow.Input(_types.Types.Integer),
            'default_input': _workflow.Input(_types.Types.Integer, default=5)
        },
        outputs={
            'out': _workflow.Output([1, 2, 3], sdk_type=[_types.Types.Integer])
        }
    )
    lp = workflow_to_test.create_launch_plan()

    # Test that required input isn't set
    with _pytest.raises(_user_exceptions.FlyteAssertion):
        lp()

    # Test that positional args are rejected
    with _pytest.raises(_user_exceptions.FlyteAssertion):
        lp(1, 2)

    # Test that type checking works
    with _pytest.raises(_user_exceptions.FlyteTypeException):
        lp(required_input='abc', default_input=1)

    # Test that bad arg name is detected
    with _pytest.raises(_user_exceptions.FlyteAssertion):
        lp(required_input=1, bad_arg=1)

    # Test default input is accounted for
    n = lp(required_input=10)

def __call__(self, *args, **input_map):
        """
        :param list[T] args: Do not specify.  Kwargs only are supported for this function.
        :param dict[str, T] input_map: Map of inputs.  Can be statically defined or OutputReference links.
        :rtype: flytekit.common.nodes.SdkNode
        """
        if len(args) > 0:
            raise _user_exceptions.FlyteAssertion(
                "When adding a task as a node in a workflow, all inputs must be specified with kwargs only.  We "
                "detected {} positional args.".format(len(args))
            )

        bindings, upstream_nodes = self.interface.create_bindings_for_inputs(input_map)

        # TODO: Remove DEADBEEF
        return _nodes.SdkNode(
            id=None,
            metadata=_workflow_model.NodeMetadata("DEADBEEF", self.metadata.timeout, self.metadata.retries, self.metadata.interruptible),
            bindings=sorted(bindings, key=lambda b: b.var),
            upstream_nodes=upstream_nodes,
            sdk_task=self
        )

def outputs(self):
        """
        Returns the outputs to the execution in the standard Python format as dictated by the type engine.  If the
        execution ended in error or the execution is in progress, an exception will be raised.
        :rtype:  dict[Text, T] or None
        """
        if not self.is_complete:
            raise _user_exceptions.FlyteAssertion("Please what until the node execution has completed before "
                                                  "requesting the outputs.")
        if self.error:
            raise _user_exceptions.FlyteAssertion("Outputs could not be found because the execution ended in failure.")

        if self._outputs is None:
            self._outputs = _type_helpers.unpack_literal_map_to_sdk_python_std(
                _engine_loader.get_engine().get_workflow_execution(self).get_outputs()
            )
        return self._outputs

def promote_from_model(cls, model):
        """
        :param flytekit.models.launch_plan.LaunchPlanSpec model:
        :rtype: SdkRunnableLaunchPlan
        """
        raise _user_exceptions.FlyteAssertion(
            "An SdkRunnableLaunchPlan must be created from a reference to local Python code only."
        )

def __enter__(self):
        if not self.local_path:
            if _data_proxy.LocalWorkingDirectoryContext.get() is None:
                raise _user_exceptions.FlyteAssertion(
                    "No temporary file system is present.  Either call this method from within the "
                    "context of a task or surround with a 'with LocalTestFileSystem():' block.  Or "
                    "specify a path when calling this function."
                )
            self._directory = _utils.AutoDeletingTempDir(
                _uuid.uuid4().hex,
                tmp_dir=_data_proxy.LocalWorkingDirectoryContext.get().name
            )
            self._is_managed = True
            self._directory.__enter__()

            if 'r' in self.mode:
                _data_proxy.Data.get_data(self.remote_location, self.local_path, is_multipart=True)

partition_name=partition_name,
                    partition_value=partition_value))
            partition_formatter = "PARTITION (\n\t{conditions}\n)"
            partition_string = partition_formatter.format(conditions=",\n\t".join(partition_conditions))

        if partitions_in_table and partitions:
            where_clauses = []
            for partition_name, partition_value in partitions:
                where_clauses.append("\n\t\t{schema_name} = {value_str} AND ".format(
                    schema_name=table_to_schema_name_map[partition_name],
                    value_str=partition_value
                ))
            where_string = "WHERE\n\t\t{where_clauses}".format(where_clauses=" AND\n\t\t".join(where_clauses))

        if where_string or partitions_in_table:
            raise _user_exceptions.FlyteAssertion(
                "Currently, the partition values should not be present in the schema pushed to Hive.")
        if append_to_partition:
            raise _user_exceptions.FlyteAssertion(
                "Currently, partitions can only be overwritten, they cannot be appended."
            )
        if not partitions:
            raise _user_exceptions.FlyteAssertion(
                "Currently, partition values MUST be specified for writing to a table."
            )

        return _format_insert_partition_query(
            remote_location=self.remote_location,
            table_name=table_name,
            partition_string=partition_string)

def fetch(cls, project, domain, name, version=None):
        """
        This function uses the engine loader to call create a hydrated task from Admin.
        :param Text project:
        :param Text domain:
        :param Text name:
        :param Text version:
        :rtype: SdkRunnableLaunchPlan
        """
        raise _user_exceptions.FlyteAssertion(
            "An SdkRunnableLaunchPlan must be created from a reference to local Python code only."
        )

def iter_chunks(self, *args, **kwargs):
        raise _user_exceptions.FlyteAssertion("{} is write only.".format(self._schema))

def upload(self):
        """
        Upload the schema to the remote location
        """
        if 'w' not in self.mode:
            raise _user_exceptions.FlyteAssertion("Cannot upload a read-only schema!")

        elif not self.local_path:
            raise _user_exceptions.FlyteAssertion("The schema is not currently backed by a local directory "
                                                  "and therefore cannot be uploaded.  Please write to this before "
                                                  "attempting an upload.")
        else:
            # TODO: Introduce system logging
            # logging.info("Putting {} -> {}".format(self.local_path, self.remote_location))
            _data_proxy.Data.put_data(
                self.local_path,
                self.remote_location,
                is_multipart=True
            )

def update(self, state):
        """
        :param int state: Enum value from flytekit.models.launch_plan.LaunchPlanState
        """
        if not self.id:
            raise _user_exceptions.FlyteAssertion(
                "Failed to update launch plan because the launch plan's ID is not set. Please call register to fetch "
                "or register the identifier first"
            )
        return _engine_loader.get_engine().get_launch_plan(self).update(self.id, state)