Begin to generalize integration tests to other queue systems

tests/integration/conftest.py

@@ -51,8 +51,8 @@ def _get_random_name(length=6):
 
 
 @pytest.fixture(scope="session")
-def slurm_ssh_port():
-    """The exposed local port for SSH connections to the Slurm container."""
+def queue_ssh_port():
+    """The exposed local port for SSH connections to the queue container."""
     return _get_free_port()
 
 
@@ -134,10 +134,10 @@ def build_and_launch_container(
 
 
 @pytest.fixture(scope="session", autouse=True)
-def slurm_container(docker_client, slurm_ssh_port):
-    """Build and launch a container running Slurm + SSH, exposed on a random available
+def queue_container(docker_client, queue_ssh_port):
+    """Build and launch a container running various queues and SSH, exposed on a random available
     port."""
-    ports = {"22/tcp": slurm_ssh_port}
+    ports = {"22/tcp": queue_ssh_port}
     yield from build_and_launch_container(
         docker_client,
         Path("./tests/integration/dockerfiles/Dockerfile.slurm"),
@@ -168,7 +168,7 @@ def store_database_name():
 def write_tmp_settings(
     random_project_name,
     store_database_name,
-    slurm_ssh_port,
+    queue_ssh_port,
     db_port,
 ):
     """Collects the various sub-configs and writes them to a temporary file in a
@@ -223,10 +223,10 @@ def write_tmp_settings(
                 work_dir=str(workdir),
                 resources={},
             ),
-            "test_remote_worker": dict(
+            "test_remote_slurm_worker": dict(
                 type="remote",
                 host="localhost",
-                port=slurm_ssh_port,
+                port=queue_ssh_port,
                 scheduler_type="slurm",
                 work_dir="/home/jobflow/jfr",
                 user="jobflow",
@@ -235,10 +235,22 @@ def write_tmp_settings(
                 resources={"partition": "debug", "ntasks": 1, "time": "00:01:00"},
                 connect_kwargs={"allow_agent": False, "look_for_keys": False},
             ),
+            "test_remote_sge_worker": dict(
+                type="remote",
+                host="localhost",
+                port=queue_ssh_port,
+                scheduler_type="sge",
+                work_dir="/home/jobflow/jfr",
+                user="jobflow",
+                password="jobflow",
+                pre_run="source /home/jobflow/.venv/bin/activate",
+                resources={"partition": "debug", "ntasks": 1, "time": "00:01:00"},
+                connect_kwargs={"allow_agent": False, "look_for_keys": False},
+            ),
             "test_batch_remote_worker": dict(
                 type="remote",
                 host="localhost",
-                port=slurm_ssh_port,
+                port=queue_ssh_port,
                 scheduler_type="slurm",
                 work_dir="/home/jobflow/jfr",
                 user="jobflow",

tests/integration/dockerfiles/Dockerfile.slurm

@@ -1,7 +1,12 @@
 # syntax=docker/dockerfile:experimental
+ARG QUEUE_SYSTEM=slurm
+FROM ubuntu:22.04 as base
 
-# Using the slurm base image, run an ssh server and install jobflow
-FROM nathanhess/slurm:full AS base
+FROM nathanhess/slurm:full AS slurm
+
+FROM ${QUEUE_SYSTEM} as final
+
+# Run an SSH server and set up Python environment and user for jobflow
 ARG USERNAME=jobflow
 ARG PASSWORD=jobflow
 WORKDIR /opt

tests/integration/test_slurm.py

@@ -18,15 +18,15 @@ def test_project_init(random_project_name) -> None:
     assert len(project.workers) == 4
 
 
-def test_paramiko_ssh_connection(job_controller, slurm_ssh_port) -> None:
+def test_paramiko_ssh_connection(job_controller, queue_ssh_port) -> None:
     from paramiko import SSHClient
     from paramiko.client import WarningPolicy
 
     client = SSHClient()
     client.set_missing_host_key_policy(WarningPolicy)
     client.connect(
         "localhost",
-        port=slurm_ssh_port,
+        port=queue_ssh_port,
         username="jobflow",
         password="jobflow",
         look_for_keys=False,
@@ -39,7 +39,7 @@ def test_project_check(job_controller, capsys) -> None:
 
     expected = [
         "✓ Worker test_local_worker",
-        "✓ Worker test_remote_worker",
+        "✓ Worker test_remote_slurm_worker",
         "✓ Jobstore",
         "✓ Queue store",
     ]
@@ -48,7 +48,7 @@ def test_project_check(job_controller, capsys) -> None:
 
 @pytest.mark.parametrize(
     "worker",
-    ["test_local_worker", "test_remote_worker"],
+    ["test_local_worker", "test_remote_slurm_worker"],
 )
 def test_submit_flow(worker, job_controller) -> None:
     from jobflow import Flow
@@ -86,7 +86,7 @@ def test_submit_flow(worker, job_controller) -> None:
 
 @pytest.mark.parametrize(
     "worker",
-    ["test_local_worker", "test_remote_worker"],
+    ["test_local_worker", "test_remote_slurm_worker"],
 )
 def test_submit_flow_with_dependencies(worker, job_controller) -> None:
     from jobflow import Flow
@@ -132,7 +132,7 @@ def test_submit_flow_with_dependencies(worker, job_controller) -> None:
 
 @pytest.mark.parametrize(
     "worker",
-    ["test_local_worker", "test_remote_worker"],
+    ["test_local_worker", "test_remote_slurm_worker"],
 )
 def test_job_with_callable_kwarg(worker, job_controller) -> None:
     """Test whether a callable can be successfully provided as a keyword
@@ -176,7 +176,7 @@ def test_job_with_callable_kwarg(worker, job_controller) -> None:
 
 @pytest.mark.parametrize(
     "worker",
-    ["test_local_worker", "test_remote_worker"],
+    ["test_local_worker", "test_remote_slurm_worker"],
 )
 def test_expected_failure(worker, job_controller) -> None:
     from jobflow import Flow
@@ -205,7 +205,7 @@ def test_expected_failure(worker, job_controller) -> None:
 
 @pytest.mark.parametrize(
     "worker",
-    ["test_local_worker", "test_remote_worker"],
+    ["test_local_worker", "test_remote_slurm_worker"],
 )
 def test_exec_config(worker, job_controller, random_project_name) -> None:
     """Tests that an environment variable set in the exec config
@@ -233,7 +233,7 @@ def test_exec_config(worker, job_controller, random_project_name) -> None:
 
 @pytest.mark.parametrize(
     "worker",
-    ["test_local_worker", "test_remote_worker"],
+    ["test_local_worker", "test_remote_slurm_worker"],
 )
 def test_additional_stores(worker, job_controller) -> None:
     from jobflow import Flow
@@ -269,7 +269,7 @@ def test_additional_stores(worker, job_controller) -> None:
 
 @pytest.mark.parametrize(
     "worker",
-    ["test_local_worker", "test_remote_worker"],
+    ["test_local_worker", "test_remote_slurm_worker"],
 )
 def test_undefined_additional_stores(worker, job_controller) -> None:
     from jobflow import Flow