Merge pull request #301 from PrefectHQ/schedule-active

Refuse to set schedules active for flows without any schedule

src/prefect_server/api/flows.py

@@ -549,6 +549,11 @@ async def set_schedule_active(flow_id: str) -> bool:
  if not all([required_names <= c for c in clock_params]):
  raise ValueError("Can not schedule a flow that has required parameters.")
 
+ # if there is no referenceable schedule for this Flow, do not change the schedule
+ # to active to avoid confusion
+ if flow.schedule is None and getattr(flow.flow_group, "schedule", None) is None:
+ return False
+
  result = await models.Flow.where(id=flow_id).update(
  set={"is_schedule_active": True}
  )

tests/api/test_flows.py

@@ -1114,6 +1114,35 @@ async def test_set_schedule_active_doesnt_duplicate_runs(self, flow_id):
 
  assert await models.FlowRun.where({"flow_id": {"_eq": flow_id}}).count() == 10
 
+ async def test_set_schedule_active_with_no_schedule_on_flow_or_flow_group(
+ self, flow_id, flow_group_id
+ ):
+ await models.Flow.where(id=flow_id).update(
+ set={"is_schedule_active": False, "schedule": None}
+ )
+ await models.FlowGroup.where(id=flow_group_id).update(set={"schedule": None})
+
+ success = await api.flows.set_schedule_active(flow_id=flow_id)
+ assert success is False
+ flow = await models.Flow.where(id=flow_id).first({"is_schedule_active"})
+ assert flow.is_schedule_active is False
+
+ async def test_set_schedule_active_with_no_schedule_on_flow(
+ self, flow_id, flow_group_id
+ ):
+ await models.Flow.where(id=flow_id).update(
+ set={"is_schedule_active": False, "schedule": None}
+ )
+ await api.flow_groups.set_flow_group_schedule(
+ flow_group_id=flow_group_id,
+ clocks=[{"type": "IntervalClock", "interval": 420000000}],
+ )
+
+ success = await api.flows.set_schedule_active(flow_id=flow_id)
+ assert success is True
+ flow = await models.Flow.where(id=flow_id).first({"is_schedule_active"})
+ assert flow.is_schedule_active is True
+
 
 class TestSetScheduleInactive:
  async def test_set_schedule_inactive(self, flow_id):