https://flyte.org logo
#ask-the-community
Title
# ask-the-community
l

L godlike

02/08/2024, 7:54 AM
(solved) Hi, community, I am wondering does anyone know that how to use
CLI Runner
to test
pyflyte run
in local environemnt? Here is my example in the thread!
Copy code
from airflow.sensors.filesystem import FileSensor
from airflow.sensors.time_sensor import TimeSensor
from datetime import datetime, timedelta
from pytz import UTC

from flytekit import task, workflow

@task()
def t1():
    print("flyte")

@workflow
def wf_time_sensor():
    # sensor = FileSensor(task_id="id", filepath="/tmp/1234")
    sensor = TimeSensor(task_id="fire_immediately", target_time=(datetime.now(tz=UTC)+timedelta(seconds=1)).time())
    sensor >> t1()

@workflow
def wf_file_sensor():
    sensor = FileSensor(task_id="id", filepath="/tmp/1234")
    sensor >> t1()


if __name__ == '__main__':
    from click.testing import CliRunner
    from flytekit.clis.sdk_in_container import pyflyte

    runner = CliRunner()
    # result = runner.invoke(cli, "--help")
    result = runner.invoke(pyflyte.main, ["run", "airflow_task.py", "wf_file_sensor"])
update: I have to print
result.output
at the end of line