"""Test task for a given dag_id."""
    
    
    
    settings.MASK_SECRETS_IN_LOGS = True
    handlers = logging.getLogger("airflow.task").handlers
    already_has_stream_handler = False
    for handler in handlers:
        already_has_stream_handler = isinstance(handler, logging.StreamHandler)
        if already_has_stream_handler:
            break
    if not already_has_stream_handler:
        logging.getLogger("airflow.task").propagate = True
    env_vars = {"AIRFLOW_TEST_MODE": "True"}
    if args.env_vars:
        env_vars.update(args.env_vars)
        os.environ.update(env_vars)
    dag = dag or get_dag(args.subdir, args.dag_id)
    task = dag.get_task(task_id=args.task_id)
    
    if args.task_params:
        passed_in_params = json.loads(args.task_params)
        task.params.update(passed_in_params)
    if task.params and isinstance(task.params, ParamsDict):
        task.params.validate()