moved methods and template

tests/conftest.py

@@ -13,11 +13,18 @@
 from streamflow.core.deployment import (
     LOCAL_LOCATION,
     Location,
+    DeploymentConfig,
 )
 from streamflow.core.persistence import PersistableEntity
 from streamflow.main import build_context
 from streamflow.persistence.loading_context import DefaultDatabaseLoadingContext
-from tests.utils.get_instances import get_deployment_config
+from tests.utils.get_instances import (
+    get_local_deployment_config,
+    get_docker_deployment_config,
+    get_kubernetes_deployment_config,
+    get_singularity_deployment_config,
+    get_ssh_deployment_config,
+)
 
 
 def csvtype(choices):
@@ -51,6 +58,38 @@ def chosen_deployment_types(request):
     return request.config.getoption("--deploys")
 
 
+async def get_deployment_config(
+    _context: StreamFlowContext, deployment_t: str
+) -> DeploymentConfig:
+    if deployment_t == "local":
+        return get_local_deployment_config()
+    elif deployment_t == "docker":
+        return get_docker_deployment_config()
+    elif deployment_t == "kubernetes":
+        return get_kubernetes_deployment_config()
+    elif deployment_t == "singularity":
+        return get_singularity_deployment_config()
+    elif deployment_t == "ssh":
+        return await get_ssh_deployment_config(_context)
+    else:
+        raise Exception(f"{deployment_t} deployment type not supported")
+
+
+def get_service(_context: StreamFlowContext, deployment_t: str) -> str | None:
+    if deployment_t == "local":
+        return None
+    elif deployment_t == "docker":
+        return None
+    elif deployment_t == "kubernetes":
+        return "sf-test"
+    elif deployment_t == "singularity":
+        return None
+    elif deployment_t == "ssh":
+        return None
+    else:
+        raise Exception(f"{deployment_t} deployment type not supported")
+
+
 def pytest_generate_tests(metafunc):
     if "deployment" in metafunc.fixturenames:
         metafunc.parametrize(

tests/test_scheduler.py

@@ -19,10 +19,9 @@
 from streamflow.core.scheduling import AvailableLocation, Hardware
 from streamflow.core.workflow import Job, Status
 from streamflow.deployment.connector import LocalConnector
+from tests.conftest import get_deployment_config, get_service
 from tests.utils.get_instances import (
     get_docker_deployment_config,
-    get_deployment_config,
-    get_service,
 )
 
 

tests/utils/get_instances.py

@@ -35,40 +35,8 @@ def get_docker_deployment_config():
     )
 
 
-async def get_deployment_config(
-    _context: StreamFlowContext, deployment_t: str
-) -> DeploymentConfig:
-    if deployment_t == "local":
-        return get_local_deployment_config()
-    elif deployment_t == "docker":
-        return get_docker_deployment_config()
-    elif deployment_t == "kubernetes":
-        return get_kubernetes_deployment_config()
-    elif deployment_t == "singularity":
-        return get_singularity_deployment_config()
-    elif deployment_t == "ssh":
-        return await get_ssh_deployment_config(_context)
-    else:
-        raise Exception(f"{deployment_t} deployment type not supported")
-
-
-def get_service(_context: StreamFlowContext, deployment_t: str) -> str | None:
-    if deployment_t == "local":
-        return None
-    elif deployment_t == "docker":
-        return None
-    elif deployment_t == "kubernetes":
-        return "sf-test"
-    elif deployment_t == "singularity":
-        return None
-    elif deployment_t == "ssh":
-        return None
-    else:
-        raise Exception(f"{deployment_t} deployment type not supported")
-
-
 def get_kubernetes_deployment_config():
-    with open(pkg_resources.resource_filename(__name__, "./pod.jinja2")) as t:
+    with open(pkg_resources.resource_filename(__name__, "pod.jinja2")) as t:
         template = Template(t.read())
     with tempfile.NamedTemporaryFile(mode="w", delete=False) as f:
         template.stream(name=utils.random_name()).dump(f.name)