From e3d38718c00c14fcdf508f61ead53f9300a6fa7e Mon Sep 17 00:00:00 2001 From: Qiu Yuzhou Date: Sat, 24 Oct 2026 14:16:34 +0800 Subject: [PATCH] test: make HTTP and plugin checks deterministic --- .../Connectivity/ConnectivityHTTPTransport.c | 8 +- .../Connectivity/ConnectivityHTTPTransport.h | 5 + .../Connectivity/ConnectivityHTTPTests.swift | 130 ++++++++---------- .../HealthCheckSupersessionController.swift | 24 +--- .../Tests/Plugin/PluginCatalogTests.swift | 51 +++---- 5 files changed, 98 insertions(+), 120 deletions(-) diff --git a/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.c b/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.c index 6072ac76..8733bcda 100644 --- a/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.c +++ b/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.c @@ -73,6 +73,11 @@ static int response_status(const char *bytes, size_t length) { ConnectivityHTTPResult ConnectivityHTTPProbe(const char *request, int proxy_port, int timeout_ms, ConnectivityHTTPCancellation *token) { + return ConnectivityHTTPProbeWithReader(request, proxy_port, timeout_ms, token, NULL, NULL); +} + +ConnectivityHTTPResult ConnectivityHTTPProbeWithReader(const char *request, int proxy_port, + int timeout_ms, ConnectivityHTTPCancellation *token, ConnectivityHTTPReader reader, void *context) { ConnectivityHTTPResult result = { .stage = 1, .local_tcp_ms = -1, .send_ms = -1, .first_byte_ms = -1, .headers_ms = -1 }; int fd = -1; @@ -109,7 +114,8 @@ ConnectivityHTTPResult ConnectivityHTTPProbe(const char *request, int proxy_port size_t used = 0, consumed = 0, cumulative = 0; while (used < sizeof(bytes)) { if (!ready(fd, POLLIN, deadline, token)) goto cleanup; - ssize_t count = recv(fd, bytes + used, sizeof(bytes) - used, 0); + ssize_t count = reader ? reader(bytes + used, sizeof(bytes) - used, context) + : recv(fd, bytes + used, sizeof(bytes) - used, 0); if (count <= 0) { if (count < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) continue; goto cleanup; diff --git a/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.h b/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.h index c08ca0a2..3d327471 100644 --- a/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.h +++ b/ShadowsocksX-NG2/App/Application/Connectivity/ConnectivityHTTPTransport.h @@ -1,6 +1,7 @@ #ifndef ConnectivityHTTPTransport_h #define ConnectivityHTTPTransport_h #include +#include typedef struct ConnectivityHTTPCancellation ConnectivityHTTPCancellation; typedef struct { @@ -20,4 +21,8 @@ void ConnectivityHTTPDestroyCancellation(ConnectivityHTTPCancellation *token); uint64_t ConnectivityHTTPNow(void); ConnectivityHTTPResult ConnectivityHTTPProbe(const char *request, int proxy_port, int timeout_ms, ConnectivityHTTPCancellation *token); +/* A reader override makes response fragmentation independent of TCP packet boundaries. */ +typedef ssize_t (*ConnectivityHTTPReader)(void *buffer, size_t capacity, void *context); +ConnectivityHTTPResult ConnectivityHTTPProbeWithReader(const char *request, int proxy_port, + int timeout_ms, ConnectivityHTTPCancellation *token, ConnectivityHTTPReader reader, void *context); #endif diff --git a/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityHTTPTests.swift b/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityHTTPTests.swift index fb09c4c1..9acf7a41 100644 --- a/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityHTTPTests.swift +++ b/ShadowsocksX-NG2/Tests/Connectivity/ConnectivityHTTPTests.swift @@ -6,12 +6,14 @@ import Testing @MainActor struct ConnectivityHTTPTests { - @Test(arguments: [204, 200, 302, 500]) - func sendsGETThroughProxyAndStopsAtHeaders(status: Int) async throws { - let proxy = try HeaderProxy(status: status) + @Test(arguments: [(204, false), (200, false), (302, false), (500, false), (200, true)]) + func sendsGETThroughProxyAndStopsAtHeaders(status: Int, informational: Bool) async throws { + let headers = + "HTTP/1.1 \(status) Response\r\nLocation: http://another.invalid/redirect\r\nContent-Length: 99999\r\n\r\n" + let proxy = try HeaderProxy( + response: (informational ? "HTTP/1.1 100 Continue\r\n\r\n" : "") + headers) defer { proxy.stop() } let port = try await proxy.start() - let started = ContinuousClock.now // This name cannot resolve; a successful response proves proxy routing. let id = NodeID.fresh() let module = ServerConnectivity( @@ -30,18 +32,52 @@ struct ConnectivityHTTPTests { } else { #expect(value == .http(status)) } - #expect(ContinuousClock.now - started < .seconds(2)) #expect(proxy.requests.count == 1) #expect(proxy.requests.first?.hasPrefix("GET http://target.invalid/test HTTP/") == true) } + @Test func informationalAndFragmentedHeadersDoNotWaitForBody() async throws { + // Leave the socket readable while the reader supplies exact response chunks. + let proxy = try HeaderProxy(response: "ready") + defer { proxy.stop() } + let port = try await proxy.start() + let result = await Task.detached { + var chunks = [ + "HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\nContent-Len", + "gth: 999999\r\n\r\n", + ].map { Array($0.utf8) } + let cancellation = ConnectivityHTTPCreateCancellation() + defer { ConnectivityHTTPDestroyCancellation(cancellation) } + return withUnsafeMutablePointer(to: &chunks) { context in + "GET http://target.invalid/test HTTP/1.1\r\nHost: target.invalid\r\n\r\n".withCString { + ConnectivityHTTPProbeWithReader( + $0, Int32(port), 5000, cancellation, + { buffer, capacity, context in + let chunks = context!.assumingMemoryBound(to: [[UInt8]].self) + guard !chunks.pointee.isEmpty else { return 0 } + let chunk = chunks.pointee.removeFirst() + precondition(chunk.count <= capacity) + chunk.withUnsafeBytes { + buffer!.copyMemory(from: $0.baseAddress!, byteCount: $0.count) + } + return chunk.count + }, context) + } + } + }.value + #expect(result.received_headers == 1) + #expect(result.status == 200) + #expect(result.timed_out == 0) + } + @Test func fixedAddressPreservesHostAndReportsClientPhases() async throws { - let proxy = try HeaderProxy(status: 204) + let proxy = try HeaderProxy( + response: "HTTP/1.1 204 No Content\r\nContent-Length: 99999\r\n\r\n") defer { proxy.stop() } let port = try await proxy.start() let result = await ConnectivityHTTPRequest().observe( url: URL(string: "http://target.invalid:8080/test?q=private%20value")!, - targetIP: "2001:db8::1", proxyPort: port, timeout: 1) + targetIP: "2001:db8::1", proxyPort: port, timeout: 5) guard case .success = result.outcome else { Issue.record("Expected successful fixed-address request") return @@ -56,61 +92,29 @@ struct ConnectivityHTTPTests { #expect(result.totalMilliseconds >= 0) } - @Test func informationalAndFragmentedHeadersDoNotWaitForBody() async throws { - let proxy = try HeaderProxy( - status: nil, - parts: [ - "HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\nContent-Len", - "gth: 999999\r\n\r\n", - ]) - defer { proxy.stop() } - let port = try await proxy.start() - let result = await ConnectivityHTTPRequest().observe( - url: URL(string: "http://target.invalid/test")!, targetIP: "192.0.2.1", proxyPort: port, - timeout: 1) - guard case .success = result.outcome else { - Issue.record("Expected complete final headers") - return - } - #expect(result.firstByteMilliseconds != nil) - #expect(result.headersMilliseconds != nil) - #expect(result.totalMilliseconds < 1000) - } - @Test(arguments: [ "HTTP/2 200 OK\r\n\r\n", "HTTP/1.1 200 OK\r\nInvalid Header\r\n\r\n", "HTTP/1.1 200 OK\r\nX-Large: " + String(repeating: "x", count: 65536), String(repeating: "HTTP/1.1 100 Continue\r\n\r\n", count: 3000), + // Each block fits; only the cumulative budget including final headers is exceeded. + "HTTP/1.1 100 Continue\r\nX-Large: " + String(repeating: "x", count: 60000) + "\r\n\r\n" + + "HTTP/1.1 204 No Content\r\nX-Large: " + String(repeating: "x", count: 6000) + "\r\n\r\n", ]) func rejectsMalformedOrUnboundedHeaders(response: String) async throws { - let proxy = try HeaderProxy(status: nil, parts: [response]) + let proxy = try HeaderProxy(response: response) defer { proxy.stop() } let port = try await proxy.start() let result = await ConnectivityHTTPRequest().observe( url: URL(string: "http://target.invalid/test")!, targetIP: "192.0.2.1", proxyPort: port, - timeout: 1) + timeout: 5) #expect(result.outcome == .network) #expect(result.failureStage == "headers") } - @Test func cumulativeHeadersIncludeFinalBlockAfterInformationalResponses() async throws { - let proxy = try HeaderProxy( - status: nil, - parts: [ - String(repeating: "HTTP/1.1 100 Continue\r\n\r\n", count: 2500), - "HTTP/1.1 204 No Content\r\nX-Large: " + String(repeating: "x", count: 5000) + "\r\n\r\n", - ]) - defer { proxy.stop() } - let port = try await proxy.start() - let result = await ConnectivityHTTPRequest().observe( - url: URL(string: "http://target.invalid/test")!, targetIP: "192.0.2.1", proxyPort: port, - timeout: 1) - #expect(result.outcome == .network) - } - @Test func httpsIsRejectedBeforeOpeningProxyConnection() async throws { - let proxy = try HeaderProxy(status: 204) + let proxy = try HeaderProxy( + response: "HTTP/1.1 204 No Content\r\nContent-Length: 99999\r\n\r\n") defer { proxy.stop() } let port = try await proxy.start() let result = await ConnectivityHTTPRequest().perform( @@ -121,7 +125,7 @@ struct ConnectivityHTTPTests { @Test(.timeLimit(.minutes(1))) func timesOutWhileWaitingForFirstByte() async throws { - let proxy = try HeaderProxy(status: nil) + let proxy = try HeaderProxy() defer { proxy.stop() } let port = try await proxy.start() let result = await ConnectivityHTTPRequest().observe( @@ -134,7 +138,7 @@ struct ConnectivityHTTPTests { } @Test func cancellationEndsRequestWithoutWaitingForHeaders() async throws { - let proxy = try HeaderProxy(status: nil) + let proxy = try HeaderProxy() defer { proxy.stop() } let port = try await proxy.start() let id = NodeID.fresh() @@ -167,16 +171,14 @@ struct ConnectivityHTTPTests { /// Controlled HTTP proxy: headers arrive, but the declared body never does. private final class HeaderProxy: @unchecked Sendable { private let listener: NWListener - private let status: Int? - private let parts: [String]? + private let response: String? private let lock = NSLock() private var received: [String] = [] private var connections: [NWConnection] = [] var requests: [String] { lock.withLock { received } } - init(status: Int?, parts: [String]? = nil) throws { - self.status = status - self.parts = parts + init(response: String? = nil) throws { + self.response = response listener = try NWListener(using: .tcp, on: .any) } @@ -186,14 +188,9 @@ private final class HeaderProxy: @unchecked Sendable { connection.start(queue: DispatchQueue(label: UUID().uuidString)) connection.receive(minimumIncompleteLength: 1, maximumLength: 65536) { [self] data, _, _, _ in if let data { lock.withLock { received.append(String(data: data, encoding: .utf8) ?? "") } } - if let parts { - send(parts, through: connection) - return + if let response { + connection.send(content: Data(response.utf8), completion: .contentProcessed { _ in }) } - guard let status else { return } - let headers = - "HTTP/1.1 \(status) Response\r\nLocation: http://another.invalid/redirect\r\nContent-Length: 99999\r\n\r\n" - connection.send(content: Data(headers.utf8), completion: .contentProcessed { _ in }) } } return try await withCheckedThrowingContinuation { continuation in @@ -212,19 +209,6 @@ private final class HeaderProxy: @unchecked Sendable { } } - private func send(_ parts: [String], through connection: NWConnection) { - guard let first = parts.first else { return } - connection.send( - content: Data(first.utf8), - completion: .contentProcessed { [self] _ in - if parts.count > 1 { - DispatchQueue.global().asyncAfter(deadline: .now() + 0.01) { [self] in - send(Array(parts.dropFirst()), through: connection) - } - } - }) - } - func stop() { listener.cancel() listener.newConnectionHandler = nil diff --git a/ShadowsocksX-NG2/Tests/Plugin/HealthCheckSupersessionController.swift b/ShadowsocksX-NG2/Tests/Plugin/HealthCheckSupersessionController.swift index ccb35697..8c4d154e 100644 --- a/ShadowsocksX-NG2/Tests/Plugin/HealthCheckSupersessionController.swift +++ b/ShadowsocksX-NG2/Tests/Plugin/HealthCheckSupersessionController.swift @@ -2,26 +2,16 @@ import Foundation @testable import ShadowsocksX_NG2 -@MainActor -struct HealthCheckSupersessionControllerContext { - let directory: URL - let catalogFile: URL - let files: RuntimeFileStore - let plugins: PluginCatalog - let agent: ProxyRuntimeFixture.FakeLaunchAgent - let probe: ProxyRuntimeFixture.BlockingProbe -} - @MainActor func makeHealthCheckSupersessionController( - _ context: HealthCheckSupersessionControllerContext + runtime: ProxyRuntimeFixture.TemporaryRuntime, catalogFile: URL, plugins: PluginCatalog, + agent: ProxyRuntimeFixture.FakeLaunchAgent, probe: ProxyRuntimeFixture.BlockingProbe ) -> ProxyRuntimeController { - let directory = context.directory - let catalogFile = context.catalogFile - let files = context.files - let plugins = context.plugins - let agent = context.agent - let probe = context.probe + let directory = runtime.directory + let files = RuntimeFileStore(fileURL: runtime.contract) + agent.onRegister = { + try? files.writePublishedRuntimeReceipt(processID: 42) + } return ProxyRuntimeController( controlPreferences: makeTestProxyControlPreferences(), catalogSnapshotReader: ProxyRuntimeFixture.catalogSnapshotReader(at: catalogFile), diff --git a/ShadowsocksX-NG2/Tests/Plugin/PluginCatalogTests.swift b/ShadowsocksX-NG2/Tests/Plugin/PluginCatalogTests.swift index 855f49fd..fff1916e 100644 --- a/ShadowsocksX-NG2/Tests/Plugin/PluginCatalogTests.swift +++ b/ShadowsocksX-NG2/Tests/Plugin/PluginCatalogTests.swift @@ -208,38 +208,38 @@ struct PluginCatalogTests { inspector: QuietPluginInspector()) try plugins.addPlugin(program: "tool", path: first.path) var catalog = ConfigurationCatalog() - let id = try ActivationFixture.addPlainServer("Plugin", in: &catalog) - var fields = try #require(catalog.entry(for: id)).serverFieldsForPluginTest + var fields = ActivationFixture.plainFields(remark: "Plugin") fields.pluginProgram = "tool" - try catalog.updateServer(id, with: fields) + let id = try catalog.addServer(fields, to: nil) let catalogFile = runtime.directory.appendingPathComponent("catalog.json") try makeTestCatalogStore(fileURL: catalogFile).save(CatalogDocument(catalog: catalog)) let agent = ProxyRuntimeFixture.FakeLaunchAgent() let files = RuntimeFileStore(fileURL: runtime.contract) - agent.onRegister = { - try? files.writePublishedRuntimeReceipt(processID: 42) - } let probe = ProxyRuntimeFixture.BlockingProbe() let controller = makeHealthCheckSupersessionController( - HealthCheckSupersessionControllerContext( - directory: runtime.directory, catalogFile: catalogFile, files: files, - plugins: plugins, agent: agent, probe: probe)) + runtime: runtime, catalogFile: catalogFile, plugins: plugins, agent: agent, probe: probe) let started = XCTestExpectation(description: "old runtime entered health check") probe.arm(started: started) - let oldActivation = Task { try await controller.activate(id) } - let wait = await XCTWaiter.fulfillment(of: [started], timeout: 3) - #expect(wait == .completed) - let applied = AsyncStream.makeStream() plugins.connect( - invalidateRuntime: { controller.pluginMappingsDidPublish() }, revalidate: {}, - converge: { - _ = await controller.catalogDidCommit(snapshot: catalog) - applied.continuation.yield(()) - }) - try plugins.updatePlugin(program: "tool", path: second.path) - probe.release() - _ = try await oldActivation.value - for await _ in applied.stream { break } + invalidateRuntime: { controller.pluginMappingsDidPublish() }, revalidate: {}, converge: {}) + let oldActivation = Task { try await controller.activate(id) } + do { + try await requireTestPhase([started]) + try plugins.updatePlugin(program: "tool", path: second.path) + probe.release() + _ = try await oldActivation.value + } catch { + probe.release() + oldActivation.cancel() + _ = await oldActivation.result + throw error + } + // A later successful deployment must not hide stale health-check side effects. + #expect(controller.state == .starting) + #expect(files.readFailure() == nil) + #expect(agent.unregisterCount == 0) + let outcome = await controller.catalogDidCommit(snapshot: catalog) + #expect(outcome == .converged(skippedServers: [])) #expect(files.loadDocument()?.servers.first?.plugin == second.path) #expect(controller.state == .running) } @@ -327,13 +327,6 @@ private struct FailingMappingStore: PluginMappingStoring { func save(_ configuration: PluginConfiguration) throws { throw PluginMappingError.unreadable } } -extension CatalogEntry { - fileprivate var serverFieldsForPluginTest: ServerFields { - guard case .server(let fields) = kind else { preconditionFailure("expected server") } - return fields - } -} - private actor HeldPluginInspector: PluginInspecting { nonisolated let started: AsyncStream private let signal: AsyncStream.Continuation