From 088785bb9451a641b89be682e66b63c9b1145b99 Mon Sep 17 00:00:00 2001 From: Marcus Messer Date: Tue, 26 May 2026 17:22:23 +0100 Subject: [PATCH 1/4] Added healthcheck handler support to RPC framework --- lf_toolkit/io/handler.py | 4 ++++ lf_toolkit/io/rpc_handler.py | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/lf_toolkit/io/handler.py b/lf_toolkit/io/handler.py index d85bcfb..059ed04 100644 --- a/lf_toolkit/io/handler.py +++ b/lf_toolkit/io/handler.py @@ -57,6 +57,10 @@ async def handle_preview(self, req: dict): return await self._call_user_handler("preview", response, request_params) + async def handle_healthcheck(self, req: dict): + from .healthcheck import run_healthcheck + return await anyio.to_thread.run_sync(run_healthcheck) + async def handle(self, name: Command, req: dict) -> dict: handler = getattr(self, f"handle_{name}", None) diff --git a/lf_toolkit/io/rpc_handler.py b/lf_toolkit/io/rpc_handler.py index 354ae95..1eaff8a 100644 --- a/lf_toolkit/io/rpc_handler.py +++ b/lf_toolkit/io/rpc_handler.py @@ -12,7 +12,7 @@ class JsonRpcHandler(Handler): def __init__(self): self._methods = { - name: jsonrpc_handler(self, name) for name in ["eval", "preview"] + name: jsonrpc_handler(self, name) for name in ["eval", "preview", "healthcheck"] } async def dispatch(self, req: str) -> str: From 16629c0ece626cee9bf93e1fd923ce52519249ee Mon Sep 17 00:00:00 2001 From: Marcus Messer Date: Tue, 26 May 2026 17:44:38 +0100 Subject: [PATCH 2/4] Updated healthcheck handler to return detailed status --- lf_toolkit/io/handler.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lf_toolkit/io/handler.py b/lf_toolkit/io/handler.py index 059ed04..2e8031f 100644 --- a/lf_toolkit/io/handler.py +++ b/lf_toolkit/io/handler.py @@ -59,7 +59,8 @@ async def handle_preview(self, req: dict): async def handle_healthcheck(self, req: dict): from .healthcheck import run_healthcheck - return await anyio.to_thread.run_sync(run_healthcheck) + result = await anyio.to_thread.run_sync(run_healthcheck) + return {"status": "OK" if result["tests_passed"] else "DEGRADED"} async def handle(self, name: Command, req: dict) -> dict: handler = getattr(self, f"handle_{name}", None) From a76f1fc36e044bb9fe93e710ce8303ece4bde160 Mon Sep 17 00:00:00 2001 From: Marcus Messer Date: Tue, 26 May 2026 17:44:50 +0100 Subject: [PATCH 3/4] Added standalone healthcheck runner with JSON output support --- lf_toolkit/io/healthcheck.py | 96 ++++++++++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 lf_toolkit/io/healthcheck.py diff --git a/lf_toolkit/io/healthcheck.py b/lf_toolkit/io/healthcheck.py new file mode 100644 index 0000000..4e2aa84 --- /dev/null +++ b/lf_toolkit/io/healthcheck.py @@ -0,0 +1,96 @@ +import os +import re +import sys +import time +import unittest +from typing import Any, List, TypedDict + +from typing_extensions import NotRequired + + +class JsonTestResult(TypedDict): + name: str + time: NotRequired[int] + + +JsonTestResults = List[JsonTestResult] + + +class HealthcheckJsonTestResult(TypedDict): + tests_passed: bool + successes: JsonTestResults + failures: JsonTestResults + errors: JsonTestResults + + +class HealthcheckResult(unittest.TextTestResult): + + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.__path_re = re.compile(r"^[\.\/\w]+\.(\w+\.\w+)$") + self.__successes_json: JsonTestResults = [] + self.__failures_json: JsonTestResults = [] + self.__errors_json: JsonTestResults = [] + + def _get_name(self, path: str) -> str: + match = self.__path_re.match(path) + return match.group(1) if match else "Unknown" + + def startTest(self, test: unittest.TestCase) -> None: + self._start_time = time.time() + super().startTest(test) + + def addSuccess(self, test: unittest.TestCase) -> None: + elapsed = time.time() - self._start_time + self.__successes_json.append( + JsonTestResult(name=self._get_name(test.id()), time=round(1e6 * elapsed)) + ) + super().addSuccess(test) + + def addFailure(self, test: unittest.TestCase, err: Any) -> None: + self.__failures_json.append(JsonTestResult(name=self._get_name(test.id()))) + super().addFailure(test, err) + + def addError(self, test: unittest.TestCase, err: Any) -> None: + self.__errors_json.append(JsonTestResult(name=self._get_name(test.id()))) + super().addError(test, err) + + def get_successes_json(self) -> JsonTestResults: + return self.__successes_json + + def get_failures_json(self) -> JsonTestResults: + return self.__failures_json + + def get_errors_json(self) -> JsonTestResults: + return self.__errors_json + + +class HealthcheckRunner(unittest.TextTestRunner): + + def __init__(self, *args, **kwargs) -> None: + super().__init__(resultclass=HealthcheckResult, *args, **kwargs) + + def run(self, test) -> HealthcheckJsonTestResult: + result: HealthcheckResult = super().run(test) # type: ignore + return HealthcheckJsonTestResult( + tests_passed=result.wasSuccessful(), + successes=result.get_successes_json(), + failures=result.get_failures_json(), + errors=result.get_errors_json(), + ) + + +def run_healthcheck() -> HealthcheckJsonTestResult: + no_stream = open(os.devnull, "w") + sys.stderr = no_stream + + try: + loader = unittest.TestLoader() + suite = loader.discover(start_dir=".", pattern="*test*.py") + runner = HealthcheckRunner(verbosity=0) + result = runner.run(suite) + finally: + sys.stderr = sys.__stderr__ + no_stream.close() + + return result \ No newline at end of file From 4db4918cf2c151673e348c2e001fc3450613d3f0 Mon Sep 17 00:00:00 2001 From: Marcus Messer Date: Tue, 26 May 2026 18:01:04 +0100 Subject: [PATCH 4/4] Simplified healthcheck handler by returning the result directly --- lf_toolkit/io/handler.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/lf_toolkit/io/handler.py b/lf_toolkit/io/handler.py index 2e8031f..059ed04 100644 --- a/lf_toolkit/io/handler.py +++ b/lf_toolkit/io/handler.py @@ -59,8 +59,7 @@ async def handle_preview(self, req: dict): async def handle_healthcheck(self, req: dict): from .healthcheck import run_healthcheck - result = await anyio.to_thread.run_sync(run_healthcheck) - return {"status": "OK" if result["tests_passed"] else "DEGRADED"} + return await anyio.to_thread.run_sync(run_healthcheck) async def handle(self, name: Command, req: dict) -> dict: handler = getattr(self, f"handle_{name}", None)