Merge branch 'setup' into template
This commit is contained in:
@@ -1,8 +1,7 @@
|
||||
import Logging
|
||||
import MySQLNIO
|
||||
import NIOCore
|
||||
import NIOPosix
|
||||
import NIOSSL
|
||||
import PostgresNIO
|
||||
import Testing
|
||||
|
||||
@testable import Persistence
|
||||
@@ -13,46 +12,81 @@ import Testing
|
||||
)
|
||||
struct TLSTests {
|
||||
|
||||
// MARK: Properties tests
|
||||
// MARK: Methods tests
|
||||
|
||||
@Test
|
||||
func `off has no TLS configuration`() {
|
||||
#expect(TLS.off.tlsConfiguration == nil)
|
||||
}
|
||||
|
||||
@Test(arguments: [
|
||||
TLS.prefer,
|
||||
TLS.require
|
||||
])
|
||||
func `maps to the default client configuration`(
|
||||
for tls: TLS
|
||||
) throws {
|
||||
let configuration = try #require(tls.tlsConfiguration)
|
||||
|
||||
#expect(configuration.bestEffortEquals(.makeClientConfiguration()))
|
||||
}
|
||||
|
||||
@Test
|
||||
func `prefer falls back to plaintext when the server offers no TLS`() async throws {
|
||||
// The fake server never advertises `CLIENT_SSL`, so this connection can only succeed by downgrading to
|
||||
// plaintext — pinning the driver behavior the `prefer` posture relies on.
|
||||
let server = try await PlaintextMySQLServer.start()
|
||||
let tlsConfiguration = try #require(TLS.prefer.tlsConfiguration)
|
||||
let connection = try await MySQLConnection.connect(
|
||||
to: .init(ipAddress: "127.0.0.1", port: server.port),
|
||||
username: "loud",
|
||||
database: "loud",
|
||||
tlsConfiguration: tlsConfiguration,
|
||||
logger: Logger(label: "test"),
|
||||
on: MultiThreadedEventLoopGroup.singleton.any()
|
||||
).get()
|
||||
func `off connects in plaintext`() async throws {
|
||||
// With TLS disabled the client skips the `SSLRequest` and sends its startup message directly,
|
||||
// which the fake server answers in plaintext.
|
||||
let server = try await PlaintextPostgresServer.start()
|
||||
let connection = try await connect(to: server, tls: .off)
|
||||
|
||||
let isConnected = !connection.isClosed
|
||||
|
||||
try await connection.close().get()
|
||||
try await connection.close()
|
||||
try await server.stop()
|
||||
|
||||
#expect(isConnected)
|
||||
}
|
||||
|
||||
@Test
|
||||
func `prefer falls back to plaintext when the server offers no TLS`() async throws {
|
||||
// The fake server refuses the `SSLRequest`, so this connection can only succeed by downgrading
|
||||
// to plaintext — pinning the driver behavior the `prefer` posture relies on.
|
||||
let server = try await PlaintextPostgresServer.start()
|
||||
let connection = try await connect(to: server, tls: .prefer)
|
||||
|
||||
let isConnected = !connection.isClosed
|
||||
|
||||
try await connection.close()
|
||||
try await server.stop()
|
||||
|
||||
#expect(isConnected)
|
||||
}
|
||||
|
||||
@Test
|
||||
func `require refuses the connection when the server offers no TLS`() async throws {
|
||||
// The fake server refuses the `SSLRequest`, so the driver must fail the connection instead of
|
||||
// downgrading — pinning the refusal the `require` posture promises.
|
||||
let server = try await PlaintextPostgresServer.start()
|
||||
|
||||
let error = await #expect(throws: PSQLError.self) {
|
||||
_ = try await connect(to: server, tls: .require)
|
||||
}
|
||||
|
||||
try await server.stop()
|
||||
|
||||
#expect(error?.code == .sslUnsupported)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// MARK: - Helpers
|
||||
|
||||
private extension TLSTests {
|
||||
|
||||
/// Opens a connection to the given fake server with the given TLS posture.
|
||||
/// - Parameters:
|
||||
/// - server: the fake server to connect to.
|
||||
/// - tls: the TLS posture to connect with.
|
||||
/// - Returns: the open connection, to be closed by the caller.
|
||||
func connect(
|
||||
to server: PlaintextPostgresServer,
|
||||
tls: TLS
|
||||
) async throws -> PostgresConnection {
|
||||
try await PostgresConnection.connect(
|
||||
on: MultiThreadedEventLoopGroup.singleton.any(),
|
||||
configuration: .init(
|
||||
host: "127.0.0.1",
|
||||
port: server.port,
|
||||
username: "loud",
|
||||
password: "loud",
|
||||
database: "loud",
|
||||
tls: tls.postgresTLS()
|
||||
),
|
||||
id: 1,
|
||||
logger: Logger(label: "test")
|
||||
)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -15,7 +15,7 @@ struct ProbeTests {
|
||||
|
||||
@Test
|
||||
func `reports a reachable database`() async throws {
|
||||
let service = Service(
|
||||
let service = try Service(
|
||||
driver: .inMemory,
|
||||
logger: Logger(label: "test")
|
||||
)
|
||||
@@ -33,8 +33,8 @@ struct ProbeTests {
|
||||
func `reports an unreachable database`() async throws {
|
||||
// Port 1 on the loopback interface has nothing listening, so the connection is refused
|
||||
// immediately instead of timing out.
|
||||
let service = Service(
|
||||
driver: .mysql(
|
||||
let service = try Service(
|
||||
driver: .postgres(
|
||||
.init(
|
||||
host: "127.0.0.1",
|
||||
port: 1,
|
||||
@@ -42,7 +42,8 @@ struct ProbeTests {
|
||||
username: "nobody",
|
||||
password: "nothing",
|
||||
tls: .off,
|
||||
maxConnectionsPerEventLoop: 1
|
||||
maxConnectionsPerEventLoop: 1,
|
||||
poolTimeout: .seconds(10)
|
||||
)
|
||||
),
|
||||
logger: Logger(label: "test")
|
||||
@@ -57,6 +58,47 @@ struct ProbeTests {
|
||||
#expect(!isReachable)
|
||||
}
|
||||
|
||||
@Test
|
||||
func `reports a hanging database as unreachable within its timeout`() async throws {
|
||||
// The silent server accepts the TCP connection and never answers, so the probe's query can only
|
||||
// ever be resolved by its deadline — without one, it would wait out the driver's own connect
|
||||
// timeout (10 seconds) instead.
|
||||
let server = try await SilentPostgresServer.start()
|
||||
let service = try Service(
|
||||
driver: .postgres(
|
||||
.init(
|
||||
host: "127.0.0.1",
|
||||
port: server.port,
|
||||
name: "hanging",
|
||||
username: "nobody",
|
||||
password: "nothing",
|
||||
tls: .off,
|
||||
maxConnectionsPerEventLoop: 1,
|
||||
poolTimeout: .seconds(10)
|
||||
)
|
||||
),
|
||||
logger: Logger(label: "test")
|
||||
)
|
||||
let fluent = service()
|
||||
let probe = Probe(
|
||||
fluent: fluent,
|
||||
timeout: .milliseconds(100)
|
||||
)
|
||||
|
||||
let clock = ContinuousClock()
|
||||
let start = clock.now
|
||||
let isReachable = await probe()
|
||||
let elapsed = clock.now - start
|
||||
|
||||
try await fluent.shutdown()
|
||||
try await server.stop()
|
||||
|
||||
#expect(!isReachable)
|
||||
// Well past the 100-millisecond deadline to absorb scheduling noise, yet far below the driver's
|
||||
// 10-second connect timeout — only the deadline can answer this fast.
|
||||
#expect(elapsed < .seconds(5))
|
||||
}
|
||||
|
||||
@Test
|
||||
func `reports a default database that is not an SQL database`() async throws {
|
||||
let fluent = Fluent(logger: Logger(label: "test"))
|
||||
|
||||
@@ -15,7 +15,7 @@ struct ServiceTests {
|
||||
|
||||
@Test
|
||||
func `registers an SQLite database as the default for the in-memory driver`() async throws {
|
||||
let service = Service(
|
||||
let service = try Service(
|
||||
driver: .inMemory,
|
||||
logger: Logger(label: "test")
|
||||
)
|
||||
@@ -31,19 +31,20 @@ struct ServiceTests {
|
||||
}
|
||||
|
||||
@Test
|
||||
func `registers a MySQL database as the default for the mysql driver`() async throws {
|
||||
func `registers a PostgreSQL database as the default for the postgres driver`() async throws {
|
||||
// Resolving the default database opens no connection — pooling is lazy — so no server
|
||||
// needs to be listening on the configured host and port.
|
||||
let service = Service(
|
||||
driver: .mysql(
|
||||
let service = try Service(
|
||||
driver: .postgres(
|
||||
.init(
|
||||
host: "127.0.0.1",
|
||||
port: 3306,
|
||||
port: 5432,
|
||||
name: "site",
|
||||
username: "site",
|
||||
password: "site",
|
||||
tls: .off,
|
||||
maxConnectionsPerEventLoop: 1
|
||||
maxConnectionsPerEventLoop: 1,
|
||||
poolTimeout: .seconds(10)
|
||||
)
|
||||
),
|
||||
logger: Logger(label: "test")
|
||||
@@ -56,12 +57,12 @@ struct ServiceTests {
|
||||
|
||||
let dialect = try #require(database?.dialect)
|
||||
|
||||
#expect(dialect.name == "mysql")
|
||||
#expect(dialect.name == "postgresql")
|
||||
}
|
||||
|
||||
@Test
|
||||
func `builds a usable in-memory database`() async throws {
|
||||
let service = Service(
|
||||
let service = try Service(
|
||||
driver: .inMemory,
|
||||
logger: Logger(label: "test")
|
||||
)
|
||||
@@ -87,12 +88,12 @@ struct ServiceTests {
|
||||
}
|
||||
|
||||
@Test(
|
||||
"mysql: migrate, insert, read back",
|
||||
.enabled(if: mysqlDriver != nil)
|
||||
"postgres: migrate, insert, read back",
|
||||
.enabled(if: postgresDriver != nil)
|
||||
)
|
||||
func mysqlRoundTrip() async throws {
|
||||
func postgresRoundTrip() async throws {
|
||||
try await roundTrip(
|
||||
driver: mysqlDriver!,
|
||||
driver: postgresDriver!,
|
||||
revertAfter: true
|
||||
)
|
||||
}
|
||||
@@ -116,11 +117,11 @@ private extension ServiceTests {
|
||||
revertAfter: Bool = false
|
||||
) async throws {
|
||||
let prepareDB = PrepareDB()
|
||||
let service = Service(
|
||||
let service = try Service(
|
||||
driver: driver,
|
||||
logger: Logger(label: "test")
|
||||
)
|
||||
|
||||
|
||||
let fluent = service()
|
||||
|
||||
do {
|
||||
@@ -147,25 +148,26 @@ private extension ServiceTests {
|
||||
|
||||
}
|
||||
|
||||
/// The MySQL/MariaDB driver built from the `MYSQL_TEST_*` environment variables, or `nil` when the gate
|
||||
/// variable `MYSQL_TEST_HOST` is unset — in which case the MySQL integration test is skipped, so the suite
|
||||
/// stays runnable with no database available.
|
||||
private let mysqlDriver: Persistence.Driver? = {
|
||||
/// The PostgreSQL driver built from the `POSTGRES_TEST_*` environment variables, or `nil` when the gate
|
||||
/// variable `POSTGRES_TEST_HOST` is unset — in which case the PostgreSQL integration test is skipped, so
|
||||
/// the suite stays runnable with no database available.
|
||||
private let postgresDriver: Persistence.Driver? = {
|
||||
let environment = ProcessInfo.processInfo.environment
|
||||
|
||||
guard let host = environment["MYSQL_TEST_HOST"] else {
|
||||
guard let host = environment["POSTGRES_TEST_HOST"] else {
|
||||
return nil
|
||||
}
|
||||
|
||||
return .mysql(
|
||||
return .postgres(
|
||||
.init(
|
||||
host: host,
|
||||
port: environment["MYSQL_TEST_PORT"].flatMap(Int.init) ?? 3306,
|
||||
name: environment["MYSQL_TEST_NAME"] ?? "site",
|
||||
username: environment["MYSQL_TEST_USERNAME"] ?? "site",
|
||||
password: environment["MYSQL_TEST_PASSWORD"] ?? "site",
|
||||
port: environment["POSTGRES_TEST_PORT"].flatMap(Int.init) ?? 5432,
|
||||
name: environment["POSTGRES_TEST_NAME"] ?? "site",
|
||||
username: environment["POSTGRES_TEST_USERNAME"] ?? "site",
|
||||
password: environment["POSTGRES_TEST_PASSWORD"] ?? "site",
|
||||
tls: .off,
|
||||
maxConnectionsPerEventLoop: 2
|
||||
maxConnectionsPerEventLoop: 2,
|
||||
poolTimeout: .seconds(10)
|
||||
)
|
||||
)
|
||||
}()
|
||||
|
||||
@@ -1,162 +0,0 @@
|
||||
import NIOCore
|
||||
import NIOPosix
|
||||
|
||||
/// A fake MySQL server speaking just enough of the wire protocol to complete a plaintext handshake.
|
||||
///
|
||||
/// Its greeting advertises the `mysql_native_password` plugin but **not** the `CLIENT_SSL` capability, and
|
||||
/// it answers the client's handshake response with a bare OK packet — so a client asking for TLS can only
|
||||
/// end up connected in plaintext. This is what the `prefer` fallback test connects to, proving the driver
|
||||
/// downgrades to plaintext rather than refusing the connection.
|
||||
final class PlaintextMySQLServer {
|
||||
|
||||
// MARK: Properties
|
||||
|
||||
/// The port the server listens on, assigned by the system at bind time.
|
||||
let port: Int
|
||||
|
||||
/// The listening channel the server accepts connections through.
|
||||
private let channel: Channel
|
||||
|
||||
// MARK: Initializers
|
||||
|
||||
private init(
|
||||
channel: Channel,
|
||||
port: Int
|
||||
) {
|
||||
self.channel = channel
|
||||
self.port = port
|
||||
}
|
||||
|
||||
// MARK: Functions
|
||||
|
||||
/// Starts a server on the loopback interface, on a system-assigned port.
|
||||
/// - Returns: the running server, ready to be connected to at ``port``.
|
||||
static func start() async throws -> PlaintextMySQLServer {
|
||||
let channel = try await ServerBootstrap(group: MultiThreadedEventLoopGroup.singleton)
|
||||
.childChannelInitializer { channel in
|
||||
channel.eventLoop.makeCompletedFuture {
|
||||
try channel.pipeline.syncOperations.addHandler(Handler())
|
||||
}
|
||||
}
|
||||
.bind(host: "127.0.0.1", port: 0)
|
||||
.get()
|
||||
|
||||
guard let port = channel.localAddress?.port else {
|
||||
throw ChannelError.unknownLocalAddress
|
||||
}
|
||||
|
||||
return .init(
|
||||
channel: channel,
|
||||
port: port
|
||||
)
|
||||
}
|
||||
|
||||
/// Stops the server, closing its listening channel.
|
||||
func stop() async throws {
|
||||
try await channel.close().get()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// MARK: - Handlers
|
||||
|
||||
private extension PlaintextMySQLServer {
|
||||
|
||||
/// Greets a freshly accepted connection, accepts whatever authentication response arrives,
|
||||
/// and closes on anything after that (e.g. a `COM_QUIT`).
|
||||
final class Handler: ChannelInboundHandler {
|
||||
|
||||
// MARK: Type aliases
|
||||
|
||||
typealias InboundIn = ByteBuffer
|
||||
typealias OutboundOut = ByteBuffer
|
||||
|
||||
// MARK: Properties
|
||||
|
||||
/// Whether the client's handshake response has already been answered with an OK packet.
|
||||
private var didAuthenticate = false
|
||||
|
||||
// MARK: Functions
|
||||
|
||||
func channelActive(context: ChannelHandlerContext) {
|
||||
context.writeAndFlush(
|
||||
wrapOutboundOut(Self.greeting(allocator: context.channel.allocator)),
|
||||
promise: nil
|
||||
)
|
||||
}
|
||||
|
||||
func channelRead(
|
||||
context: ChannelHandlerContext,
|
||||
data: NIOAny
|
||||
) {
|
||||
guard didAuthenticate else {
|
||||
didAuthenticate = true
|
||||
|
||||
context.writeAndFlush(
|
||||
wrapOutboundOut(Self.ok(allocator: context.channel.allocator)),
|
||||
promise: nil
|
||||
)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
context.close(promise: nil)
|
||||
}
|
||||
|
||||
// MARK: Helpers
|
||||
|
||||
/// The `HandshakeV10` greeting, framed and ready to send as the connection's first packet.
|
||||
///
|
||||
/// The advertised capabilities are `CLIENT_LONG_PASSWORD`, `CLIENT_PROTOCOL_41`,
|
||||
/// `CLIENT_SECURE_CONNECTION`, and `CLIENT_PLUGIN_AUTH` — deliberately **not** `CLIENT_SSL`,
|
||||
/// so the client cannot upgrade the connection to TLS.
|
||||
private static func greeting(allocator: ByteBufferAllocator) -> ByteBuffer {
|
||||
var payload = allocator.buffer(capacity: 80)
|
||||
|
||||
payload.writeInteger(10, endianness: .little, as: UInt8.self) // protocol version
|
||||
payload.writeNullTerminatedString("8.0.0") // server version
|
||||
payload.writeInteger(1, endianness: .little, as: UInt32.self) // connection id
|
||||
payload.writeBytes([1, 2, 3, 4, 5, 6, 7, 8]) // auth plugin data, part 1
|
||||
payload.writeInteger(0, endianness: .little, as: UInt8.self) // filler
|
||||
payload.writeInteger(0x8201, endianness: .little, as: UInt16.self) // capabilities, lower: LONG_PASSWORD | PROTOCOL_41 | SECURE_CONNECTION
|
||||
payload.writeInteger(0x21, endianness: .little, as: UInt8.self) // character set (utf8)
|
||||
payload.writeInteger(0x0002, endianness: .little, as: UInt16.self) // status flags (autocommit)
|
||||
payload.writeInteger(0x0008, endianness: .little, as: UInt16.self) // capabilities, upper: PLUGIN_AUTH
|
||||
payload.writeInteger(21, endianness: .little, as: UInt8.self) // auth plugin data length
|
||||
payload.writeBytes([UInt8](repeating: 0, count: 10)) // reserved
|
||||
payload.writeBytes([9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 0]) // auth plugin data, part 2
|
||||
payload.writeNullTerminatedString("mysql_native_password") // auth plugin name
|
||||
|
||||
return framed(payload, sequence: 0, allocator: allocator)
|
||||
}
|
||||
|
||||
/// A bare OK packet, framed as the reply to the client's handshake response.
|
||||
private static func ok(allocator: ByteBufferAllocator) -> ByteBuffer {
|
||||
var payload = allocator.buffer(capacity: 8)
|
||||
|
||||
payload.writeBytes([0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00]) // OK, no rows, autocommit, no warnings
|
||||
|
||||
return framed(payload, sequence: 2, allocator: allocator)
|
||||
}
|
||||
|
||||
/// Wraps a payload in the MySQL packet frame: a 3-byte little-endian length and a sequence byte.
|
||||
private static func framed(
|
||||
_ payload: ByteBuffer,
|
||||
sequence: UInt8,
|
||||
allocator: ByteBufferAllocator
|
||||
) -> ByteBuffer {
|
||||
var packet = allocator.buffer(capacity: payload.readableBytes + 4)
|
||||
var payload = payload
|
||||
|
||||
packet.writeInteger(UInt8(payload.readableBytes & 0xff))
|
||||
packet.writeInteger(UInt8((payload.readableBytes >> 8) & 0xff))
|
||||
packet.writeInteger(UInt8((payload.readableBytes >> 16) & 0xff))
|
||||
packet.writeInteger(sequence)
|
||||
packet.writeBuffer(&payload)
|
||||
|
||||
return packet
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
import NIOCore
|
||||
import NIOPosix
|
||||
|
||||
/// A fake PostgreSQL server speaking just enough of the wire protocol to complete a plaintext startup.
|
||||
///
|
||||
/// It answers the client's `SSLRequest` with `'N'` (no SSL) and the subsequent startup message with
|
||||
/// `AuthenticationOk`, `BackendKeyData`, and `ReadyForQuery` — so a client asking for TLS can only end up
|
||||
/// connected in plaintext. This is what the `prefer` fallback test connects to, proving the driver
|
||||
/// downgrades to plaintext rather than refusing the connection, and what the `require` test connects to,
|
||||
/// proving the driver refuses the connection instead of downgrading.
|
||||
final class PlaintextPostgresServer {
|
||||
|
||||
// MARK: Properties
|
||||
|
||||
/// The port the server listens on, assigned by the system at bind time.
|
||||
let port: Int
|
||||
|
||||
/// The listening channel the server accepts connections through.
|
||||
private let channel: Channel
|
||||
|
||||
// MARK: Initializers
|
||||
|
||||
private init(
|
||||
channel: Channel,
|
||||
port: Int
|
||||
) {
|
||||
self.channel = channel
|
||||
self.port = port
|
||||
}
|
||||
|
||||
// MARK: Functions
|
||||
|
||||
/// Starts a server on the loopback interface, on a system-assigned port.
|
||||
/// - Returns: the running server, ready to be connected to at ``port``.
|
||||
static func start() async throws -> PlaintextPostgresServer {
|
||||
let channel = try await ServerBootstrap(group: MultiThreadedEventLoopGroup.singleton)
|
||||
.childChannelInitializer { channel in
|
||||
channel.eventLoop.makeCompletedFuture {
|
||||
try channel.pipeline.syncOperations.addHandler(Handler())
|
||||
}
|
||||
}
|
||||
.bind(host: "127.0.0.1", port: 0)
|
||||
.get()
|
||||
|
||||
guard let port = channel.localAddress?.port else {
|
||||
throw ChannelError.unknownLocalAddress
|
||||
}
|
||||
|
||||
return .init(
|
||||
channel: channel,
|
||||
port: port
|
||||
)
|
||||
}
|
||||
|
||||
/// Stops the server, closing its listening channel.
|
||||
func stop() async throws {
|
||||
try await channel.close().get()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// MARK: - Handlers
|
||||
|
||||
private extension PlaintextPostgresServer {
|
||||
|
||||
/// Refuses the client's `SSLRequest`, accepts whatever startup message arrives, and closes on anything
|
||||
/// after that (e.g. a `Terminate`).
|
||||
///
|
||||
/// Unlike MySQL, the PostgreSQL client speaks first, so nothing is written on `channelActive`.
|
||||
final class Handler: ChannelInboundHandler {
|
||||
|
||||
// MARK: Type aliases
|
||||
|
||||
typealias InboundIn = ByteBuffer
|
||||
typealias OutboundOut = ByteBuffer
|
||||
|
||||
// MARK: Enumerations
|
||||
|
||||
/// The startup phases the connection moves through.
|
||||
private enum State {
|
||||
case awaitingSSLRequest
|
||||
case awaitingStartup
|
||||
case established
|
||||
}
|
||||
|
||||
// MARK: Constants
|
||||
|
||||
/// The magic code identifying an `SSLRequest` message.
|
||||
private static let sslRequestCode: Int32 = 80877103
|
||||
|
||||
// MARK: Properties
|
||||
|
||||
/// The startup phase the connection is currently in.
|
||||
private var state: State = .awaitingSSLRequest
|
||||
|
||||
// MARK: Functions
|
||||
|
||||
func channelRead(
|
||||
context: ChannelHandlerContext,
|
||||
data: NIOAny
|
||||
) {
|
||||
let buffer = unwrapInboundIn(data)
|
||||
|
||||
switch state {
|
||||
case .awaitingSSLRequest:
|
||||
// Peek past the Int32 length at the Int32 code: an `SSLRequest` is refused with a bare
|
||||
// 'N', while a direct startup message (a client connecting with TLS disabled) is
|
||||
// answered straight away.
|
||||
guard buffer.getInteger(at: buffer.readerIndex + 4, as: Int32.self) == Self.sslRequestCode else {
|
||||
completeStartup(context: context)
|
||||
return
|
||||
}
|
||||
|
||||
state = .awaitingStartup
|
||||
|
||||
var refusal = context.channel.allocator.buffer(capacity: 1)
|
||||
|
||||
refusal.writeInteger(UInt8(ascii: "N"))
|
||||
|
||||
context.writeAndFlush(
|
||||
wrapOutboundOut(refusal),
|
||||
promise: nil
|
||||
)
|
||||
case .awaitingStartup:
|
||||
completeStartup(context: context)
|
||||
case .established:
|
||||
context.close(promise: nil)
|
||||
}
|
||||
}
|
||||
|
||||
/// Answers a startup message and marks the connection established.
|
||||
private func completeStartup(context: ChannelHandlerContext) {
|
||||
state = .established
|
||||
|
||||
context.writeAndFlush(
|
||||
wrapOutboundOut(Self.startupResponse(allocator: context.channel.allocator)),
|
||||
promise: nil
|
||||
)
|
||||
}
|
||||
|
||||
// MARK: Helpers
|
||||
|
||||
/// The reply completing a plaintext startup: `AuthenticationOk`, `BackendKeyData`, and
|
||||
/// `ReadyForQuery` in a single flush.
|
||||
///
|
||||
/// `BackendKeyData` is not optional filler — the client requires it before `ReadyForQuery` by
|
||||
/// default and fails the connection when it is missing.
|
||||
private static func startupResponse(allocator: ByteBufferAllocator) -> ByteBuffer {
|
||||
var buffer = allocator.buffer(capacity: 32)
|
||||
|
||||
buffer.writeInteger(UInt8(ascii: "R")) // AuthenticationOk
|
||||
buffer.writeInteger(Int32(8))
|
||||
buffer.writeInteger(Int32(0))
|
||||
|
||||
buffer.writeInteger(UInt8(ascii: "K")) // BackendKeyData
|
||||
buffer.writeInteger(Int32(12))
|
||||
buffer.writeInteger(Int32(1)) // process id
|
||||
buffer.writeInteger(Int32(0)) // secret key
|
||||
|
||||
buffer.writeInteger(UInt8(ascii: "Z")) // ReadyForQuery
|
||||
buffer.writeInteger(Int32(5))
|
||||
buffer.writeInteger(UInt8(ascii: "I")) // idle
|
||||
|
||||
return buffer
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
import NIOCore
|
||||
import NIOPosix
|
||||
|
||||
/// A fake server accepting connections and never answering.
|
||||
///
|
||||
/// A client connecting to it completes the TCP handshake and then waits forever for the first protocol byte — the shape of a database that hangs rather
|
||||
/// than refuses. This is what the probe's deadline test connects to, proving the probe answers within its timeout instead of hanging alongside the server.
|
||||
final class SilentPostgresServer {
|
||||
|
||||
// MARK: Properties
|
||||
|
||||
/// The port the server listens on, assigned by the system at bind time.
|
||||
let port: Int
|
||||
|
||||
/// The listening channel the server accepts connections through.
|
||||
private let channel: Channel
|
||||
|
||||
// MARK: Initializers
|
||||
|
||||
private init(
|
||||
channel: Channel,
|
||||
port: Int
|
||||
) {
|
||||
self.channel = channel
|
||||
self.port = port
|
||||
}
|
||||
|
||||
// MARK: Functions
|
||||
|
||||
/// Starts a server on the loopback interface, on a system-assigned port.
|
||||
/// - Returns: the running server, ready to be connected to at ``port``.
|
||||
static func start() async throws -> SilentPostgresServer {
|
||||
let channel = try await ServerBootstrap(group: MultiThreadedEventLoopGroup.singleton)
|
||||
.bind(host: "127.0.0.1", port: 0)
|
||||
.get()
|
||||
|
||||
guard let port = channel.localAddress?.port else {
|
||||
throw ChannelError.unknownLocalAddress
|
||||
}
|
||||
|
||||
return .init(
|
||||
channel: channel,
|
||||
port: port
|
||||
)
|
||||
}
|
||||
|
||||
/// Stops the server, closing its listening channel.
|
||||
func stop() async throws {
|
||||
try await channel
|
||||
.close()
|
||||
.get()
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user