fsapp: set control_folder broker option

iterative · May 25, 2022 · 833d8cb · 833d8cb
1 parent 3a1e4d5
commit 833d8cb
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 1 deletion.
diff --git a/src/dvc_task/app/filesystem.py b/src/dvc_task/app/filesystem.py
@@ -20,17 +20,24 @@ def _get_fs_config(
     result_serializer: str = "json",
 ) -> Dict[str, Any]:
     broker_path = os.path.join(wdir, "broker")
+    broker_control_path = unc_path(os.path.join(broker_path, "control"))
     broker_in_path = unc_path(os.path.join(broker_path, "in"))
     broker_processed_path = unc_path(os.path.join(broker_path, "processed"))
     result_path = os.path.join(wdir, "result")
 
     if mkdir:
-        for path in (broker_in_path, broker_processed_path, result_path):
+        for path in (
+            broker_control_path,
+            broker_in_path,
+            broker_processed_path,
+            result_path,
+        ):
             makedirs(path, exist_ok=True)
 
     return {
         "broker_url": "filesystem://",
         "broker_transport_options": {
+            "control_folder": broker_control_path,
             "data_folder_in": broker_in_path,
             "data_folder_out": broker_in_path,
             "processed_folder": broker_processed_path,

diff --git a/tests/app/test_filesystem.py b/tests/app/test_filesystem.py
@@ -30,6 +30,7 @@
 def test_config(tmp_dir: TmpDir):
     """Should return a filesystem broker/resut config."""
     config = _get_fs_config(str(tmp_dir), mkdir=True)
+    assert (tmp_dir / "broker" / "control").is_dir()
     assert (tmp_dir / "broker" / "in").is_dir()
     assert (tmp_dir / "broker" / "processed").is_dir()
     assert (tmp_dir / "result").is_dir()