test: make HTTP and plugin checks deterministic

2026-10-29 05:33:45 +00:00 · 2026-10-24 14:16:34 +08:00
parent 0757af988f
commit e3d38718c0
5 changed files with 98 additions and 120 deletions
@@ -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;
@@ -1,6 +1,7 @@
 #ifndef ConnectivityHTTPTransport_h
 #define ConnectivityHTTPTransport_h
 #include <stdint.h>
+#include <sys/types.h>

 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
@@ -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
@@ -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),
@@ -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<Void>.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<Void>
  private let signal: AsyncStream<Void>.Continuation