Repository: AgnostiqHQ/covalent
Pull request #1641 (2023-05-12 02:44:07)

`workdir` for local and dask executors along with `create_unique_workdir` option
Closes https://github.com/AgnostiqHQ/covalent/issues/1619 and https://github.com/AgnostiqHQ/covalent/issues/1628

- [x] I have added the tests to cover my changes.
- [x] I have updated the documentation and CHANGELOG accordingly.
- [x] I have read the CONTRIBUTING document.

Stuff that remains:
- ~~Tests (both fixing + adding new ones...)~~
- New Docs

But here is an example that showcases how this works:

If no `workdir` is specified, the default value is `os.path.join(os.environ["HOME"], "covalent", "workdir")`. I think that's a reasonable default...

```
import covalent as ct

executor = ct.executor.LocalExecutor(workdir="/Users/USER/SOME-PATH/try-ct-work")

@ct.electron(executor=executor)
def job(val1, val2):
    with open("job.txt", "w") as w:
        w.write(str(val1 + val2))
    return "Done!"

@ct.lattice(executor=executor)
def workflow(val1, val2, val3, val4):
    job1 = job(val1, val2)
    job2 = job(val3, val4)
    return "Done!"

dispatch_id = ct.dispatch(workflow)(1, 2, 3, 4)
result = ct.get_result(dispatch_id)
print(result)
```

Similarly for `dask` executor as well.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1641?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) Report
> Merging [#1641](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1641?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (27e26b7) into [develop](https://app.codecov.io/gh/AgnostiqHQ/covalent/commit/01738d33b2eead05d59cea5fa5d0365c4ac1bead?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (01738d3) will **increase** coverage by `1.34%`.
> The diff coverage is `80.00%`.

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1641/graphs/tree.svg?width=650&height=150&src=pr&token=YGHCB3DE4P&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1641?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)

```diff
@@             Coverage Diff             @@
##           develop    #1641      +/-   ##
===========================================
+ Coverage    86.21%   87.56%   +1.34%     
===========================================
  Files          161       85      -76     
  Lines         6762     5097    -1665     
  Branches       145        0     -145     
===========================================
- Hits          5830     4463    -1367     
+ Misses         830      634     -196     
+ Partials       102        0     -102     
```

| Flag | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| Dispatcher | `87.55% <ø> (ø)` | | Carriedforward from [01738d3](https://app.codecov.io/gh/AgnostiqHQ/covalent/commit/01738d33b2eead05d59cea5fa5d0365c4ac1bead?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) |
| Functional_Tests | `?` | | |
| SDK | `87.56% <80.00%> (-0.13%)` | :arrow_down: | |
| UI_Frontend | `?` | | |

*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) to find out more.

</details>
madhur-tandon: CC: @kessler-frost / @AlejandroEsquivel / @santoshkumarradha for some initial comments about the approach etc...
madhur-tandon: The two new tests pass when `covalent server` is up and running...
But the CI doesn't start it, thus the failures for now.
madhur-tandon: This should be ready, CC: @arosen93 @wjcunningham7 
kessler-frost: @arosen93 @madhur-tandon What do you guys think about moving this attribute to `BaseExecutor` instead? Thus making the change in `covalent/executor/base.py::wrapper_fn` instead of `io_wrapper`. I think this'll be a great feature for all the executors and not just dask and local. @wjcunningham7 any thoughts?
kessler-frost: Nice! As for #1628 I think, we should be able to do the same as well (i.e. add to the `BaseExecutor`).
kessler-frost: Oh yeah, my bad, I forgot it's `_AbstractBaseExecutor` now. 
madhur-tandon: @kessler-frost I am yet to move stuff to `wrapper_fn` but can you re-trigger CI again? It seems to be failing in uploading `codecov` reports...
kessler-frost: @madhur-tandon Feel free to ping me whenever this is ready for review. I'm excited to include this feature in Covalent 😄 .
madhur-tandon: @kessler-frost This should be ready!

CC: @arosen93
The subsequent PR will have to make quite a few changes now 😅 but I will be glad to assist in whichever way I can.
arosen93: Not a problem! I was overly ambitious 😊😉
kessler-frost: Hey @madhur-tandon @arosen93 , so since this is quite an important feature, we're still internally discussing about its potential implications on other remote executors and their compatibility with latest covalent. I'll keep you guys posted once we determine what should be the best course of action.
madhur-tandon: @kessler-frost I will do your suggested changes later today...kinda late because of timezone difference 😅 
But, let me know about the course of action and I will adjust accordingly.
madhur-tandon: @kessler-frost requesting a re-review...
madhur-tandon: @santoshkumarradha can you re-trigger the `version` check?
kessler-frost: Hey @madhur-tandon @arosen93 ! Sorry, to get back to you guys this late 😅 . After discussion with the team we've decided that for now it's better to keep this feature limited to Dask and Local executors (and any other local-ish ones that might be added later) in order to prevent breaking backwards compatibility of other executors.

Again, I apologize since now you might have to switch back to modifying the io_wrapper function @madhur-tandon 😶 but one good thing is that now we have established that executors should be evolving independently according to their requirements and needs instead of us trying to make general changes for feature additions that might not even be relevant to some of them.

Thanks a lot for working on this 🙏! We appreciate it a lot and these kinds of discussions really help in shaping things.
madhur-tandon: @kessler-frost Reverted changes here: 0ebb38f099f8591c8b752da88a683f21d392d5b6
madhur-tandon: @kessler-frost Please don't merge it yet
madhur-tandon: @kessler-frost This should be ready now with changes reverted...
madhur-tandon: @arosen93 I have accommodated changes for `create_unique_workdir` within this PR itself.

CC: @kessler-frost @santoshkumarradha 
arosen93: Just chiming in to say that I've tested this branch on my workflows and everything seems to work as expected! :D Thanks! Looking forward to getting this merged!!
madhur-tandon: @kessler-frost any ETA on the review for this?
kessler-frost: I should be able to finish the review by today eod at max.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 569428e8..e16ed1b4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `executor` property to `Electron` class, allowing updation of executor after electron function definition
 - Added ability to hide post-processing electrons on the UI.
 - Added prettify of names for the graph screen on the UI.
+- Ability to specify a `workdir` for `local` and `dask` executors along with `create_unique_workdir` option for each electron / node.
 
 ### Operations
 
diff --git a/covalent/_shared_files/defaults.py b/covalent/_shared_files/defaults.py
index 20d3b262..42c54035 100644
--- a/covalent/_shared_files/defaults.py
+++ b/covalent/_shared_files/defaults.py
@@ -52,7 +52,10 @@ def get_default_sdk_config():
         + "/covalent.conf",
         "log_dir": (
             os.environ.get("COVALENT_LOGDIR")
-            or (os.environ.get("XDG_CACHE_HOME") or (os.environ["HOME"] + "/.cache/covalent"))
+            or os.path.join(
+                os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+                "covalent",
+            )
         ),
         "log_level": os.environ.get("LOGLEVEL", "WARNING").lower(),
         "enable_logging": os.environ.get("COVALENT_LOG_TO_FILE", "true").lower(),
@@ -71,11 +74,17 @@ def get_default_dispatcher_config():
         "address": os.environ.get("COVALENT_DISPATCHER_ADDR", "localhost"),
         "port": int(os.environ.get("COVALENT_SVC_PORT", 48008)),
         "cache_dir": os.environ.get("COVALENT_CACHE_DIR")
-        or ((os.environ.get("XDG_CACHE_HOME") or (os.environ["HOME"] + "/.cache")) + "/covalent"),
+        or os.path.join(
+            os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+            "covalent",
+        ),
         "results_dir": os.environ.get("COVALENT_DATA_DIR")
         or os.path.join(os.environ["HOME"], ".local/share/covalent/data"),
         "log_dir": os.environ.get("COVALENT_LOGDIR")
-        or ((os.environ.get("XDG_CACHE_HOME") or (os.environ["HOME"] + "/.cache/covalent"))),
+        or os.path.join(
+            os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+            "covalent",
+        ),
         "db_path": os.environ.get("COVALENT_DATABASE")
         or (
             (os.environ.get("XDG_DATA_HOME") or (os.environ["HOME"] + "/.local/share"))
@@ -87,9 +96,15 @@ def get_default_dispatcher_config():
 def get_default_dask_config():
     return {
         "cache_dir": os.environ.get("COVALENT_CACHE_DIR")
-        or ((os.environ.get("XDG_CACHE_HOME") or (os.environ["HOME"] + "/.cache/covalent"))),
+        or os.path.join(
+            os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+            "covalent",
+        ),
         "log_dir": os.environ.get("COVALENT_LOGDIR")
-        or ((os.environ.get("XDG_CACHE_HOME") or (os.environ["HOME"] + "/.cache/covalent"))),
+        or os.path.join(
+            os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+            "covalent",
+        ),
         "mem_per_worker": "auto",
         "threads_per_worker": 1,
         "num_workers": dask.system.CPU_COUNT,
@@ -115,7 +130,10 @@ def get_default_ui_config():
         else 48008,
         "dev_port": 49009,
         "log_dir": os.environ.get("COVALENT_LOGDIR")
-        or ((os.environ.get("XDG_CACHE_HOME") or (os.environ["HOME"] + "/.cache/covalent"))),
+        or os.path.join(
+            os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+            "covalent",
+        ),
     }
 
 
diff --git a/covalent/executor/base.py b/covalent/executor/base.py
index 6e7b8ccf..e46f6590 100644
--- a/covalent/executor/base.py
+++ b/covalent/executor/base.py
@@ -162,7 +162,6 @@ class _AbstractBaseExecutor(ABC):
         return active_dispatch_info_manager.claim(dispatch_info)
 
     def short_name(self):
-        module = self.__module__
         return self.__module__.split("/")[-1].split(".")[-1]
 
     def to_dict(self) -> dict:
diff --git a/covalent/executor/executor_plugins/dask.py b/covalent/executor/executor_plugins/dask.py
index c1ae4c1e..8347619f 100644
--- a/covalent/executor/executor_plugins/dask.py
+++ b/covalent/executor/executor_plugins/dask.py
@@ -26,7 +26,7 @@ This is a plugin executor module; it is loaded if found and properly structured.
 """
 
 import os
-from typing import Callable, Dict, List, Literal
+from typing import Callable, Dict, List, Literal, Optional
 
 from dask.distributed import CancelledError, Client, Future
 
@@ -51,6 +51,12 @@ _EXECUTOR_PLUGIN_DEFAULTS = {
     "cache_dir": os.path.join(
         os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"), "covalent"
     ),
+    "workdir": os.path.join(
+        os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+        "covalent",
+        "workdir",
+    ),
+    "create_unique_workdir": False,
 }
 
 
@@ -67,12 +73,27 @@ class DaskExecutor(AsyncBaseExecutor):
         conda_env: str = "",
         cache_dir: str = "",
         current_env_on_conda_fail: bool = False,
+        workdir: str = "",
+        create_unique_workdir: Optional[bool] = None,
     ) -> None:
         if not cache_dir:
-            cache_dir = os.path.join(
-                os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
-                "covalent",
-            )
+            cache_dir = _EXECUTOR_PLUGIN_DEFAULTS["cache_dir"]
+
+        if not workdir:
+            try:
+                workdir = get_config("executors.dask.workdir")
+            except KeyError:
+                workdir = _EXECUTOR_PLUGIN_DEFAULTS["workdir"]
+                debug_msg = f"Couldn't find `executors.dask.workdir` in config, using default value {workdir}."
+                app_log.debug(debug_msg)
+
+        if create_unique_workdir is None:
+            try:
+                create_unique_workdir = get_config("executors.dask.create_unique_workdir")
+            except KeyError:
+                create_unique_workdir = _EXECUTOR_PLUGIN_DEFAULTS["create_unique_workdir"]
+                debug_msg = f"Couldn't find `executors.dask.create_unique_workdir` in config, using default value {create_unique_workdir}."
+                app_log.debug(debug_msg)
 
         if not scheduler_address:
             try:
@@ -82,8 +103,16 @@ class DaskExecutor(AsyncBaseExecutor):
                     "No dask scheduler address found in config. Address must be set manually."
                 )
 
-        super().__init__(log_stdout, log_stderr, conda_env, cache_dir, current_env_on_conda_fail)
+        super().__init__(
+            log_stdout,
+            log_stderr,
+            cache_dir,
+            conda_env,
+            current_env_on_conda_fail,
+        )
 
+        self.workdir = workdir
+        self.create_unique_workdir = create_unique_workdir
         self.scheduler_address = scheduler_address
 
     async def run(self, function: Callable, args: List, kwargs: Dict, task_metadata: Dict):
@@ -93,6 +122,7 @@ class DaskExecutor(AsyncBaseExecutor):
             app_log.debug("Task has cancelled")
             raise TaskCancelledError
 
+        dispatch_id = task_metadata["dispatch_id"]
         node_id = task_metadata["node_id"]
 
         dask_client = _address_client_mapper.get(self.scheduler_address)
@@ -102,7 +132,12 @@ class DaskExecutor(AsyncBaseExecutor):
             _address_client_mapper[self.scheduler_address] = dask_client
             await dask_client
 
-        future = dask_client.submit(dask_wrapper, function, args, kwargs)
+        if self.create_unique_workdir:
+            current_workdir = os.path.join(self.workdir, dispatch_id, f"node_{node_id}")
+        else:
+            current_workdir = self.workdir
+
+        future = dask_client.submit(dask_wrapper, function, args, kwargs, current_workdir)
         await self.set_job_handle(future.key)
         app_log.debug(f"Submitted task {node_id} to dask with key {future.key}")
 
diff --git a/covalent/executor/executor_plugins/local.py b/covalent/executor/executor_plugins/local.py
index f5e2497e..0a9ff389 100644
--- a/covalent/executor/executor_plugins/local.py
+++ b/covalent/executor/executor_plugins/local.py
@@ -27,10 +27,11 @@ This is a plugin executor module; it is loaded if found and properly structured.
 
 import os
 from concurrent.futures import ProcessPoolExecutor
-from typing import Any, Callable, Dict, List
+from typing import Any, Callable, Dict, List, Optional
 
 # Relative imports are not allowed in executor plugins
 from covalent._shared_files import TaskCancelledError, TaskRuntimeError, logger
+from covalent._shared_files.config import get_config
 from covalent.executor import BaseExecutor
 
 # Store the wrapper function in an external module to avoid module
@@ -49,6 +50,12 @@ _EXECUTOR_PLUGIN_DEFAULTS = {
     "cache_dir": os.path.join(
         os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"), "covalent"
     ),
+    "workdir": os.path.join(
+        os.environ.get("XDG_CACHE_HOME") or os.path.join(os.environ["HOME"], ".cache"),
+        "covalent",
+        "workdir",
+    ),
+    "create_unique_workdir": False,
 }
 
 proc_pool = ProcessPoolExecutor()
@@ -59,6 +66,30 @@ class LocalExecutor(BaseExecutor):
     Local executor class that directly invokes the input function.
     """
 
+    def __init__(
+        self, workdir: str = "", create_unique_workdir: Optional[bool] = None, *args, **kwargs
+    ) -> None:
+        if not workdir:
+            try:
+                workdir = get_config("executors.local.workdir")
+            except KeyError:
+                workdir = _EXECUTOR_PLUGIN_DEFAULTS["workdir"]
+                debug_msg = f"Couldn't find `executors.local.workdir` in config, using default value {workdir}."
+                app_log.debug(debug_msg)
+
+        if create_unique_workdir is None:
+            try:
+                create_unique_workdir = get_config("executors.local.create_unique_workdir")
+            except KeyError:
+                create_unique_workdir = _EXECUTOR_PLUGIN_DEFAULTS["create_unique_workdir"]
+                debug_msg = f"Couldn't find `executors.local.create_unique_workdir` in config, using default value {create_unique_workdir}."
+                app_log.debug(debug_msg)
+
+        super().__init__(*args, **kwargs)
+
+        self.workdir = workdir
+        self.create_unique_workdir = create_unique_workdir
+
     def run(self, function: Callable, args: List, kwargs: Dict, task_metadata: Dict) -> Any:
         """
         Execute the function locally
@@ -72,6 +103,7 @@ class LocalExecutor(BaseExecutor):
         Return(s)
             Task output
         """
+
         app_log.debug(f"Running function {function} locally")
 
         self.set_job_handle(42)
@@ -80,8 +112,16 @@ class LocalExecutor(BaseExecutor):
             app_log.debug("Task has been cancelled don't proceed")
             raise TaskCancelledError
 
+        dispatch_id = task_metadata["dispatch_id"]
+        node_id = task_metadata["node_id"]
+
+        if self.create_unique_workdir:
+            current_workdir = os.path.join(self.workdir, dispatch_id, f"node_{node_id}")
+        else:
+            current_workdir = self.workdir
+
         # Run the target function in a separate process
-        fut = proc_pool.submit(io_wrapper, function, args, kwargs)
+        fut = proc_pool.submit(io_wrapper, function, args, kwargs, current_workdir)
 
         output, worker_stdout, worker_stderr, tb = fut.result()
 
diff --git a/covalent/executor/utils/wrappers.py b/covalent/executor/utils/wrappers.py
index c39a3802..f38b8cc5 100644
--- a/covalent/executor/utils/wrappers.py
+++ b/covalent/executor/utils/wrappers.py
@@ -23,9 +23,11 @@ Helper functions for the local executor
 """
 
 import io
+import os
 import traceback
 from contextlib import redirect_stderr, redirect_stdout
 from enum import Enum
+from pathlib import Path
 from typing import Any, Callable, Dict, List, Tuple
 
 
@@ -39,14 +41,24 @@ class Signals(Enum):
     EXIT = 2
 
 
-def io_wrapper(fn: Callable, args: List, kwargs: Dict) -> Tuple[Any, str, str, str]:
+def io_wrapper(
+    fn: Callable,
+    args: List,
+    kwargs: Dict,
+    workdir: str = ".",
+) -> Tuple[Any, str, str, str]:
     """Wrapper function to execute the given function in a separate
     process and capture stdout and stderr"""
     with redirect_stdout(io.StringIO()) as stdout, redirect_stderr(io.StringIO()) as stderr:
         try:
+            Path(workdir).mkdir(parents=True, exist_ok=True)
+            current_dir = os.getcwd()
+            os.chdir(workdir)
             output = fn(*args, **kwargs)
             tb = ""
         except Exception as ex:
             output = None
             tb = "".join(traceback.TracebackException.from_exception(ex).format())
+        finally:
+            os.chdir(current_dir)
     return output, stdout.getvalue(), stderr.getvalue(), tb


Test patch:
diff --git a/tests/covalent_tests/executor/executor_plugins/dask_test.py b/tests/covalent_tests/executor/executor_plugins/dask_test.py
index 9f72a817..979249bf 100644
--- a/tests/covalent_tests/executor/executor_plugins/dask_test.py
+++ b/tests/covalent_tests/executor/executor_plugins/dask_test.py
@@ -21,23 +21,73 @@
 """Tests for Covalent dask executor."""
 
 import asyncio
+import os
+import tempfile
 from unittest.mock import AsyncMock
 
 import pytest
 
+import covalent as ct
 from covalent._shared_files import TaskRuntimeError
 from covalent._shared_files.exceptions import TaskCancelledError
-from covalent.executor.executor_plugins.dask import DaskExecutor
+from covalent.executor.executor_plugins.dask import _EXECUTOR_PLUGIN_DEFAULTS, DaskExecutor
 
 
 def test_dask_executor_init(mocker):
     """Test dask executor constructor"""
 
-    from covalent.executor import DaskExecutor
+    mocker.patch("covalent.executor.executor_plugins.dask.get_config", side_effect=KeyError())
+    default_workdir_path = _EXECUTOR_PLUGIN_DEFAULTS["workdir"]
 
     de = DaskExecutor("127.0.0.1")
 
     assert de.scheduler_address == "127.0.0.1"
+    assert de.workdir == default_workdir_path
+    assert de.create_unique_workdir is False
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        de = DaskExecutor("127.0.0.1", workdir=tmp_dir, create_unique_workdir=True)
+        assert de.scheduler_address == "127.0.0.1"
+        assert de.workdir == tmp_dir
+        assert de.create_unique_workdir is True
+
+
+def test_dask_executor_with_workdir(mocker):
+    from dask.distributed import LocalCluster
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        lc = LocalCluster()
+        de = ct.executor.DaskExecutor(
+            lc.scheduler_address, workdir=tmp_dir, create_unique_workdir=True
+        )
+
+        @ct.lattice
+        @ct.electron(executor=de)
+        def simple_task(x, y):
+            with open("job.txt", "w") as w:
+                w.write(str(x + y))
+            return "Done!"
+
+        mock_get_cancel_requested = mocker.patch.object(
+            de, "get_cancel_requested", AsyncMock(return_value=False)
+        )
+        mock_set_job_handle = mocker.patch.object(de, "set_job_handle", AsyncMock())
+
+        args = [1, 2]
+        kwargs = {}
+        task_metadata = {"dispatch_id": "asdf", "node_id": 1}
+        result = asyncio.run(de.run(simple_task, args, kwargs, task_metadata))
+        assert result == "Done!"
+
+        target_dir = os.path.join(
+            tmp_dir, task_metadata["dispatch_id"], f"node_{task_metadata['node_id']}"
+        )
+
+        assert os.listdir(target_dir) == ["job.txt"]
+        assert open(os.path.join(target_dir, "job.txt")).read() == "3"
+
+        mock_get_cancel_requested.assert_awaited()
+        mock_set_job_handle.assert_awaited()
 
 
 def test_dask_wrapper_fn(mocker):
diff --git a/tests/covalent_tests/executor/executor_plugins/local_test.py b/tests/covalent_tests/executor/executor_plugins/local_test.py
index c30a168a..82ce2d76 100644
--- a/tests/covalent_tests/executor/executor_plugins/local_test.py
+++ b/tests/covalent_tests/executor/executor_plugins/local_test.py
@@ -21,6 +21,7 @@
 """Tests for Covalent local executor."""
 
 import io
+import os
 import tempfile
 from functools import partial
 from unittest.mock import MagicMock
@@ -32,7 +33,56 @@ from covalent._shared_files import TaskRuntimeError
 from covalent._shared_files.exceptions import TaskCancelledError
 from covalent._workflow.transport import TransportableObject
 from covalent.executor.base import wrapper_fn
-from covalent.executor.executor_plugins.local import LocalExecutor
+from covalent.executor.executor_plugins.local import _EXECUTOR_PLUGIN_DEFAULTS, LocalExecutor
+
+
+def test_local_executor_init(mocker):
+    """Test local executor constructor"""
+
+    mocker.patch("covalent.executor.executor_plugins.local.get_config", side_effect=KeyError())
+    default_workdir_path = _EXECUTOR_PLUGIN_DEFAULTS["workdir"]
+
+    le = LocalExecutor()
+
+    assert le.workdir == default_workdir_path
+    assert le.create_unique_workdir is False
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        le = LocalExecutor(workdir=tmp_dir, create_unique_workdir=True)
+        assert le.workdir == tmp_dir
+        assert le.create_unique_workdir is True
+
+
+def test_local_executor_with_workdir(mocker):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        le = ct.executor.LocalExecutor(workdir=tmp_dir, create_unique_workdir=True)
+
+        @ct.lattice
+        @ct.electron(executor=le)
+        def simple_task(x, y):
+            with open("job.txt", "w") as w:
+                w.write(str(x + y))
+            return "Done!"
+
+        mock_set_job_handle = mocker.patch.object(le, "set_job_handle", MagicMock(return_value=42))
+        mock_get_cancel_requested = mocker.patch.object(
+            le, "get_cancel_requested", MagicMock(return_value=False)
+        )
+
+        args = [1, 2]
+        kwargs = {}
+        task_metadata = {"dispatch_id": "asdf", "node_id": 1}
+        assert le.run(simple_task, args, kwargs, task_metadata)
+
+        target_dir = os.path.join(
+            tmp_dir, task_metadata["dispatch_id"], f"node_{task_metadata['node_id']}"
+        )
+
+        assert os.listdir(target_dir) == ["job.txt"]
+        assert open(os.path.join(target_dir, "job.txt")).read() == "3"
+
+        mock_set_job_handle.assert_called_once()
+        mock_get_cancel_requested.assert_called_once()
 
 
 def test_local_executor_passes_results_dir(mocker):
diff --git a/tests/covalent_tests/shared_files/defaults_test.py b/tests/covalent_tests/shared_files/defaults_test.py
index be26fbfb..dc0d351b 100644
--- a/tests/covalent_tests/shared_files/defaults_test.py
+++ b/tests/covalent_tests/shared_files/defaults_test.py
@@ -43,7 +43,7 @@ def env_vars(mocker):
     }
 
 
-def test_get_default_sdk_config(env_vars, mocker):
+def test_get_default_sdk_config(env_vars):
     """Tests that the default SDK configuration is as expected"""
 
     with mock.patch.dict(os.environ, env_vars, clear=True):