Repository: AgnostiqHQ/covalent (language: python)

Issue:
Executor Resource Deployment: CloudResourceManager Implementation
## Overview

The goal is to delegate the logic for spinning our resources for each plugin to the `CloudResourceManager`

The responsibility of the `CloudResourceManager` will be to:
- Provision and tear down resources of each plugin by resolving the terraform scripts located in the `infra` repo of each plugin
- To validate provided arguments of the infra (terraform variables) and also executor arguments to ensure they are valid (using Pydantic models from each plugin repo)
- To validate the user's environment to make sure they have the correct tools installed such as `terraform`, and `docker`
- To write to covalent config all of the required arguments for a given executor plugin
- To provide the necessary logging and error messages for the CLI to utilize
- To query the status of each installed plugin to determine if a particular plugin has been: deployed, not-deployed/torn down, or in a dirty state (partially torn down or partially deployed)

```
class CloudResourceManager:
    """
    Base cloud resource manager class
    """
    def __init__(
        self,
        executor_name: str,
        executor_module_path: str,
        options: Optional[Dict[str, str]] = None,
    ):
        self.executor_name = executor_name
        self.executor_tf_path = os.path.join(executor_module_path, "assets/infra")
        self.executor_options = options

    def up(self):
       ...

    def down(self):
        ...

    def status(self):
        ...
```

### Acceptance Criteria

- [ ] Implement the `up`, `down`, and `status` methods of the `CloudResourceManager` according to the interface above
- [ ] Ensure that all user provided arguments are validated according to the schema provided by each plugin's pydantic models that contain Terraform args and Executor args. Also support catching and gracefully handling any custom validators added to pydantic models
- [ ] Well defined custom error classes are thrown that indicate either validation errors, unsupported user environment (missing tools like terraform, docker), failed teardown or deployments

Refer to the following existing branch: https://github.com/AgnostiqHQ/covalent/pull/1556/files


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/requirements.yml b/.github/workflows/requirements.yml
index a4bd6cbb6..d9bb5e71a 100644
--- a/.github/workflows/requirements.yml
+++ b/.github/workflows/requirements.yml
@@ -53,6 +53,7 @@ jobs:
           --ignore-module=pkg_resources
           --ignore-file=covalent/executor/**
           --ignore-file=covalent/triggers/**
+          --ignore-file=covalent/cloud_resource_manager/**
           covalent
 
       - name: Check missing dispatcher requirements
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c26701d6a..9885fb3a1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [UNRELEASED]
 
+### Added
+
+- Added the `CloudResourceManager` class
+
+### Tests
+
+- Added tests for the `CloudResourceManager` class
+
 ### Docs
 
 - Redispatch feature page in Read the Docs.
diff --git a/covalent/_shared_files/exceptions.py b/covalent/_shared_files/exceptions.py
index 3b57c52cb..c7a097975 100644
--- a/covalent/_shared_files/exceptions.py
+++ b/covalent/_shared_files/exceptions.py
@@ -29,3 +29,7 @@ class TaskRuntimeError(Exception):
 
 class TaskCancelledError(Exception):
     pass
+
+
+class CommandNotFoundError(Exception):
+    pass
diff --git a/covalent/cloud_resource_manager/__init__.py b/covalent/cloud_resource_manager/__init__.py
new file mode 100644
index 000000000..d1f2c47e0
--- /dev/null
+++ b/covalent/cloud_resource_manager/__init__.py
@@ -0,0 +1,21 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+from .core import CloudResourceManager
diff --git a/covalent/cloud_resource_manager/core.py b/covalent/cloud_resource_manager/core.py
new file mode 100644
index 000000000..94947afdd
--- /dev/null
+++ b/covalent/cloud_resource_manager/core.py
@@ -0,0 +1,317 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+
+import importlib
+import os
+import shutil
+import subprocess
+from configparser import ConfigParser
+from pathlib import Path
+from types import ModuleType
+from typing import Dict, Optional
+
+from covalent._shared_files.config import get_config, set_config
+from covalent._shared_files.exceptions import CommandNotFoundError
+from covalent.executor import _executor_manager
+
+
+def get_executor_module(executor_name: str) -> ModuleType:
+    """
+    Get the executor module from the executor name
+
+    Args:
+        executor_name: Name of the executor
+
+    Returns:
+        The executor module
+
+    """
+    return importlib.import_module(
+        _executor_manager.executor_plugins_map[executor_name].__module__
+    )
+
+
+def validate_options(
+    ExecutorPluginDefaults, ExecutorInfraDefaults, executor_options: Dict[str, str]
+) -> None:
+    """
+    Validate the options passed to the CRM
+
+    Args:
+        ExecutorPluginDefaults: Executor plugin defaults validation class
+        ExecutorInfraDefaults: Executor infra defaults validation class
+        executor_options: Options passed to the CRM
+
+    Raises:
+        pydantic.ValidationError: If the options are invalid
+
+    """
+
+    # Validating the passed options:
+
+    plugin_attrs = list(ExecutorPluginDefaults.schema()["properties"].keys())
+    infra_attrs = list(ExecutorInfraDefaults.schema()["properties"].keys())
+
+    plugin_params = {k: v for k, v in executor_options.items() if k in plugin_attrs}
+    infra_params = {k: v for k, v in executor_options.items() if k in infra_attrs}
+
+    # Validate options
+    ExecutorPluginDefaults(**plugin_params)
+    ExecutorInfraDefaults(**infra_params)
+
+
+class CloudResourceManager:
+    """
+    Base cloud resource manager class
+    """
+
+    def __init__(
+        self,
+        executor_name: str,
+        executor_module_path: str,
+        options: Optional[Dict[str, str]] = None,
+    ):
+        self.executor_name = executor_name
+        self.executor_tf_path = str(
+            Path(executor_module_path).expanduser().resolve() / "assets" / "infra"
+        )
+
+        # Includes both plugin and infra options
+        self.executor_options = options
+
+        # Importing validation classes from the executor module
+        module = get_executor_module(executor_name)
+        self.ExecutorPluginDefaults = getattr(module, "ExecutorPluginDefaults")
+        self.ExecutorInfraDefaults = getattr(module, "ExecutorInfraDefaults")
+
+        if self.executor_options:
+            validate_options(
+                self.ExecutorPluginDefaults, self.ExecutorInfraDefaults, self.executor_options
+            )
+
+    def _print_stdout(self, process: subprocess.Popen) -> int:
+        """
+        Print the stdout from the subprocess to console
+
+        Args:
+            process: Python subprocess whose stdout is to be printed to screen
+
+        Returns:
+            returncode of the process
+        """
+
+        while process.poll() is None:
+            if proc_stdout := process.stdout.readline():
+                print(proc_stdout.strip().decode("utf-8"))
+            else:
+                break
+        return process.poll()
+
+        # TODO: Return the command output alongwith returncode
+
+    def _run_in_subprocess(
+        self, cmd: str, workdir: str, env_vars: Optional[Dict[str, str]] = None
+    ) -> None:
+        """
+        Run the `cmd` in a subprocess shell with the env_vars set in the process's new environment
+
+        Args:
+            cmd: Command to execute in the subprocess
+            workdir: Working directory of the subprocess
+            env_vars: Dictionary of environment variables to set in the processes execution environment
+
+        Returns:
+            None
+
+        """
+        proc = subprocess.Popen(
+            args=cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            cwd=workdir,
+            shell=True,
+            env=env_vars,
+        )
+        retcode = self._print_stdout(proc)
+
+        if retcode != 0:
+            raise subprocess.CalledProcessError(returncode=retcode, cmd=cmd)
+
+        # TODO: Return the command output
+
+    def _update_config(self, tf_executor_config_file: str) -> None:
+        """
+        Update covalent configuration with the executor
+        config values as obtained from terraform
+
+        Args:
+            tf_executor_config_file: Path to the terraform executor config file
+
+        Returns:
+            None
+
+        """
+        # Puts the plugin options in covalent's config
+        executor_config = ConfigParser()
+        executor_config.read(tf_executor_config_file)
+
+        filtered_executor_config = {
+            k: v if v != "null" else None for k, v in executor_config[self.executor_name].items()
+        }
+
+        validated_config = self.ExecutorPluginDefaults(**filtered_executor_config).dict()
+
+        for key, value in validated_config.items():
+            set_config({f"executors.{self.executor_name}.{key}": value})
+
+    def _get_tf_path(self) -> str:
+        """
+        Get the terraform path
+
+        Args:
+            None
+
+        Returns:
+            Path to terraform executable
+
+        """
+        if terraform := shutil.which("terraform"):
+            return terraform
+        else:
+            raise CommandNotFoundError("Terraform not found on system")
+
+    def _get_tf_statefile_path(self) -> str:
+        """
+        Get the terraform state file path
+
+        Args:
+            None
+
+        Returns:
+            Path to terraform state file
+
+        """
+        # Saving in a directory which doesn't get deleted on purge
+        return str(Path(get_config("dispatcher.db_path")).parent / f"{self.executor_name}.tfstate")
+
+    def up(self, dry_run: bool = True) -> None:
+        """
+        Spin up executor resources with terraform
+
+        Args:
+            dry_run: If True, only run terraform plan and not apply
+
+        Returns:
+            None
+
+        """
+        terraform = self._get_tf_path()
+        tf_state_file = self._get_tf_statefile_path()
+
+        tfvars_file = str(Path(self.executor_tf_path) / "terraform.tfvars")
+        tf_executor_config_file = str(Path(self.executor_tf_path) / f"{self.executor_name}.conf")
+
+        tf_init = " ".join([terraform, "init"])
+        tf_plan = " ".join([terraform, "plan", "-out", "tf.plan", f"-state={tf_state_file}"])
+        tf_apply = " ".join([terraform, "apply", "tf.plan", f"-state={tf_state_file}"])
+
+        # Run `terraform init`
+        self._run_in_subprocess(cmd=tf_init, workdir=self.executor_tf_path)
+
+        # Setup terraform infra variables as passed by the user
+        tf_vars_env_dict = os.environ.copy()
+        if self.executor_options:
+            with open(tfvars_file, "w") as f:
+                for key, value in self.executor_options.items():
+                    tf_vars_env_dict[f"TF_VAR_{key}"] = value
+
+                    # Write whatever the user has passed to the terraform.tfvars file
+                    f.write(f'{key}="{value}"\n')
+
+        # Run `terraform plan`
+        cmd_output = self._run_in_subprocess(
+            cmd=tf_plan, workdir=self.executor_tf_path, env_vars=tf_vars_env_dict
+        )
+
+        # Create infrastructure as per the plan
+        # Run `terraform apply`
+        if not dry_run:
+            cmd_output = self._run_in_subprocess(
+                cmd=tf_apply, workdir=self.executor_tf_path, env_vars=tf_vars_env_dict
+            )
+
+            # Update covalent executor config based on Terraform output
+            self._update_config(tf_executor_config_file)
+
+        return cmd_output
+
+    def down(self) -> None:
+        """
+        Teardown previously spun up executor resources with terraform
+
+        Args:
+            None
+
+        Returns:
+            None
+
+        """
+        terraform = self._get_tf_path()
+        tf_state_file = self._get_tf_statefile_path()
+
+        tfvars_file = str(Path(self.executor_tf_path) / "terraform.tfvars")
+
+        tf_destroy = " ".join([terraform, "destroy", "-auto-approve", f"-state={tf_state_file}"])
+
+        # Run `terraform destroy`
+        cmd_output = self._run_in_subprocess(cmd=tf_destroy, workdir=self.executor_tf_path)
+
+        if Path(tfvars_file).exists():
+            Path(tfvars_file).unlink()
+
+        if Path(tf_state_file).exists():
+            Path(tf_state_file).unlink()
+            Path(f"{tf_state_file}.backup").unlink()
+
+        return cmd_output
+
+    def status(self) -> None:
+        """
+        Get the status of the spun up executor resources
+
+        TODO: Return the list of resources being managed by terraform, i.e.
+        if empty, then either the resources have not been created or
+        have been destroyed already.
+
+        Args:
+            None
+
+        Returns:
+            None
+
+        """
+        terraform = self._get_tf_path()
+        tf_state_file = self._get_tf_statefile_path()
+
+        tf_state = " ".join([terraform, "state", "list", f"-state={tf_state_file}"])
+
+        # Run `terraform state list`
+        return self._run_in_subprocess(cmd=tf_state, workdir=self.executor_tf_path)


Test patch:
diff --git a/tests/covalent_tests/cloud_resource_manager/__init__.py b/tests/covalent_tests/cloud_resource_manager/__init__.py
new file mode 100644
index 000000000..9d1b05526
--- /dev/null
+++ b/tests/covalent_tests/cloud_resource_manager/__init__.py
@@ -0,0 +1,19 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
diff --git a/tests/covalent_tests/cloud_resource_manager/core_test.py b/tests/covalent_tests/cloud_resource_manager/core_test.py
new file mode 100644
index 000000000..da675f9e1
--- /dev/null
+++ b/tests/covalent_tests/cloud_resource_manager/core_test.py
@@ -0,0 +1,488 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+
+import subprocess
+from configparser import ConfigParser
+from functools import partial
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+from covalent._shared_files.exceptions import CommandNotFoundError
+from covalent.cloud_resource_manager.core import (
+    CloudResourceManager,
+    get_executor_module,
+    validate_options,
+)
+
+
+@pytest.fixture
+def executor_name():
+    return "test_executor"
+
+
+@pytest.fixture
+def executor_module_path():
+    return "test_executor_module_path"
+
+
+@pytest.fixture
+def crm(mocker, executor_name, executor_module_path):
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.get_executor_module",
+    )
+
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.getattr",
+    )
+
+    return CloudResourceManager(
+        executor_name=executor_name,
+        executor_module_path=executor_module_path,
+    )
+
+
+def test_get_executor_module(mocker):
+    """
+    Unit test for get_executor_module method
+    """
+
+    test_executor_name = "test_executor"
+    test_executor_module = "test_executor_module"
+
+    mock_import_module = mocker.patch(
+        "covalent.cloud_resource_manager.core.importlib.import_module",
+        return_value=test_executor_module,
+    )
+
+    mocker.patch(
+        "covalent.cloud_resource_manager.core._executor_manager.executor_plugins_map",
+        return_value={test_executor_name: "test"},
+    )
+
+    returned_module = get_executor_module(test_executor_name)
+
+    assert returned_module == test_executor_module
+    mock_import_module.assert_called_once()
+
+
+def test_validate_options(mocker):
+    """
+    Unit test for validate_options method
+    """
+
+    executor_options = {"test_key": "test_value"}
+
+    mock_defaults_model = mocker.MagicMock()
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.getattr",
+        return_value=mock_defaults_model,
+    )
+
+    mock_list = mocker.patch(
+        "covalent.cloud_resource_manager.core.list",
+        return_value=list(executor_options.keys()),
+    )
+
+    validate_options(mock_defaults_model, mock_defaults_model, executor_options)
+
+    assert mock_list.call_count == 2
+
+    mock_defaults_model.assert_any_call(**executor_options)
+    assert mock_defaults_model.call_count == 2
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        None,
+        {"test_key": "test_value"},
+    ],
+)
+def test_cloud_resource_manager_init(mocker, options, executor_name, executor_module_path):
+    """
+    Unit test for CloudResourceManager's init method
+    """
+
+    mock_validate_options = mocker.patch(
+        "covalent.cloud_resource_manager.core.validate_options",
+    )
+
+    mock_get_executor_module = mocker.patch(
+        "covalent.cloud_resource_manager.core.get_executor_module",
+    )
+
+    mock_model_class = mocker.MagicMock()
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.getattr",
+        return_value=mock_model_class,
+    )
+
+    crm = CloudResourceManager(
+        executor_name=executor_name,
+        executor_module_path=executor_module_path,
+        options=options,
+    )
+
+    assert crm.executor_name == executor_name
+    assert crm.executor_tf_path == str(
+        Path(executor_module_path).expanduser().resolve() / "assets" / "infra"
+    )
+
+    mock_get_executor_module.assert_called_once_with(executor_name)
+    assert crm.executor_options == options
+
+    if options:
+        mock_validate_options.assert_called_once_with(mock_model_class, mock_model_class, options)
+    else:
+        mock_validate_options.assert_not_called()
+
+
+def test_print_stdout(mocker, crm):
+    """
+    Unit test for CloudResourceManager._print_stdout() method
+    """
+
+    test_stdout = "test_stdout".encode("utf-8")
+    test_return_code = 0
+
+    mock_process = mocker.MagicMock()
+
+    mock_process.poll.side_effect = partial(next, iter([None, test_return_code, test_return_code]))
+    mock_process.stdout.readline.side_effect = partial(next, iter([test_stdout, None]))
+
+    mock_print = mocker.patch("covalent.cloud_resource_manager.core.print")
+
+    return_code = crm._print_stdout(mock_process)
+
+    mock_process.stdout.readline.assert_called_once()
+    mock_print.assert_called_once_with(test_stdout.decode("utf-8"))
+    assert mock_process.poll.call_count == 3
+    assert return_code == test_return_code
+
+
+@pytest.mark.parametrize(
+    "test_retcode",
+    [
+        0,
+        1,
+    ],
+)
+def test_run_in_subprocess(mocker, test_retcode, crm):
+    """
+    Unit test for CloudResourceManager._run_in_subprocess() method.
+    """
+
+    test_cmd = "test_cmd"
+    test_workdir = "test_workdir"
+    test_env_vars = {"test_env_key": "test_env_value"}
+
+    mock_process = mocker.MagicMock()
+    mock_popen = mocker.patch(
+        "covalent.cloud_resource_manager.core.subprocess.Popen",
+        return_value=mock_process,
+    )
+
+    mock_print_stdout = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._print_stdout",
+        return_value=test_retcode,
+    )
+
+    if test_retcode != 0:
+        exception = subprocess.CalledProcessError(returncode=test_retcode, cmd=test_cmd)
+        with pytest.raises(Exception, match=str(exception)):
+            crm._run_in_subprocess(
+                cmd=test_cmd,
+                workdir=test_workdir,
+                env_vars=test_env_vars,
+            )
+    else:
+        crm._run_in_subprocess(
+            cmd=test_cmd,
+            workdir=test_workdir,
+            env_vars=test_env_vars,
+        )
+
+    mock_popen.assert_called_once_with(
+        args=test_cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        cwd=test_workdir,
+        shell=True,
+        env=test_env_vars,
+    )
+
+    mock_print_stdout.assert_called_once_with(mock_process)
+
+
+def test_update_config(mocker, crm, executor_name):
+    """
+    Unit test for CloudResourceManager._update_config() method.
+    """
+
+    test_tf_executor_config_file = "test_tf_executor_config_file"
+    test_key = "test_key"
+    test_value = "test_value"
+
+    test_config_parser = ConfigParser()
+    test_config_parser[executor_name] = {test_key: test_value}
+    test_config_parser.read = mocker.MagicMock()
+
+    crm.ExecutorPluginDefaults = mocker.MagicMock()
+    crm.ExecutorPluginDefaults.return_value.dict.return_value = {test_key: test_value}
+
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.ConfigParser",
+        return_value=test_config_parser,
+    )
+
+    mock_set_config = mocker.patch(
+        "covalent.cloud_resource_manager.core.set_config",
+    )
+
+    crm._update_config(
+        tf_executor_config_file=test_tf_executor_config_file,
+    )
+
+    test_config_parser.read.assert_called_once_with(test_tf_executor_config_file)
+    mock_set_config.assert_called_once_with(
+        {f"executors.{executor_name}.{test_key}": test_value},
+    )
+
+
+@pytest.mark.parametrize(
+    "test_tf_path",
+    [
+        "test_tf_path",
+        None,
+    ],
+)
+def test_get_tf_path(mocker, test_tf_path, crm):
+    """
+    Unit test for CloudResourceManager._get_tf_path() method
+    """
+
+    mock_shutil_which = mocker.patch(
+        "covalent.cloud_resource_manager.core.shutil.which",
+        return_value=test_tf_path,
+    )
+
+    if test_tf_path:
+        assert crm._get_tf_path() == test_tf_path
+    else:
+        with pytest.raises(CommandNotFoundError, match="Terraform not found on system"):
+            crm._get_tf_path()
+
+    mock_shutil_which.assert_called_once_with("terraform")
+
+
+def test_get_tf_statefile_path(mocker, crm, executor_name):
+    """
+    Unit test for CloudResourceManager._get_tf_statefile_path() method
+    """
+
+    test_tf_state_file = "test_tf_state_file"
+
+    mock_get_config = mocker.patch(
+        "covalent.cloud_resource_manager.core.get_config",
+        return_value=test_tf_state_file,
+    )
+
+    assert crm._get_tf_statefile_path() == f"{executor_name}.tfstate"
+
+    mock_get_config.assert_called_once_with("dispatcher.db_path")
+
+
+@pytest.mark.parametrize(
+    "dry_run, executor_options",
+    [
+        (True, None),
+        (False, {"test_key": "test_value"}),
+    ],
+)
+def test_up(mocker, dry_run, executor_options, executor_name, executor_module_path):
+    """
+    Unit test for CloudResourceManager.up() method
+    """
+
+    test_tf_path = "test_tf_path"
+    test_tf_state_file = "test_tf_state_file"
+
+    mock_get_tf_path = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._get_tf_path",
+        return_value=test_tf_path,
+    )
+
+    mock_get_tf_statefile_path = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._get_tf_statefile_path",
+        return_value=test_tf_state_file,
+    )
+
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.get_executor_module",
+    )
+
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.getattr",
+    )
+
+    # Mocking as we are instantiating with executor options
+    mocker.patch(
+        "covalent.cloud_resource_manager.core.validate_options",
+    )
+
+    mock_run_in_subprocess = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._run_in_subprocess",
+    )
+
+    test_tf_dict = {"test_tf_key": "test_tf_value"}
+    mock_environ_copy = mocker.patch(
+        "covalent.cloud_resource_manager.core.os.environ.copy",
+        return_value=test_tf_dict,
+    )
+
+    mock_update_config = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._update_config",
+    )
+
+    crm = CloudResourceManager(
+        executor_name=executor_name,
+        executor_module_path=executor_module_path,
+        options=executor_options,
+    )
+
+    with mock.patch(
+        "covalent.cloud_resource_manager.core.open",
+        mock.mock_open(),
+    ) as mock_file:
+        crm.up(dry_run=dry_run)
+
+    mock_get_tf_path.assert_called_once()
+    mock_get_tf_statefile_path.assert_called_once()
+    mock_run_in_subprocess.assert_any_call(
+        cmd=f"{test_tf_path} init",
+        workdir=crm.executor_tf_path,
+    )
+
+    mock_environ_copy.assert_called_once()
+
+    if executor_options:
+        mock_file.assert_called_once_with(
+            f"{crm.executor_tf_path}/terraform.tfvars",
+            "w",
+        )
+
+        key, value = list(executor_options.items())[0]
+        mock_file().write.assert_called_once_with(f'{key}="{value}"\n')
+
+    mock_run_in_subprocess.assert_any_call(
+        cmd=f"{test_tf_path} plan -out tf.plan -state={test_tf_state_file}",
+        workdir=crm.executor_tf_path,
+        env_vars=test_tf_dict,
+    )
+
+    if not dry_run:
+        mock_run_in_subprocess.assert_any_call(
+            cmd=f"{test_tf_path} apply tf.plan -state={test_tf_state_file}",
+            workdir=crm.executor_tf_path,
+            env_vars=test_tf_dict,
+        )
+
+        mock_update_config.assert_called_once_with(
+            f"{crm.executor_tf_path}/{executor_name}.conf",
+        )
+
+
+def test_down(mocker, crm):
+    """
+    Unit test for CloudResourceManager.down() method.
+    """
+
+    test_tf_path = "test_tf_path"
+    test_tf_state_file = "test_tf_state_file"
+
+    mock_get_tf_path = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._get_tf_path",
+        return_value=test_tf_path,
+    )
+
+    mock_get_tf_statefile_path = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._get_tf_statefile_path",
+        return_value=test_tf_state_file,
+    )
+
+    mock_run_in_subprocess = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._run_in_subprocess",
+    )
+
+    mock_path_exists = mocker.patch(
+        "covalent.cloud_resource_manager.core.Path.exists",
+        return_value=True,
+    )
+
+    mock_path_unlink = mocker.patch(
+        "covalent.cloud_resource_manager.core.Path.unlink",
+    )
+
+    crm.down()
+
+    mock_get_tf_path.assert_called_once()
+    mock_get_tf_statefile_path.assert_called_once()
+    mock_run_in_subprocess.assert_called_once_with(
+        cmd=f"{mock_get_tf_path.return_value} destroy -auto-approve -state={test_tf_state_file}",
+        workdir=crm.executor_tf_path,
+    )
+
+    assert mock_path_exists.call_count == 2
+    assert mock_path_unlink.call_count == 3
+
+
+def test_status(mocker, crm):
+    """
+    Unit test for CloudResourceManager.status() method.
+    """
+
+    test_tf_path = "test_tf_path"
+    test_tf_state_file = "test_tf_state_file"
+
+    mock_get_tf_path = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._get_tf_path",
+        return_value=test_tf_path,
+    )
+
+    mock_get_tf_statefile_path = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._get_tf_statefile_path",
+        return_value=test_tf_state_file,
+    )
+
+    mock_run_in_subprocess = mocker.patch(
+        "covalent.cloud_resource_manager.core.CloudResourceManager._run_in_subprocess",
+    )
+
+    crm.status()
+
+    mock_get_tf_path.assert_called_once()
+    mock_get_tf_statefile_path.assert_called_once()
+    mock_run_in_subprocess.assert_called_once_with(
+        cmd=f"{test_tf_path} state list -state={test_tf_state_file}",
+        workdir=crm.executor_tf_path,
+    )