Add a test with expected job failures

src/jobflow_remote/testing/__init__.py

@@ -8,6 +8,12 @@ def add(a, b):
     return a + b
 
 
+@job
+def always_fails():
+    """A job that always fails."""
+    raise RuntimeError("This job failed.")
+
+
 @job
 def write_file(n):
     with open("results.txt", "w") as f:

tests/integration/conftest.py

@@ -95,7 +95,12 @@ def build_and_launch_container(
     try:
         print(f"\n * Launching container for {image_name}...")
         container = docker_client.containers.run(
-            image_name, detach=True, remove=True, tty=True, ports=ports
+            image_name,
+            detach=True,
+            remove=True,
+            auto_remove=True,
+            tty=True,
+            ports=ports,
         )
         assert isinstance(container, Container)
         print(" * Waiting for container to be ready...", end="")

tests/integration/test_slurm.py

@@ -34,8 +34,9 @@ def test_submit_flow(worker, job_controller):
     runner = Runner()
     runner.run(ticks=10)
 
-    jobs = job_controller.get_jobs({})
-    assert len(jobs) == 2
+    assert job_controller.count_jobs({}) == 2
+    assert len(job_controller.get_jobs({})) == 2
+    assert job_controller.count_flows({}) == 1
 
 
 @pytest.mark.parametrize("worker", ["test_local_worker", "test_remote_worker"])
@@ -57,4 +58,37 @@ def test_submit_flow_with_dependencies(worker, job_controller):
     runner = Runner()
     runner.run(ticks=10)
 
+    assert job_controller.count_jobs({}) == 4
     assert len(job_controller.get_jobs({})) == 4
+    assert job_controller.count_flows({}) == 1
+
+
+@pytest.mark.parametrize("worker", ["test_local_worker", "test_remote_worker"])
+def test_expected_failure(worker, job_controller):
+    from jobflow import Flow
+
+    from jobflow_remote import submit_flow
+    from jobflow_remote.jobs.runner import Runner
+    from jobflow_remote.jobs.state import FlowState, JobState
+    from jobflow_remote.testing import always_fails
+
+    job_1 = always_fails()
+    job_2 = always_fails()
+
+    flow = Flow([job_1, job_2])
+    submit_flow(flow, worker=worker)
+
+    runner = Runner()
+    runner.run(ticks=10)
+
+    assert job_controller.count_jobs({}) == 2
+    assert len(job_controller.get_jobs({})) == 2
+    assert job_controller.count_flows({}) == 1
+
+    if worker == "test_local_worker":
+        assert job_controller.count_jobs(state=JobState.FAILED) == 2
+        assert job_controller.count_flows(state=FlowState.FAILED) == 1
+    else:
+        # The remote flow will be left in a running state until the remote errors can be rectified
+        assert job_controller.count_jobs(state=JobState.REMOTE_ERROR) == 2
+        assert job_controller.count_flows(state=FlowState.RUNNING) == 1