llm4s-docs
API
Generated with
llm4s-docs/org.llm4s/org.llm4s.testkit/LocalProviderTestServer

LocalProviderTestServer

org.llm4s.testkit.LocalProviderTestServer
object LocalProviderTestServer

A local HTTP server to point a provider client at, so a test exercises the client's real request and response handling - including streaming - without a network or an API key.

It is the JDK's com.sun.net.httpserver.HttpServer on an ephemeral port, so it adds no dependency.

withServer("/v1/chat")(exchange => sendSseResponse(exchange, body)) { baseUrl =>
 val client = assertBuildsClient(AcmeProvider, section.copy(baseUrl = Some(BaseUrl(baseUrl))))
 assertStreams(client)
}

holdOpen and streamThenHold are handlers that never finish answering, for checking that a client's call can be interrupted (see ProviderModuleChecks.assertCancelsWhenInterrupted). Handlers run on virtual threads, so this needs JDK 21 (see docs/guide/writing-a-provider.md).

Attributes

Graph
Supertypes
class Object
trait Matchable
class Any
Self type

Members list

Value members

Concrete methods

def anthropicMessage(content: String, model: String): String

A minimal Anthropic Messages (/v1/messages) response carrying content as one text block.

A minimal Anthropic Messages (/v1/messages) response carrying content as one text block.

Attributes

def holdOpen(exchange: HttpExchange): Unit

Holds the request open without answering until the enclosing withServer block ends - a provider that never replies, for checking that a client's call can be interrupted.

Holds the request open without answering until the enclosing withServer block ends - a provider that never replies, for checking that a client's call can be interrupted.

Attributes

def openAICompletion(content: String, model: String): String

A minimal OpenAI-format (/chat/completions) completion response carrying content.

A minimal OpenAI-format (/chat/completions) completion response carrying content.

Attributes

def openAISseBody(chunks: Seq[String], model: String): String

An OpenAI-format streaming body: one SSE data: event per chunk of content, the last carrying finish_reason and usage, then data: [DONE].

An OpenAI-format streaming body: one SSE data: event per chunk of content, the last carrying finish_reason and usage, then data: [DONE].

Attributes

def sendJsonResponse(exchange: HttpExchange, statusCode: Int, body: String): Unit

Sends a JSON response with the given status code and body.

Sends a JSON response with the given status code and body.

Attributes

def sendSseResponse(exchange: HttpExchange, body: String): Unit

Sends a 200 SSE (text/event-stream) response with the given raw body.

Sends a 200 SSE (text/event-stream) response with the given raw body.

Attributes

def streamThenHold(exchange: HttpExchange, firstEvent: String, contentType: String): Unit

Starts a 200 streaming response, sends firstEvent (one complete event in the provider's stream format), then holds the response open like holdOpen.

Starts a 200 streaming response, sends firstEvent (one complete event in the provider's stream format), then holds the response open like holdOpen.

Attributes

def withServer(path: String)(handler: HttpExchange => Unit)(test: String => Any): Unit

Starts a local HTTP server with handler bound to path, runs test with the server's base URL (http://localhost:<port>, no trailing slash), then stops the server - also when test throws, whose exception is then rethrown.

Starts a local HTTP server with handler bound to path, runs test with the server's base URL (http://localhost:<port>, no trailing slash), then stops the server - also when test throws, whose exception is then rethrown.

Attributes

Generated with