Do not turn an interruption into a failed Python activity

CHANGELOG.md

@@ -5,6 +5,13 @@
 
 [Unreleased]: https://github.com/chaostoolkit/chaostoolkit-lib/compare/1.45.0...HEAD
 
+### Fixed
+
+* An interruption raised while a Python activity runs, for instance on
+  SIGTERM, is no longer turned into a failure of that activity. The run is
+  now interrupted and reported as such, instead of carrying on and possibly
+  completing
+
 ## [1.45.0][] - 2026-08-08
 
 [1.45.0]: https://github.com/chaostoolkit/chaostoolkit-lib/compare/1.44.0...1.45.0

chaoslib/provider/python.py

@@ -6,7 +6,11 @@
 from typing import Any
 
 from chaoslib import substitute
-from chaoslib.exceptions import ActivityFailed, InvalidActivity
+from chaoslib.exceptions import (
+    ActivityFailed,
+    InterruptExecution,
+    InvalidActivity,
+)
 from chaoslib.types import Activity, Configuration, Secrets
 
 __all__ = ["run_python_activity", "validate_python_activity"]
@@ -54,6 +58,11 @@ def run_python_activity(
 
     try:
         return func(**arguments)
+    except InterruptExecution:
+        # the whole execution is being interrupted, by SIGTERM or a control
+        # for instance, while this activity ran: that is not a failure of
+        # the activity and must reach the runner as is
+        raise
     except Exception as x:  # noqa: BLE001 - wrap any user-code exception
         raise ActivityFailed(
             traceback.format_exception_only(type(x), x)[0].strip()

tests/fixtures/interrupter.py

@@ -1,5 +1,6 @@
 import time
 
+from chaoslib.exceptions import InterruptExecution
 from chaoslib.exit import exit_gracefully, exit_ungracefully
 
 
@@ -11,3 +12,7 @@ def interrupt_gracefully_in(seconds: int = 1.5):
 def interrupt_ungracefully_in(seconds: int = 1.5):
     time.sleep(seconds)
     exit_ungracefully()
+
+
+def raise_interruption() -> None:
+    raise InterruptExecution("interrupted from an activity")

tests/test_exit.py

@@ -121,7 +121,9 @@ def test_wait_for_background_activity_on_graceful_exit():
             journal = runner.run(x)
 
             assert journal["status"] == "interrupted"
-            assert 3.0 < journal["run"][0]["duration"] < 3.2
+            # the background activity ran its full 3s: the upper bound only
+            # tells it apart from the 5s HTTP call, with room for slow runners
+            assert 3.0 < journal["run"][0]["duration"] < 4.0
     finally:
         stop_http_server(httpd, server)
 
@@ -155,3 +157,69 @@ def _exit_soon():
         journal = runner.run(x)
         assert journal["status"] == "interrupted"
         assert journal["run"][0]["status"] == "succeeded"
+
+
+def _interrupted_python_experiment(module: str, func: str, arguments=None):
+    return {
+        "title": "interrupted while a python activity runs",
+        "description": "n/a",
+        "method": [
+            {
+                "type": "action",
+                "name": "long-python-call",
+                "provider": {
+                    "type": "python",
+                    "module": module,
+                    "func": func,
+                    "arguments": arguments or {},
+                },
+            },
+            {
+                "type": "action",
+                "name": "must-not-run",
+                "provider": {"type": "process", "path": "echo"},
+            },
+        ],
+        "rollbacks": [
+            {
+                "type": "action",
+                "name": "rollback",
+                "provider": {"type": "process", "path": "echo"},
+            }
+        ],
+    }
+
+
+def test_sigterm_during_python_activity_interrupts_the_run():
+    import os
+    import signal
+
+    x = _interrupted_python_experiment(
+        "fixtures.longpythonfunc", "pause", {"howlong": 5}
+    )
+    timer = threading.Timer(0.5, os.kill, (os.getpid(), signal.SIGTERM))
+    with Runner(Strategy.DEFAULT) as runner:
+        timer.start()
+        journal = runner.run(
+            x, settings={"runtime": {"rollbacks": {"strategy": "always"}}}
+        )
+
+    assert journal["status"] == "interrupted"
+    assert [r["activity"]["name"] for r in journal["run"]] == [
+        "long-python-call"
+    ]
+    assert journal["run"][0].get("status") != "failed"
+    assert len(journal["rollbacks"]) == 1
+
+
+def test_interruption_raised_by_python_activity_interrupts_the_run():
+    x = _interrupted_python_experiment(
+        "fixtures.interrupter", "raise_interruption"
+    )
+    with Runner(Strategy.DEFAULT) as runner:
+        journal = runner.run(
+            x, settings={"runtime": {"rollbacks": {"strategy": "always"}}}
+        )
+
+    assert journal["status"] == "interrupted"
+    assert len(journal["run"]) == 1