Merge pull request #1785 from PrefectHQ/update-fix

Fix issue with flow.update not transferring constants

CHANGELOG.md

@@ -19,6 +19,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 ### Fixes
 
 - Fix issue with `flow.visualize()` for mapped tasks which are skipped - [#1765](https://github.com/PrefectHQ/prefect/issues/1765)
+- Fix issue with `flow.update()` not transferring constants - [#1785](https://github.com/PrefectHQ/prefect/pull/1785)
 
 ### Deprecations
 

src/prefect/core/flow.py

@@ -557,6 +557,8 @@ def update(self, flow: "Flow", validate: bool = None) -> None:
                     validate=validate,
                 )
 
+        self.constants.update(flow.constants or {})
+
     @cache
     def all_upstream_edges(self) -> Dict[Task, Set[Edge]]:
         """

tests/core/test_flow.py

@@ -725,6 +725,19 @@ def test_update():
     assert len(f2.edges) == 2
 
 
+def test_update_with_constants():
+    with Flow("math") as f:
+        x = Parameter("x")
+        d = x["d"] + 4
+
+    new_flow = Flow("test")
+    new_flow.update(f)
+
+    flow_state = new_flow.run(x=dict(d=42))
+    assert flow_state.is_successful()
+    assert flow_state.result[d].result == 46
+
+
 def test_update_with_mapped_edges():
     t1 = Task()
     t2 = Task()

tests/engine/cloud/test_cloud_task_runner.py

@@ -1,6 +1,7 @@
 import datetime
 import json
 import os
+import sys
 import tempfile
 import time
 import uuid
@@ -560,6 +561,10 @@ def multiprocessing_helper(executor):
         assert isinstance(res, TimedOut)
         assert len(results.split()) >= 30
 
+    @pytest.mark.skipif(
+        sys.platform == "win32",
+        reason="Randomly fails on Windows and Windows is not currently fully supported for Cloud Deployments",
+    )
     @pytest.mark.parametrize(
         "executor", ["local", "sync", "mproc", "mthread"], indirect=True
     )