improved test quality following coderabbit suggestions

Lucio Anderlini · Lucio Anderlini · commit 099e8d5e53c5 · 2025-06-19T13:54:38.000Z
diff --git a/src/snakemake/cli.py b/src/snakemake/cli.py
@@ -1587,9 +1587,9 @@ def get_argument_parser(profiles=None):
         "generated job id to the first stdout line, {dependencies} will "
         "be filled with space separated job ids this job depends on. "
         "Does not work for workflows that contain checkpoint rules, "
-        "and localrules will fail. The additional argument `--notemp` "
+        "and localrules will be skipped. The additional argument `--notemp` "
         "should be specified. Most often, `--not-retrieve-storage` is "
-        "also required to avoid Snakemake trying to download output files "
+        "also recommended to avoid Snakemake trying to download output files "
         "before the jobs producing them are executed. ",
     )
     group_cluster.add_argument(
diff --git a/tests/test_immediate_submit_without_shared_fs/Snakefile b/tests/test_immediate_submit_without_shared_fs/Snakefile
@@ -7,15 +7,16 @@ rule produce:
         txt="{fname}.txt"
 
     
-    shell: "echo {wildcards.fname} > {output.txt}"
+    shell: "printf '%s\n' {wildcards.fname} > {output.txt}"
+
 
 rule count:
     input:
         txt="{fname}.txt"
     output:
         dat="{fname}.dat"
 
-    shell: "cat {input.txt} | wc -m > {output.dat}"
+    shell: "wc -m < {input.txt} > {output.dat}"
 
 rule sum:
     input:
@@ -28,7 +29,7 @@ rule sum:
         """
         SUM=0
         for i in {input.counts}; do
-            SUM=$(($SUM + $(cat $i)));
+            SUM=$(($SUM + $(cat "$i")));
         done;
         echo "sum: $SUM" > {output.yaml}
         """
diff --git a/tests/test_immediate_submit_without_shared_fs/clustersubmit b/tests/test_immediate_submit_without_shared_fs/clustersubmit
@@ -1,17 +1,20 @@
 #!/usr/bin/env python3
+import sys
 import asyncio
 import json
-import sys
 from argparse import ArgumentParser 
 import logging 
 import sqlite3
 import string
 import random
 import os
+import time
+import tempfile
 
 ## Configuration 
-DATABASE_FILE = "/tmp/clustersubmit.sqlite3"
-LOG_FILE = "/tmp/clustersubmit.log"
+DATABASE_FILE = os.environ.get("DATABASE_FILE", os.path.join(tempfile.gettempdir(), "clustersubmit.sqlite3"))
+LOG_FILE = os.environ.get("LOG_FILE", os.path.join(tempfile.gettempdir(), "clustersubmit.log"))
+TIMEOUT_SECONDS = float(os.environ.get("TIMEOUT_SECONDS", "300"))
 
 logging.basicConfig(
     level=logging.DEBUG,
@@ -41,10 +44,27 @@ args = parser.parse_args()
 
 ## Initialization
 with sqlite3.connect(DATABASE_FILE) as db:
-    db.execute("CREATE TABLE IF NOT EXISTS jobs(jobid TEXT, jobscript TEXT, done INT, testtime)") 
-    db.execute("CREATE TABLE IF NOT EXISTS edges(parent TEXT, child TEXT, testtime)") 
-    db.execute("DELETE FROM jobs WHERE testtime <= datetime('now', '-1 hour')")
-    db.execute("DELETE FROM edges WHERE testtime <= datetime('now', '-1 hour')")
+    db.execute("""
+        CREATE TABLE IF NOT EXISTS jobs(
+            jobid TEXT PRIMARY KEY, 
+            jobscript TEXT, 
+            done INT, 
+            created_at TEXT
+        )
+    """) 
+
+    db.execute("""
+        CREATE TABLE IF NOT EXISTS edges(
+            parent TEXT, 
+            child TEXT, 
+            created_at TEXT,
+            FOREIGN KEY(parent) REFERENCES jobs(jobid),
+            FOREIGN KEY(child) REFERENCES jobs(jobid)
+        );
+    """) 
+
+    db.execute("DELETE FROM jobs WHERE created_at <= datetime('now', '-1 hour')")
+    db.execute("DELETE FROM edges WHERE created_at <= datetime('now', '-1 hour')")
 
 
 
@@ -57,69 +77,79 @@ async def execute_single_job(jobid, jobscript):
     """
     logging.info(f"Executing {jobid}")
     proc = await asyncio.create_subprocess_shell(jobscript)
-    await proc.communicate()
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        logging.critical(stdout)
+        logging.critical(stderr)
+        raise RuntimeError(f"Falied executing job {jobid}")
+
     async with db_lock:
         with sqlite3.connect(DATABASE_FILE, timeout=30) as db:
             db.execute("PRAGMA journal_mode=WAL")
             db.execute("UPDATE jobs SET done = 1 WHERE jobid = ?", (jobid,))
     logging.info(f"Marked {jobid} as done")
-async def execute_jobs():
+
+
+async def execute_jobs(timeout_seconds=TIMEOUT_SECONDS):
     """
     Simplistic scheduler checking jobs that can run and executing them
     """
-    for _ in range(100):
+    start = time.perf_counter()
+    while (time.perf_counter() - start) < timeout_seconds:
         with sqlite3.connect(DATABASE_FILE) as db:
             result = db.execute(QUERY_RUNNABLE).fetchall()
-        if len(result):
-            async with asyncio.TaskGroup() as tg:
-                for row in result:
-                    tg.create_task(execute_single_job(*row))
-        else:
+        if not result:
             logging.info("Could not find any other job to execute.")
             return 
+
+        async with asyncio.TaskGroup() as tg:
+            for row in result:
+                tg.create_task(execute_single_job(*row))
+
+    raise TimeoutError(f"Exceeded timeout of {timeout_seconds} seconds")
                 
 
 if args.execute:
     asyncio.run(execute_jobs())
     exit(0)
 elif args.jobscript is None:
+    logging.critical(f"Failed to submit a job: missing jobscript. {sys.argv}")
     raise ValueError("Expected jobscript as the last positional argument")
 
     
 
-
 ## Submission 
 with open(args.jobscript) as input_file:
-    for line in input_file:
-        if line.startswith("# properties = "):
-            properties = json.loads(line[len("# properties = "):-1])
-            break
+    jobscript_content = input_file.read()
+
+for line in jobscript_content.splitlines():
+    if line.startswith("# properties = "):
+        properties = json.loads(line[len("# properties = "):])
+        break
 
 logging.info(f"---")
 logging.info(f"Processing rule: {properties['rule']}")
 
 job_name = ''.join(
-    [properties['rule'], '-'] + 
+    [properties['rule'], '-', str(time.time()), '-'] + 
     [random.choice(string.ascii_lowercase) for _ in range(5)]
 )
 
-dependencies = [s for s in args.dependencies.split(" ") if s not in [' ', '', None]]
+dependencies = args.dependencies.split()
 logging.info(f"Submitting `{job_name}` which depends on {dependencies}")
 
 with sqlite3.connect(DATABASE_FILE) as db:
     db.execute(
-        "INSERT INTO jobs(jobid, jobscript, done, testtime) VALUES (?, ?, 0, datetime('now'))",
-        (job_name, open(args.jobscript).read())
+        "INSERT INTO jobs(jobid, jobscript, done, created_at) VALUES (?, ?, 0, datetime('now'))",
+        (job_name, jobscript_content)
     )
 
     for dependency in dependencies:
         db.execute(
-            "INSERT INTO edges(parent, child, testtime) VALUES (?, ?, datetime('now'))",
+            "INSERT INTO edges(parent, child, created_at) VALUES (?, ?, datetime('now'))",
             (dependency, job_name)
         )
 
-        
-
 print(job_name)
 
 
diff --git a/tests/tests.py b/tests/tests.py
@@ -2478,14 +2478,14 @@ def test_params_empty_inherit():
 def test_immediate_submit_without_shared_fs():
     run(
         dpath("test_immediate_submit_without_shared_fs"), 
-        shellcmd="snakemake "
-        "--executor cluster-generic "
-        "--cluster-generic-submit-cmd \"./clustersubmit --dependencies \\\"{dependencies}\\\"\" "
-        "--forceall " 
-        "--immediate-submit "
-        "--notemp "
-        "--jobs 10 " 
-        "; ./clustersubmit --execute" 
+        shellcmd="""snakemake \
+        --executor cluster-generic \
+        --cluster-generic-submit-cmd "./clustersubmit --dependencies \\\"{dependencies}\\\"" \
+        --forceall \
+        --immediate-submit \
+        --notemp \
+        --jobs 10 \
+        ; ./clustersubmit --execute """
     )
 
 
