How to test a fastapi endpoint that uses signal for timeout check

[tauferp]

Hello,

I have a situation where I'm using FastApi with starlette and have few endpoints with a following timeout check (these endpoints don't use async internally):

def timeout_check(
    fn: Callable[P, Coroutine[Any, Any, R | JSONResponse]]
) -> Callable[P, Coroutine[Any, Any, R | JSONResponse]]:
    @wraps(fn)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R | JSONResponse:
        request: Request = kwargs.get("request")  # type: ignore
        timeout: int = int(kwargs.get("timeout") or 10) # type: ignore
        print(f'{timeout=} {threading.get_native_id()=}')
        signal.alarm(timeout)
        try:
            return await fn(*args, **kwargs)
        except TimeoutException as exc:
            return JSONResponse(
                status_code=413,
                content={"message": f"Timed out: {exc}"},
            )
        finally:
            signal.alarm(0)

and set up using

import signal
import threading

class TimeoutException(Exception):
    pass

def timeout_handler(signum: int, frame: Any) -> None:
    print(f"{threading.get_native_id()=}")
    raise TimeoutException(f"end of time {frame}")

signal.signal(signal.SIGALRM, timeout_handler)

Example endpoint:

class InternalResponse(BaseModel):
    message: str

class SleepActiveRequest(BaseModel):
    duration: float

@router.post('/sleep_active',
    tags=["debug"],
    response_model=InternalResponse,
)
@timeout_check
async def sleep_active_handler(request: Request, body: SleepActiveRequest, timeout: TimeoutArgType=None) -> InternalResponse:
    print(f"sleeping for {body}")
    start = time.time()
    while time.time() < start + body.duration:
        pass
    return InternalResponse(message=f"Actively slept {body} seconds.")

When triggering the timeout check using uvicorn, everything looks OK, but in a test the signal gets sent to a different thread id.

def test_timeout(client: TestClient) -> None:
    response = client.post(
        "/debug/sleep_active?timeout=1", json={"duration":2}
    )
    assert response.status_code == 413
    assert "message" in response.json()

with output for example:

timeout=1.0 threading.get_native_id()=443875
sleeping for duration=2.0
threading.get_native_id()=443814

and traceback:

venv/lib/python3.11/site-packages/starlette/testclient.py:591: in post
    return super().post(
venv/lib/python3.11/site-packages/httpx/_client.py:1132: in post
    return self.request(
venv/lib/python3.11/site-packages/starlette/testclient.py:466: in request
    return super().request(
venv/lib/python3.11/site-packages/httpx/_client.py:814: in request
    return self.send(request, auth=auth, follow_redirects=follow_redirects)
venv/lib/python3.11/site-packages/httpx/_client.py:901: in send
    response = self._send_handling_auth(
venv/lib/python3.11/site-packages/httpx/_client.py:929: in _send_handling_auth
    response = self._send_handling_redirects(
venv/lib/python3.11/site-packages/httpx/_client.py:966: in _send_handling_redirects
    response = self._send_single_request(request)
venv/lib/python3.11/site-packages/httpx/_client.py:1002: in _send_single_request
    response = transport.handle_request(request)
venv/lib/python3.11/site-packages/starlette/testclient.py:343: in handle_request
    raise exc
venv/lib/python3.11/site-packages/starlette/testclient.py:340: in handle_request
    portal.call(self.app, scope, receive, send)
venv/lib/python3.11/site-packages/anyio/from_thread.py:277: in call
    return cast(T_Retval, self.start_task_soon(func, *args).result())
/usr/lib/python3.11/concurrent/futures/_base.py:451: in result
    self._condition.wait(timeout)
/usr/lib/python3.11/threading.py:327: in wait
    waiter.acquire()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

signum = 14
frame = <frame at 0x7f1b00e23680, file '/usr/lib/python3.11/threading.py', line 339, code wait>

    def timeout_handler(signum: int, frame: Any) -> None:
        print(f"{threading.get_native_id()=}")
>       raise TimeoutException(f"end of time {frame}")
E       TimeoutException: end of time <frame at 0x7f1b00e23680, file '/usr/lib/python3.11/threading.py', line 327, code wait>

app.py:33: TimeoutException

Based on my search it's because signals are sent to the main thread and testclient is riunning the server in a different thread. Sadly I'm not aware of any other way to ensure that a task is terminated after given amount of time (it's a CPU bound task).

Is there anything that can be done so that I can include this test? (Other than move it to integration level and test against uvicorn)