Cleanup deployment-start-time feature branch, add tests

src/prefect/cli/deployment.py

@@ -384,7 +384,8 @@ async def run(
     """
     Create a flow run for the given flow and deployment.
 
-    The flow run will be scheduled for now and an agent must execute it.
+    The flow run will be scheduled for now if start-in or start-at are unspecified;
+    and an agent must execute it.
 
     The flow run will not execute until an agent starts.
     """
@@ -411,15 +412,19 @@ async def run(
         )
     parameters = {**multi_params, **cli_params}
 
-    # Parse the start time if provided
-    if start_in:
-        start_time_raw = "in " + start_in
-    elif start_at:
-        start_time_raw = "at " + start_at
-    else:
-        start_time_raw = None
+    if start_in and start_at:
+        exit_with_error(
+            "Expected optional start_in field or start_at field but not both."
+        )
 
-    if start_time_raw:
+    elif start_in is None and start_at is None:
+        scheduled_start_time = now
+        human_readable_dt_diff = " (now)"
+    else:
+        if start_in:
+            start_time_raw = "in " + start_in
+        else:
+            start_time_raw = "at " + start_at
         with warnings.catch_warnings():
             # PyTZ throws a warning based on dateparser usage of the library
             # See https://github.com/scrapinghub/dateparser/issues/1089
@@ -433,9 +438,11 @@ async def run(
             )
         if start_time_parsed is None:
             exit_with_error(f"Unable to parse scheduled start time {start_time_raw!r}.")
+
         scheduled_start_time = pendulum.instance(start_time_parsed)
-    else:
-        scheduled_start_time = now
+        human_readable_dt_diff = (
+            " (" + pendulum.format_diff(scheduled_start_time.diff(now)) + ")"
+        )
 
     async with get_client() as client:
         deployment = await get_deployment(client, name, deployment_id)
@@ -474,21 +481,10 @@ async def run(
     else:
         run_url = "<no dashboard available>"
 
-    if start_in:
-        scheduled_display = (
-            scheduled_start_time.in_tz(
-                pendulum.tz.local_timezone()
-            ).to_datetime_string()
-            + " ("
-            + pendulum.format_diff(scheduled_start_time.diff(now))
-            + ")"
-        )
-    elif start_at:
-        scheduled_display = scheduled_start_time.in_tz(
-            pendulum.tz.local_timezone()
-        ).to_datetime_string()
-    else:
-        scheduled_display = "now"
+    scheduled_display = (
+        scheduled_start_time.in_tz(pendulum.tz.local_timezone()).to_datetime_string()
+        + human_readable_dt_diff
+    )
 
     app.console.print(f"Created flow run {flow_run.name!r}.")
     app.console.print(

tests/cli/deployment/test_deployment_run.py

@@ -0,0 +1,212 @@
+import dateparser
+import pendulum
+import pytest
+
+import prefect
+from prefect.testing.cli import invoke_and_assert
+from prefect.utilities.asyncutils import run_sync_in_worker_thread
+
+
+@pytest.fixture
+async def deployment_name(deployment, orion_client):
+    flow = await orion_client.read_flow(deployment.flow_id)
+    return f"{flow.name}/{deployment.name}"
+
+
+@pytest.fixture
+def frozen_now(monkeypatch):
+    now = pendulum.now()
+    monkeypatch.setattr("pendulum.now", lambda *_: now)
+    yield now
+
+
+def test_both_start_in_and_start_at_raises():
+    invoke_and_assert(
+        command=["deployment", "run", "--start-in", "foo", "--start-at", "bar"],
+        expected_code=1,
+        expected_output="Expected optional start_in field or start_at field but not both.",
+    )
+
+
+@pytest.mark.parametrize(
+    "start_at,expected_start_time",
+    [
+        ("12-1-22 5pm", "2022-12-01 17:00:00"),
+        ("1/1/30", "2030-01-01 00:00:00"),
+        ("13/1/20 13:31", "2020-01-13 13:31:00"),
+        ("9pm December 31st 2022", "2022-12-31 21:00:00"),
+        ("January 2nd 2023", "2023-01-02 00:00:00"),
+    ],
+)
+async def test_start_at_option_schedules_flow_run_in_future(
+    deployment_name: str,
+    orion_client: prefect.OrionClient,
+    start_at: str,
+    expected_start_time,
+):
+
+    await run_sync_in_worker_thread(
+        invoke_and_assert,
+        command=[
+            "deployment",
+            "run",
+            deployment_name,
+            "--start-at",
+            start_at,
+        ],
+        expected_output_contains=["Scheduled start time:", expected_start_time],
+    )
+
+    flow_runs = await orion_client.read_flow_runs()
+    assert len(flow_runs) == 1
+    flow_run = flow_runs[0]
+
+    assert flow_run.state.is_scheduled()
+
+
+@pytest.mark.parametrize(
+    "start_in,expected_duration",
+    [
+        ("10 minutes", pendulum.duration(minutes=10)),
+        ("5 days", pendulum.duration(days=5)),
+        ("3 seconds", pendulum.duration(seconds=3)),
+        (None, pendulum.duration(seconds=0)),
+        ("1 year and 3 months", pendulum.duration(years=1, months=3)),
+        ("2 weeks & 1 day", pendulum.duration(weeks=2, days=1)),
+        ("27 hours + 4 mins", pendulum.duration(days=1, hours=3, minutes=4)),
+    ],
+)
+async def test_start_in_option_schedules_flow_run_in_future(
+    deployment_name: str,
+    frozen_now,
+    orion_client: prefect.OrionClient,
+    start_in: str,
+    expected_duration,
+):
+    expected_start_time = frozen_now + expected_duration
+    expected_display = expected_start_time.in_tz(
+        pendulum.tz.local_timezone()
+    ).to_datetime_string()
+
+    await run_sync_in_worker_thread(
+        invoke_and_assert,
+        command=[
+            "deployment",
+            "run",
+            deployment_name,
+            "--start-in",
+            start_in,
+        ],
+        expected_output_contains=f"Scheduled start time: {expected_display}",
+    )
+
+    flow_runs = await orion_client.read_flow_runs()
+    assert len(flow_runs) == 1
+    flow_run = flow_runs[0]
+
+    assert flow_run.state.is_scheduled()
+    scheduled_time = flow_run.state.state_details.scheduled_time
+
+    assert scheduled_time == expected_start_time
+
+
+@pytest.mark.parametrize(
+    "start_in, expected_display",
+    [
+        ("10 minutes", "in 10 minutes"),
+        ("5 days", "in 5 days"),
+        ("3 seconds", "in a few seconds"),
+        (None, "now"),
+        ("1 year and 3 months", "in 1 year"),
+        ("2 weeks & 1 day", "in 2 weeks"),
+        ("27 hours + 4 mins", "in 1 day"),
+    ],
+)
+async def test_start_in_displays_scheduled_start_time(
+    deployment_name: str,
+    orion_client: prefect.OrionClient,
+    start_in: str,
+    expected_display: str,
+):
+
+    await run_sync_in_worker_thread(
+        invoke_and_assert,
+        command=[
+            "deployment",
+            "run",
+            deployment_name,
+            "--start-in",
+            start_in,
+        ],
+        expected_output_contains=["Scheduled start time:", expected_display],
+    )
+
+    flow_runs = await orion_client.read_flow_runs()
+    assert len(flow_runs) == 1
+    flow_run = flow_runs[0]
+
+    assert flow_run.state.is_scheduled()
+
+
+def test_start_at_displays_scheduled_start_time(deployment_name: str, monkeypatch):
+    monkeypatch.setattr(
+        "pendulum.now",
+        lambda *_: pendulum.instance(dateparser.parse("January 1st 2023")),
+    )
+    invoke_and_assert(
+        command=[
+            "deployment",
+            "run",
+            deployment_name,
+            "--start-at",
+            "January 1st 2022",
+        ],
+        expected_output_contains="Scheduled start time: 2022-01-01 00:00:00 (1 year ago)",
+    )
+
+
+def test_start_at_invalid_input(deployment_name: str):
+    invoke_and_assert(
+        command=[
+            "deployment",
+            "run",
+            deployment_name,
+            "--start-at",
+            "foobar",
+        ],
+        expected_code=1,
+        expected_output="Unable to parse scheduled start time 'at foobar'.",
+    )
+
+
+def test_start_in_invalid_input(deployment_name: str):
+    invoke_and_assert(
+        command=[
+            "deployment",
+            "run",
+            deployment_name,
+            "--start-in",
+            "foobar",
+        ],
+        expected_code=1,
+        expected_output="Unable to parse scheduled start time 'in foobar'.",
+    )
+
+
+async def test_run_deployment_only_creates_one_flow_run(
+    deployment_name: str, orion_client: prefect.OrionClient, deployment
+):
+    await run_sync_in_worker_thread(
+        invoke_and_assert,
+        command=[
+            "deployment",
+            "run",
+            deployment_name,
+        ],
+    )
+
+    flow_runs = await orion_client.read_flow_runs()
+    assert len(flow_runs) == 1
+    flow_run = flow_runs[0]
+
+    assert flow_run.deployment_id == deployment.id