Fix type hints and abstract classes for ComputeEnvironment

This commit is contained in:
mvdbeek
2021-11-29 11:13:08 +01:00
parent a2d9ce67b3
commit 7ec0963a7c
3 changed files with 8 additions and 4 deletions
+5 -1
View File
@@ -2475,6 +2475,10 @@ class ComputeEnvironment(metaclass=ABCMeta):
def config_directory(self):
""" Directory containing config files (potentially remote) """
@abstractmethod
def env_config_directory(self):
"""Working directory (possibly as environment variable evaluation)."""
@abstractmethod
def sep(self):
""" os.path.sep for the platform this job will execute in.
@@ -2514,7 +2518,7 @@ class SimpleComputeEnvironment:
return os.path.sep
class SharedComputeEnvironment(SimpleComputeEnvironment):
class SharedComputeEnvironment(SimpleComputeEnvironment, ComputeEnvironment):
""" Default ComputeEnvironment for job and task wrapper to pass
to ToolEvaluator - valid when Galaxy and compute share all the relevant
file systems.
+2 -2
View File
@@ -9,7 +9,7 @@ from typing import Any, Dict, List
from galaxy import model
from galaxy.files import ProvidesUserFileSourcesUserContext
from galaxy.job_execution.setup import ensure_configs_directory
from galaxy.jobs import SharedComputeEnvironment, SimpleComputeEnvironment
from galaxy.jobs import ComputeEnvironment
from galaxy.model.none_like import NoneDataset
from galaxy.security.object_wrapper import wrap_with_safe_string
from galaxy.tools import global_tool_errors
@@ -103,7 +103,7 @@ class ToolEvaluator:
self.tool = tool
self.local_working_directory = local_working_directory
def set_compute_environment(self, compute_environment: Union[SimpleComputeEnvironment, SharedComputeEnvironment], get_special: Optional[Callable] = None):
def set_compute_environment(self, compute_environment: ComputeEnvironment, get_special: Optional[Callable] = None):
"""
Setup the compute environment and established the outline of the param_dict
for evaluating command and config cheetah templates.
+1 -1
View File
@@ -210,7 +210,7 @@ class ToolEvaluatorTestCase(TestCase, UsesApp):
kwds["working_directory"] = self.test_directory
if "new_file_path" not in kwds:
kwds["new_file_path"] = self.app.config.new_file_path
self.evaluator.set_compute_environment(ComputeEnvironment(**kwds))
self.evaluator.set_compute_environment(ComputeEnvironment(**kwds)) # type: ignore[arg-type]
assert "exec_before_job" in self.tool.hooks_called
def _setup_test_bwa_job(self):