Bring Service classes more in sync with Java impl

* Add missing Service class methods:
  - kill, wait_for, is_alive, invalid_lines, error_lines

* Make crash reporting report stdout and stderr lines:
  - As per apposed/appose-java@177bb42.

* Migrate corresponding unit tests.

* Add missing type hints.

Author: ctrueden

src/appose/service.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import os
 import subprocess
 import tempfile
 import threading
@@ -54,6 +55,8 @@ def __init__(
         self._tasks: dict[str, "Task"] = {}
         self._service_id: int = Service._service_count
         Service._service_count += 1
+        self._invalid_lines: list[str] = []
+        self._error_lines: list[str] = []
         self._process: subprocess.Popen | None = None
         self._stdout_thread: threading.Thread | None = None
         self._stderr_thread: threading.Thread | None = None
@@ -298,7 +301,60 @@ def close(self) -> None:
             raise RuntimeError("Service has not been started")
         self._process.stdin.close()
 
+    def kill(self) -> None:
+        """
+        Force the service's worker process to begin shutting down. Any tasks still
+        pending completion will be interrupted, reporting TaskStatus.CRASHED.
+
+        To shut down the service more gently, allowing any pending tasks to run to
+        completion, use close() instead.
+
+        To wait until the service's worker process has completely shut down
+        and all output has been reported, call wait_for() afterward.
+        """
+        self._process.kill()
+
+    def wait_for(self) -> int:
+        """
+        Wait for the service's worker process to terminate.
+
+        Returns:
+            Exit value of the worker process.
+        """
+        self._process.wait()
+
+        # Wait for worker output processing threads to finish up.
+        self._stdout_thread.join()
+        self._stderr_thread.join()
+        self._monitor_thread.join()
 
+        return self._process.returncode
+
+    def is_alive(self) -> bool:
+        """
+        Return true if the service's worker process is currently running,
+        or false if it has not yet started or has already shut down or crashed.
+
+        Returns:
+            Whether the service's worker process is currently running.
+        """
+        return self._process is not None and self._process.poll() is None
+
+    def invalid_lines(self) -> list[str]:
+        """
+        Unparseable lines emitted by the worker process on its stdout stream,
+        collected over the lifetime of the service.
+        Can be useful for analyzing why a worker process has crashed.
+        """
+        return self._invalid_lines
+
+    def error_lines(self) -> list[str]:
+        """
+        Lines emitted by the worker process on its stderr stream,
+        collected over the lifetime of the service.
+        Can be useful for analyzing why a worker process has crashed.
+        """
+        return self._error_lines
 
     def _stdout_loop(self) -> None:
         """
@@ -310,7 +366,7 @@ def _stdout_loop(self) -> None:
             try:
                 line = None if stdout is None else stdout.readline()
             except Exception:
-                # Something went wrong reading the line. Panic!
+                # Something went wrong reading the stdout line. Panic!
                 self._debug_service(format_exc())
                 break
 
@@ -336,22 +392,26 @@ def _stdout_loop(self) -> None:
                 # Something went wrong decoding the line of JSON.
                 # Skip it and keep going, but log it first.
                 self._debug_service(f"<INVALID> {line}")
+                self._invalid_lines.append(line.rstrip("\n\r"))
 
     def _stderr_loop(self) -> None:
         """
         Input loop processing lines from the worker's stderr stream.
         """
-        # noinspection PyBroadException
-        try:
-            while True:
-                stderr = self._process.stderr
+        while True:
+            stderr = self._process.stderr
+            # noinspection PyBroadException
+            try:
                 line = None if stderr is None else stderr.readline()
-                if not line:  # readline returns empty string upon EOF
-                    self._debug_service("<worker stderr closed>")
-                    return
-                self._debug_worker(line)
-        except Exception:
-            self._debug_service(format_exc())
+            except Exception:
+                # Something went wrong reading the stderr line. Panic!
+                self._debug_service(format_exc())
+                break
+            if not line:  # readline returns empty string upon EOF
+                self._debug_service("<worker stderr closed>")
+                break
+            self._debug_worker(line)
+            self._error_lines.append(line.rstrip("\n\r"))
 
     def _monitor_loop(self) -> None:
         # Wait until the worker process terminates.
@@ -364,15 +424,33 @@ def _monitor_loop(self) -> None:
                 f"<worker process terminated with exit code {exit_code}>"
             )
         task_count = len(self._tasks)
-        if task_count > 0:
-            self._debug_service(
-                f"<worker process terminated with {task_count} pending tasks>"
-            )
+        if task_count == 0:
+            # No hanging tasks to clean up.
+            return
+
+        self._debug_service(
+            "<worker process terminated with "
+            + f"{task_count} pending task{'' if task_count == 1 else 's'}>"
+        )
 
         # Notify any remaining tasks about the process crash.
+        nl = os.linesep
+        error_parts = [f"Worker crashed with exit code {exit_code}."]
+        error_parts.append("")
+        error_parts.append("[stdout]")
+        if len(self._invalid_lines) == 0:
+            error_parts.append("<none>")
+        else:
+            error_parts.extend(self._invalid_lines)
+        error_parts.append("")
+        error_parts.append("[stderr]")
+        if len(self._error_lines) == 0:
+            error_parts.append("<none>")
+        else:
+            error_parts.extend(self._error_lines)
+        error = nl.join(error_parts) + nl
         for task in self._tasks.values():
-            task._crash()
-
+            task._crash(error)
         self._tasks.clear()
 
     def _debug_service(self, message: str) -> None:
@@ -405,12 +483,17 @@ class TaskStatus(Enum):
     FAILED = "FAILED"
     CRASHED = "CRASHED"
 
-    def is_finished(self):
+    def is_finished(self) -> bool:
         """
         True iff status is COMPLETE, CANCELED, FAILED, or CRASHED.
         """
+        return self == TaskStatus.COMPLETE or self.is_error()
+
+    def is_error(self) -> bool:
+        """
+        True iff status is CANCELED, FAILED, or CRASHED.
+        """
         return self in (
-            TaskStatus.COMPLETE,
             TaskStatus.CANCELED,
             TaskStatus.FAILED,
             TaskStatus.CRASHED,
@@ -434,7 +517,7 @@ class ResponseType(Enum):
     True iff response type is COMPLETE, CANCELED, FAILED, or CRASHED.
     """
 
-    def is_terminal(self):
+    def is_terminal(self) -> bool:
         return self in (
             ResponseType.COMPLETION,
             ResponseType.CANCELATION,
@@ -629,9 +712,10 @@ def _handle(self, response: Args) -> None:
             with self.cv:
                 self.cv.notify_all()
 
-    def _crash(self):
+    def _crash(self, error: str):
         event = TaskEvent(self, ResponseType.CRASH)
         self.status = TaskStatus.CRASHED
+        self.error = error
         for listener in self.listeners:
             listener(event)
         with self.cv:

tests/test_service.py

@@ -4,9 +4,12 @@
 
 
 import appose
-from appose.service import TaskStatus
+from appose.service import ResponseType, TaskException, TaskStatus
 from tests.test_base import execute_and_assert, maybe_debug
 from pathlib import Path
+import time
+import os
+import re
 
 collatz_groovy = """
 // Computes the stopping time of a given value
@@ -190,3 +193,164 @@ def test_init_numpy():
             actual_val = result[i]
             assert isinstance(actual_val, (int, float))
             assert abs(actual_val - expected_val) < 1e-10
+
+
+def test_task_failure_python():
+    env = appose.system()
+    with env.python() as service:
+        maybe_debug(service)
+        script = "whee\n"
+        try:
+            service.task(script).wait_for()
+            raise AssertionError("Expected TaskException for failed script")
+        except TaskException as e:
+            expected_error = "NameError: name 'whee' is not defined"
+            assert expected_error in str(e)
+
+
+def test_startup_crash():
+    env = appose.system()
+    python_exes = ["python", "python3", "python.exe"]
+    service = env.service(python_exes, "-c", "import nonexistentpackage").start()
+    # Wait up to 500ms for the crash.
+    for i in range(100):
+        if not service.is_alive():
+            break
+        time.sleep(0.005)
+    assert not service.is_alive()
+    # Check that the crash happened and was recorded correctly.
+    error_lines = service.error_lines()
+    assert error_lines is not None
+    assert len(error_lines) > 0
+    error = error_lines[-1]
+    assert error == "ModuleNotFoundError: No module named 'nonexistentpackage'"
+
+
+def test_python_sys_exit():
+    env = appose.system()
+    with env.python() as service:
+        maybe_debug(service)
+
+        # Launch a task that calls sys.exit. This is a nasty thing to do
+        # because Python does not exit the worker process when sys.exit is
+        # called within a dedicated threading.Thread; the thread just dies.
+        # So in addition to testing the Python code here, we are also testing
+        # that Appose's python_worker handles this situation well.
+        try:
+            service.task("import sys\nsys.exit(123)").wait_for()
+            raise AssertionError("Expected TaskException for sys.exit")
+        except TaskException as e:
+            # The failure should be either "thread death" or a "SystemExit" message.
+            assert "thread death" in str(e) or "SystemExit: 123" in str(e)
+
+
+def test_crash_with_active_task():
+    env = appose.system()
+    with env.python() as service:
+        maybe_debug(service)
+        # Create a "long-running" task.
+        script = (
+            "import sys\n"
+            "sys.stderr.write('one\\n')\n"
+            "sys.stderr.flush()\n"
+            "print('two')\n"
+            "sys.stdout.flush()\n"
+            "sys.stderr.write('three\\n')\n"
+            "sys.stderr.flush()\n"
+            "task.update('halfway')\n"
+            "print('four')\n"
+            "sys.stdout.flush()\n"
+            "sys.stderr.write('five\\n')\n"
+            "sys.stderr.flush()\n"
+            "print('six')\n"
+            "sys.stdout.flush()\n"
+            "sys.stderr.write('seven\\n')\n"
+            "task.update('crash-me')\n"
+            "import time; time.sleep(999)\n"
+        )
+        ready = [False]
+
+        def on_crash_me(event):
+            if event.message == "crash-me":
+                ready[0] = True
+
+        task = service.task(script)
+        task.listen(on_crash_me)
+
+        # Record any crash reported in the task notifications.
+        reported_error = [None]
+
+        def on_crash(event):
+            if event.response_type == ResponseType.CRASH:
+                reported_error[0] = task.error
+
+        task.listen(on_crash)
+
+        # Launch the task.
+        task.start()
+
+        # Simulate a crash after the script has emitted its output.
+        while not ready[0]:
+            time.sleep(0.005)
+        service.kill()
+
+        # Wait for the service to fully shut down after the crash.
+        exit_code = service.wait_for()
+        assert exit_code != 0
+
+        # Is the task flagged as crashed?
+        assert TaskStatus.CRASHED == task.status
+
+        # Was the crash error successfully and consistently recorded?
+        assert reported_error[0] is not None
+        nl = os.linesep
+        assert service.invalid_lines() == ["two", "four", "six"]
+        assert service.error_lines() == ["one", "three", "five", "seven"]
+        expected = (
+            f"Worker crashed with exit code ###.{nl}"
+            f"{nl}"
+            f"[stdout]{nl}"
+            f"two{nl}"
+            f"four{nl}"
+            f"six{nl}"
+            f"{nl}"
+            f"[stderr]{nl}"
+            f"one{nl}"
+            f"three{nl}"
+            f"five{nl}"
+            f"seven{nl}"
+        )
+        generalized_error = re.sub(r"exit code -?[0-9]+", "exit code ###", task.error)
+        assert expected == generalized_error
+
+
+def test_task_result():
+    """Tests Task.result() convenience method."""
+    env = appose.system()
+    with env.python() as service:
+        maybe_debug(service)
+
+        # Create a task that produces a result.
+        task = service.task("'success'").wait_for()
+        assert TaskStatus.COMPLETE == task.status
+
+        # Test the result() convenience method.
+        result = task.result()
+        assert result == "success"
+
+        # Verify it's the same as directly accessing outputs.
+        assert task.outputs.get("result") == result
+
+
+def test_task_result_null():
+    """Tests Task.result() returns None when no result is set."""
+    env = appose.system()
+    with env.python() as service:
+        maybe_debug(service)
+
+        # Create a task that doesn't set a result.
+        task = service.task("print('no result')").wait_for()
+        assert TaskStatus.COMPLETE == task.status
+
+        # result() should return None.
+        assert task.result() is None