Basic job state recovery tests.

src/tomato/daemon/cmd.py

@@ -15,7 +15,7 @@ def merge_pipelines(
             if pip.jobid is not None:
                 ret[pname] = pip
         else:
-            if pip == new[pname]:
+            if pip.devs == new[pname].devs:
                 ret[pname] = pip
             elif pip.jobid is None:
                 ret[pname] = new[pname]

src/tomato/daemon/io.py

@@ -23,3 +23,4 @@ def load(daemon: Daemon):
     daemon.jobs = loaded.jobs
     daemon.pips = loaded.pips
     daemon.devs = loaded.devs
+    daemon.nextjob = loaded.nextjob

src/tomato/daemon/job.py

@@ -2,6 +2,7 @@
 import subprocess
 import logging
 import json
+import time
 from datetime import datetime, timezone
 from pathlib import Path
 from threading import currentThread
@@ -176,7 +177,7 @@ def manager(port: int, context: zmq.Context):
     poller.register(req, zmq.POLLIN)
     timeout = 1000
     while getattr(thread, "do_run"):
-        req.send_pyobj(dict(cmd="status", with_data=True))
+        req.send_pyobj(dict(cmd="status", with_data=True, sender=f"{__name__}.manager"))
         events = dict(poller.poll(timeout))
         if req not in events:
             logger.warning(f"could not contact daemon in {timeout} ms")
@@ -186,4 +187,5 @@ def manager(port: int, context: zmq.Context):
         manage_running_pips(daemon, req)
         matched_pips = check_queued_jobs(daemon, req)
         action_queued_jobs(daemon, matched_pips, req)
+        time.sleep(0.5)
     logger.info("instructed to quit")

src/tomato/drivers/driver_funcs.py

@@ -233,8 +233,6 @@ def driver_worker(
     log.debug(f"started 'log_listener' on pid {listener.pid}")
 
     jobs = []
-    print(f"{devices=}")
-    print(f"{pipeline['devs']=}")
     for cname, comp in pipeline["devs"].items():
         dev = devices[cname]
         log.info(f"device id: {cname}")

src/tomato/tomato/__init__.py

@@ -64,7 +64,6 @@ def load_device_file(yamlpath: Path) -> dict:
 def get_pipelines(devs: dict[str, Device], pipelines: list) -> dict[str, Pipeline]:
     pips = {}
     for pip in pipelines:
-        print(f"{pip=}")
         if "*" in pip["name"]:
             data = {"name": pip["name"], "devs": {}}
             if len(pip["devices"]) > 1:
@@ -106,7 +105,7 @@ def status(
     logger.debug(f"checking status of tomato on port {port}")
     req = context.socket(zmq.REQ)
     req.connect(f"tcp://127.0.0.1:{port}")
-    req.send_pyobj(dict(cmd="status", with_data=with_data))
+    req.send_pyobj(dict(cmd="status", with_data=with_data, sender=f"{__name__}.status"))
     poller = zmq.Poller()
     poller.register(req, zmq.POLLIN)
     events = dict(poller.poll(timeout))

tests/test_03_state.py

@@ -0,0 +1,61 @@
+import os
+from pathlib import Path
+import zmq
+import time
+
+from tomato import ketchup, tomato
+
+PORT = 12345
+CTXT = zmq.Context()
+
+kwargs = dict(port=PORT, timeout=1000, context=CTXT)
+
+
+def test_recover_queued_jobs(datadir, start_tomato_daemon, stop_tomato_daemon):
+    os.chdir(datadir)
+    ketchup.submit(payload="dummy_random_5_2.yml", jobname="job-1", **kwargs)
+    ketchup.submit(payload="dummy_random_5_2.yml", jobname="job-2", **kwargs)
+    tomato.stop(**kwargs)
+    assert os.path.exists("tomato_state_12345.pkl")
+    tomato.start(**kwargs, appdir=Path(), logdir=Path(), verbosity=0)
+    ret = tomato.status(**kwargs, with_data=True)
+    print(f"{ret=}")
+    assert ret.success
+    assert len(ret.data.jobs) == 2
+    assert ret.data.nextjob == 3
+
+
+def test_recover_running_jobs(datadir, start_tomato_daemon, stop_tomato_daemon):
+    os.chdir(datadir)
+    ketchup.submit(payload="dummy_random_5_2.yml", jobname="job-1", **kwargs)
+    tomato.pipeline_load(**kwargs, pipeline="dummy-5", sampleid="dummy_random_5_2")
+    tomato.pipeline_ready(**kwargs, pipeline="dummy-5")
+    time.sleep(2)
+    tomato.stop(**kwargs)
+    assert os.path.exists("tomato_state_12345.pkl")
+    tomato.start(**kwargs, appdir=Path(), logdir=Path(), verbosity=0)
+    ret = tomato.status(**kwargs, with_data=True)
+    print(f"{ret=}")
+    assert ret.success
+    assert len(ret.data.jobs) == 1
+    assert ret.data.nextjob == 2
+    assert ret.data.jobs[1].status == "r"
+
+
+def test_recover_waiting_jobs(datadir, start_tomato_daemon, stop_tomato_daemon):
+    os.chdir(datadir)
+    ketchup.submit(payload="dummy_random_5_2.yml", jobname="job-1", **kwargs)
+    tomato.pipeline_load(**kwargs, pipeline="dummy-5", sampleid="dummy_random_5_2")
+    tomato.pipeline_ready(**kwargs, pipeline="dummy-5")
+    time.sleep(2)
+    tomato.stop(**kwargs)
+    assert os.path.exists("tomato_state_12345.pkl")
+    time.sleep(5)
+    tomato.start(**kwargs, appdir=Path(), logdir=Path(), verbosity=0)
+    time.sleep(2)
+    ret = tomato.status(**kwargs, with_data=True)
+    print(f"{ret=}")
+    assert ret.success
+    assert len(ret.data.jobs) == 1
+    assert ret.data.nextjob == 2
+    assert ret.data.jobs[1].status == "c"