fix: full tets

Hocnonsense · Hocnonsense · commit 14a1f3ca0955 · 2025-08-26T02:23:08.000+08:00
diff --git a/src/snakemake/dag.py b/src/snakemake/dag.py
@@ -1448,14 +1448,14 @@ async def updated_input():
                             # change after evaluating the input function of the job in the second pass.
 
                             if RerunTrigger.CODE in self.workflow.rerun_triggers:
-                                # job metadata can be missed, but the separate scripts don't.
-                                # but it should not work if unwanted
-                                reason.code_changed = any(
-                                    [
-                                        f
-                                        async for f in job.outputs_older_than_script_or_notebook()
-                                    ]
-                                )
+                                # Prefer script/notebook mtime over metadata when CODE trigger is enabled.
+                                # Short-circuit on first older output to avoid building a list.
+                                reason.code_changed = False
+                                async for (
+                                    _
+                                ) in job.outputs_older_than_script_or_notebook():
+                                    reason.code_changed = True
+                                    break
                             if not self.workflow.persistence.has_metadata(job):
                                 reason.no_metadata = True
                             elif self.workflow.persistence.has_outdated_metadata(job):
diff --git a/tests/common.py b/tests/common.py
@@ -50,7 +50,7 @@ def is_connected():
     try:
         urllib.request.urlopen("http://www.google.com", timeout=1)
         return True
-    except urllib.request.URLError:
+    except (urllib.request.URLError, TimeoutError):
         return False
 
 
diff --git a/tests/tests.py b/tests/tests.py
@@ -480,18 +480,16 @@ def test_empty_include():
 
 def test_script_python():
     tmpdir = run(dpath("test_script_py"), cleanup=False)
-    outfile_path = os.path.join(tmpdir, "test.out")
+    outfile_path = os.path.join(tmpdir, "explicit_import.py.out")
     outfile_timestamp_orig = os.path.getmtime(outfile_path)
-    # update timestamp but not contents of input file
-    script_file = os.path.join(tmpdir, "scripts/test.py")
+    # update timestamp but not contents of script
+    script_file = os.path.join(tmpdir, "scripts/test_explicit_import.py")
     os.utime(script_file)
-    run(
-        dpath("test_script_py"),
-        rerun_triggers=frozenset([RerunTrigger.MTIME]),
-        cleanup=False,
-    )
-    outfile_timestamp_new = os.path.getmtime(outfile_path)
-    assert outfile_timestamp_orig == outfile_timestamp_new
+    shell(f"snakemake -s {tmpdir}/Snakefile -d {tmpdir} --rerun-triggers mtime -c 1")
+    assert outfile_timestamp_orig == os.path.getmtime(outfile_path)
+    shell(f"snakemake -s {tmpdir}/Snakefile -d {tmpdir} -c 1")
+    assert outfile_timestamp_orig != os.path.getmtime(outfile_path)
+    shutil.rmtree(tmpdir, ignore_errors=ON_WINDOWS)
 
 
 @skip_on_windows  # Test relies on perl
