// Configure transport to throw exception on release
        Mockito.doThrow(new RuntimeException("Test exception")).when(mockTransport).release();

        session.acquire();

        // Should handle exceptions gracefully
        assertDoesNotThrow(() -> session.release());
    }

    /**
     * Test AutoCloseable pattern
     */
    @Test
    @Timeout(5)

 * possibly run in another thread. That cause should itself be an {@code Error}; if not, use {@code
 * ExecutionException} or {@link UncheckedExecutionException}. This allows the client code to
 * continue to distinguish between exceptions and errors, even when they come from other threads.
 *
 * @author Chris Povirk
 * @since 10.0
 */
@GwtCompatible
public class ExecutionError extends Error {
  /*

from contextvars import ContextVar
from typing import Annotated, Any

import pytest
from fastapi import Depends, FastAPI, WebSocket
from fastapi.exceptions import FastAPIError
from fastapi.testclient import TestClient

global_context: ContextVar[dict[str, Any]] = ContextVar("global_context", default={})  # noqa: B039


class Session:
    def __init__(self) -> None:
        self.open = True


async def dep_session() -> Any:

        }

        // Wait for all threads to complete
        for (Thread thread : threads) {
            thread.join();
        }

        // Then - no exceptions should have been thrown
        assertTrue(true); // If we reach here, no exceptions occurred
    }

    /**
     * Test consistency across multiple calls
     */
    @Test
    @DisplayName("Test consistency across multiple calls")

            // Wait for completion
            for (Thread t : threads) {
                t.join();
            }

            // Verify thread safety - no unexpected exceptions
            assertEquals(0, exceptionCount.get(), "Should not have unexpected exceptions");

            // At least some threads should complete successfully
            assertTrue(successfulThreads.get() > 0, "At least some threads should complete successfully");

一般情况下可能不会需要在代码中直接使用响应头。

但对于某些高级应用场景，还是需要添加自定义响应头：

{* ../../docs_src/handling_errors/tutorial002.py hl[14] *}

## 安装自定义异常处理器

添加自定义处理器，要使用 [Starlette 的异常工具](https://www.starlette.dev/exceptions/)。

假设要触发的自定义异常叫作 `UnicornException`。

且需要 FastAPI 实现全局处理该异常。

此时，可以用 `@app.exception_handler()` 添加自定义异常控制器：

{* ../../docs_src/handling_errors/tutorial003.py hl[5:7,13:18,24] *}

        assertNull(delegatingConfig.getMaximumVersion(), "Should return null when delegate returns null");
    }

    @Test
    @DisplayName("Delegation should handle exceptions from delegate")
    void testExceptionHandling() {
        // Given
        RuntimeException testException = new RuntimeException("Test exception");
        when(mockDelegate.getResponseTimeout()).thenThrow(testException);

                            // as it would pull in dependency on:
                            // - maven-plugin-api (for model)
                            // - Plexus Container (for model supporting classes and exceptions)
                            XmlNode root = XmlService.read(is, null);
                            String groupId = mayGetChild(root, "groupId");
                            String artifactId = mayGetChild(root, "artifactId");

        assertNotNull(stringRep);
        assertTrue(stringRep.contains("SmbException"));
        assertTrue(stringRep.contains(message));
    }

    @Test
    @DisplayName("Should handle authentication exceptions")
    void testAuthenticationExceptions() {
        // Given
        int ntStatus = NtStatus.NT_STATUS_LOGON_FAILURE;

        // When
        SmbException exception = new SmbException(ntStatus, false);

import json
from typing import Annotated, Any

import pytest
from fastapi import APIRouter, Depends, FastAPI, HTTPException
from fastapi.exceptions import FastAPIError
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient


class Session:
    def __init__(self) -> None:
        self.open = True


def dep_session() -> Any:
    s = Session()
    yield s
    s.open = False