Update name of heartbeat config for Cloud

CHANGELOG.md

@@ -10,7 +10,8 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 
 ### Enhancements
 
-- - Add examples to Interactive API Docs [#2122](https://github.com/PrefectHQ/prefect/pull/2122)
+- Update Cloud config name for heartbeat settings - [#2081](https://github.com/PrefectHQ/prefect/pull/2081)
+- Add examples to Interactive API Docs [#2122](https://github.com/PrefectHQ/prefect/pull/2122)
 
 ### Task Library
 

src/prefect/engine/cloud/flow_runner.py

@@ -75,7 +75,7 @@ def _heartbeat(self) -> bool:
                 }
             }
             flow_run = self.client.graphql(query).data.flow_run_by_pk
-            if flow_run.flow.settings.get("disable_heartbeat"):
+            if not flow_run.flow.settings.get("heartbeat_enabled", True):
                 return False
             return True
         except Exception as exc:

src/prefect/engine/cloud/task_runner.py

@@ -72,7 +72,7 @@ def _heartbeat(self) -> bool:
                 }
             }
             flow_run = self.client.graphql(query).data.flow_run_by_pk
-            if flow_run.flow.settings.get("disable_heartbeat"):
+            if not flow_run.flow.settings.get("heartbeat_enabled", True):
                 return False
             return True
         except Exception as exc:

tests/engine/cloud/test_cloud_flow_runner.py

@@ -387,14 +387,17 @@ def test_heartbeat_traps_errors_caused_by_client(caplog, monkeypatch):
     assert "Heartbeat failed for Flow 'bad'" in log.message
 
 
-def test_flow_runner_heartbeat_sets_command(monkeypatch):
+@pytest.mark.parametrize("setting_available", [True, False])
+def test_flow_runner_heartbeat_sets_command(monkeypatch, setting_available):
     client = MagicMock()
     monkeypatch.setattr(
         "prefect.engine.cloud.flow_runner.Client", MagicMock(return_value=client)
     )
-    client.graphql.return_value.data.flow_run_by_pk.flow.settings = dict(
-        disable_heartbeat=False
+
+    client.graphql.return_value.data.flow_run_by_pk.flow.settings = (
+        dict(heartbeat_enabled=True) if setting_available else {}
     )
+
     runner = CloudFlowRunner(flow=prefect.Flow(name="test"))
     with prefect.context(flow_run_id="foo"):
         res = runner._heartbeat()
@@ -409,7 +412,7 @@ def test_flow_runner_does_not_have_heartbeat_if_disabled(monkeypatch):
         "prefect.engine.cloud.flow_runner.Client", MagicMock(return_value=client)
     )
     client.graphql.return_value.data.flow_run_by_pk.flow.settings = dict(
-        disable_heartbeat=True
+        heartbeat_enabled=False
     )
 
     # set up the CloudFlowRunner

tests/engine/cloud/test_cloud_task_runner.py

@@ -483,14 +483,16 @@ def test_heartbeat_traps_errors_caused_by_bad_attributes(self, caplog, monkeypat
         assert log.levelname == "ERROR"
         assert "Heartbeat failed for Task 'Task'" in log.message
 
-    def test_task_runner_heartbeat_sets_command(self, monkeypatch):
+    @pytest.mark.parametrize("setting_available", [True, False])
+    def test_task_runner_heartbeat_sets_command(self, monkeypatch, setting_available):
         client = MagicMock()
         monkeypatch.setattr(
             "prefect.engine.cloud.task_runner.Client", MagicMock(return_value=client)
         )
-        client.graphql.return_value.data.flow_run_by_pk.flow.settings = dict(
-            disable_heartbeat=False
+        client.graphql.return_value.data.flow_run_by_pk.flow.settings = (
+            dict(heartbeat_enabled=True) if setting_available else {}
         )
+
         runner = CloudTaskRunner(task=Task())
         runner.task_run_id = "foo"
         res = runner._heartbeat()
@@ -504,7 +506,7 @@ def test_task_runner_does_not_have_heartbeat_if_disabled(self, monkeypatch):
             "prefect.engine.cloud.task_runner.Client", MagicMock(return_value=client)
         )
         client.graphql.return_value.data.flow_run_by_pk.flow.settings = dict(
-            disable_heartbeat=True
+            heartbeat_enabled=False
         )
         runner = CloudTaskRunner(task=Task())
         runner.task_run_id = "foo"