From 845901bf3e6c5bd574f1faa0b59bbae1e31c00ed Mon Sep 17 00:00:00 2001
From: Michael Adkins <michael@prefect.io>
Date: Thu, 7 Oct 2021 16:11:35 -0500
Subject: [PATCH] Refuse to set schedules active for flows without any schedule

---
 src/prefect_server/api/flows.py |  5 +++++
 tests/api/test_flows.py         | 29 +++++++++++++++++++++++++++++
 2 files changed, 34 insertions(+)

diff --git a/src/prefect_server/api/flows.py b/src/prefect_server/api/flows.py
index 01baaad4..bbcec754 100644
--- a/src/prefect_server/api/flows.py
+++ b/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}
     )
diff --git a/tests/api/test_flows.py b/tests/api/test_flows.py
index 707e6f98..69d39912 100644
--- a/tests/api/test_flows.py
+++ b/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):