Alawn/ShadowsocksX-NG
1
0
Fork 0
mirror of https://github.com/shadowsocks/ShadowsocksX-NG.git synced 2026-10-29 05:33:45 +00:00
Code Releases Activity

fix(tests): avoid blocking async test workers

Browse Source
This commit is contained in:
Qiu Yuzhou
2026-10-23 03:46:19 +08:00
parent 17848b6fde
commit 09e4d2e097
19 changed files with 252 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ShadowsocksX-NG2/App/Application/ProxyRuntime/ProxyRuntimeController+SystemProxyGate.swift
+2 -2
View File
@@ -60,7 +60,7 @@ extension ProxyRuntimeController {
private func firewallStatus(for executableURL: URL) async -> FirewallBlockStatus {
let checker = firewallChecker
return await Task.detached(priority: .utility) {
checker.status(for: executableURL)
await checker.status(for: executableURL)
}.value
}
@@ -69,7 +69,7 @@ extension ProxyRuntimeController {
) async -> EndpointHealthProbe.Outcome {
let probe = probe
return await Task.detached(priority: .utility) {
probe.probe(host: host, port: port, timeout: timeout)
await probe.probe(host: host, port: port, timeout: timeout)
}.value
}
ShadowsocksX-NG2/App/Application/ProxyRuntime/SystemEndpointProbe.swift
+2 -2
View File
@@ -2,11 +2,11 @@ import Foundation
/// 端点探测缝：注入以便控制器单测（真实探测走 EndpointHealthProbe）。
protocol EndpointProbing: Sendable {
func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome
func probe(host: String, port: Int, timeout: TimeInterval) async -> EndpointHealthProbe.Outcome
}
struct SystemEndpointProbe: EndpointProbing {
func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome {
func probe(host: String, port: Int, timeout: TimeInterval) async -> EndpointHealthProbe.Outcome {
EndpointHealthProbe.probe(host: host, port: port, timeout: timeout)
}
}
ShadowsocksX-NG2/App/Application/Services/FirewallStatusChecker.swift
+6 -2
View File
@@ -7,7 +7,7 @@ enum FirewallBlockStatus: Equatable, Sendable {
}
protocol FirewallStatusChecking: Sendable {
func status(for executableURL: URL) -> FirewallBlockStatus
func status(for executableURL: URL) async -> FirewallBlockStatus
}
/// `socketfilterfw --getappblocked` 是应用防火墙拒绝态的只读系统边界。
@@ -23,7 +23,11 @@ struct SocketFilterFirewallChecker: FirewallStatusChecking, @unchecked Sendable
self.runCommand = runCommand
}
func status(for executableURL: URL) -> FirewallBlockStatus {
func status(for executableURL: URL) async -> FirewallBlockStatus {
statusSynchronously(for: executableURL)
}
private func statusSynchronously(for executableURL: URL) -> FirewallBlockStatus {
guard
let output = try? runCommand(
Self.toolURL, ["--getappblocked", executableURL.path])
ShadowsocksX-NG2/App/Application/SettingsWorkflow/ListenerOccupancy.swift
+10 -11
View File
@@ -154,11 +154,11 @@ final class ListenerOccupancy: ObservableObject {
let excluded = Set(ProxyEndpointKind.allCases.map { context.proposedListen.port(for: $0) })
let probe = probe
let candidate = await Task.detached(priority: .utility) {
PortSuggestion.firstFree(excluding: excluded) { port in
await PortSuggestion.firstFreeAsync(excluding: excluded) { port in
let request = PortOccupancyProbeRequest(
endpoint: endpoint,
listen: context.proposedListen.replacingPort(port, for: endpoint), port: port)
if case .free = probe.occupancy(for: request) { return true }
if case .free = await probe.occupancy(for: request) { return true }
return false
}
}.value
@@ -202,15 +202,14 @@ final class ListenerOccupancy: ObservableObject {
-> [SettingsPortID: PortOccupancy]
{
await Task.detached(priority: .utility) {
Dictionary(
uniqueKeysWithValues: SettingsPortID.allCases.map { id in
let endpoint = SettingsPortAdapter.endpoint(for: id, group: context.draft.group)
return (
id,
probe.occupancy(
for: PortOccupancyProbeRequest(endpoint: endpoint, listen: context.proposedListen))
)
})
var results: [SettingsPortID: PortOccupancy] = [:]
for id in SettingsPortID.allCases {
let endpoint = SettingsPortAdapter.endpoint(for: id, group: context.draft.group)
let request = PortOccupancyProbeRequest(
endpoint: endpoint, listen: context.proposedListen)
results[id] = await probe.occupancy(for: request)
}
return results
}.value
}
}
ShadowsocksX-NG2/Domain/Rules/BuiltinRuleCatalog.swift
+2 -2
View File
@@ -52,7 +52,7 @@ struct BuiltinRuleCatalog {
/// App-process source facts. Concurrent consumers share one attempt per source;
/// successful bundled snapshots never expire, failed attempts require explicit retry.
actor BuiltinRuleSnapshots {
typealias Loader = @Sendable (RulesSource) throws -> RuleSnapshot
typealias Loader = @Sendable (RulesSource) async throws -> RuleSnapshot
private struct Attempt {
let id: UUID
let task: Task<RuleSnapshot, Error>
@@ -81,7 +81,7 @@ actor BuiltinRuleSnapshots {
attempt = Attempt(
id: UUID(),
task: Task.detached(priority: .userInitiated) {
let snapshot = try loader(source)
let snapshot = try await loader(source)
guard let kind = source.sourceKind, kind != .custom else {
throw RuleSnapshotError.missing
}
ShadowsocksX-NG2/Domain/Runtime/PortOccupancyProbe.swift
+2 -2
View File
@@ -148,7 +148,7 @@ struct PortOccupancyProbeRequest: Equatable, Sendable {
/// 端口占用探测缝：设置编辑期的即时占用校验。激活是否成功
/// 仍以 runtime 实际绑定为准（健康门禁），本探测不构成权威判定。
protocol PortOccupancyProbing: Sendable {
func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy
func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy
}
/// 系统实现：对 `bindAddress:port` 做一次不带复用选项的 TCP bind + listen，
@@ -156,7 +156,7 @@ protocol PortOccupancyProbing: Sendable {
/// 允许共存，到 listen 阶段才报告冲突，因此两步都要覆盖。占用时尽力以
/// `lsof` 解析监听进程名。探测不设置 SO_REUSEADDR/SO_REUSEPORT，避免假空闲。
struct SystemPortOccupancyProbe: PortOccupancyProbing {
func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy {
func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy {
occupancy(port: request.port, listenerMode: request.listenerMode)
}
ShadowsocksX-NG2/Domain/Runtime/ProxyPortSemantics.swift
+9
View File
@@ -39,6 +39,15 @@ enum PortSuggestion {
}
return nil
}
static func firstFreeAsync(
excluding excludedPorts: Set<Int>, isFree: (Int) async -> Bool
) async -> Int? {
for candidate in range where !excludedPorts.contains(candidate) {
if await isFree(candidate) { return candidate }
}
return nil
}
}
extension ProxyListeningConfiguration {
ShadowsocksX-NG2/Tests/ProxyControl/ProxyPortSemanticsTests.swift
+2 -2
View File
@@ -105,14 +105,14 @@ final class ProxyPortSemanticsTests: XCTestCase {
}
}
func testBindProbeReportsUnknownWhenBindIsNotPermitted() throws {
func testBindProbeReportsUnknownWhenBindIsNotPermitted() async throws {
try XCTSkipIf(getuid() == 0, "root 可绑定特权端口，无法用 EACCES 构造 .unknown")
// 特权端口：非 root 进程 bind 失败（EACCES）≠ 占用，也不得冒充空闲。
let request = PortOccupancyProbeRequest(
endpoint: .socks,
listen: RuntimeListenFacts(listen: ProxyListeningConfiguration()),
port: 80)
let outcome = SystemPortOccupancyProbe().occupancy(for: request)
let outcome = await SystemPortOccupancyProbe().occupancy(for: request)
guard case .unknown = outcome else {
return XCTFail("绑定失败时必须报 .unknown，不得冒充空闲或占用，实际 \(outcome)")
ShadowsocksX-NG2/Tests/ProxyRuntime/ProxyRuntimeControllerModeTests.swift
+4 -4
View File
@@ -72,12 +72,12 @@ extension ProxyRuntimeControllerModeTests {
func testAgentOffDuringModeRuleDerivationSupersedesDeployment() async throws {
let seeded = try makeSeededCatalog()
let started = expectation(description: "rule snapshot load paused")
let release = DispatchSemaphore(value: 0)
defer { release.signal() }
let release = TestAsyncGate("rule snapshot load release")
defer { release.release() }
let loader = BuiltinRuleSnapshots(loader: { source -> RuleSnapshot in
if source == .geolocationCN {
started.fulfill()
release.wait()
_ = await release.wait()
}
return rulesFixture(source)
})
@@ -94,7 +94,7 @@ extension ProxyRuntimeControllerModeTests {
await controller.setAgentEnabled(false)
XCTAssertEqual(controller.state, .off)
let registrationsAfterStop = agent.registerCount
release.signal()
release.release()
await modeSwitch.value
XCTAssertEqual(agent.registerCount, registrationsAfterStop, "派生被取代后不得部署")
ShadowsocksX-NG2/Tests/Rules/BuiltinRuleCatalogTests.swift
+8 -7
View File
@@ -9,11 +9,12 @@ final class BuiltinRuleCatalogTests: XCTestCase {
func testConcurrentConsumersShareOneSourceLoad() async throws {
let started = expectation(description: "source read")
let loader = SharedRuleSourceLoader(started: started)
let snapshots = BuiltinRuleSnapshots(loader: { try loader.load($0) })
defer { loader.release.release() }
let snapshots = BuiltinRuleSnapshots(loader: { try await loader.load($0) })
async let first = snapshots.load(.gfwlist)
await fulfillment(of: [started], timeout: 3)
async let second = snapshots.load(.gfwlist)
loader.release.signal()
loader.release.release()
let (one, two) = try await (first, second)
XCTAssertEqual(one, two)
_ = try await snapshots.load(.gfwlist)
@@ -22,7 +23,7 @@ final class BuiltinRuleCatalogTests: XCTestCase {
func testRetryReadsOnlyFailedSourcesAndSuccessNeverExpires() async throws {
let loader = SharedRuleSourceLoader()
let snapshots = BuiltinRuleSnapshots(loader: { try loader.load($0) })
let snapshots = BuiltinRuleSnapshots(loader: { try await loader.load($0) })
let first = await snapshots.browsingSources()
XCTAssertThrowsError(try first[.chinaIPv4]?.get())
_ = await snapshots.browsingSources()
@@ -37,7 +38,7 @@ final class BuiltinRuleCatalogTests: XCTestCase {
@MainActor
func testFirstBrowsingRetainsRuntimeSourceFailureUntilExplicitRetry() async throws {
let loader = SharedRuleSourceLoader()
let snapshots = BuiltinRuleSnapshots(loader: { try loader.load($0) })
let snapshots = BuiltinRuleSnapshots(loader: { try await loader.load($0) })
do {
_ = try await snapshots.load(.chinaIPv4)
XCTFail("The first runtime source request should fail")
@@ -187,17 +188,17 @@ private final class SharedRuleSourceLoader: @unchecked Sendable {
private let lock = NSLock()
private var counts: [RulesSource: Int] = [:]
private let started: XCTestExpectation?
let release = DispatchSemaphore(value: 0)
let release = TestAsyncGate("shared rule source release")
init(started: XCTestExpectation? = nil) { self.started = started }
func count(_ source: RulesSource) -> Int { lock.withLock { counts[source, default: 0] } }
func load(_ source: RulesSource) throws -> RuleSnapshot {
func load(_ source: RulesSource) async throws -> RuleSnapshot {
let count = lock.withLock {
counts[source, default: 0] += 1
return counts[source, default: 0]
}
if let started {
started.fulfill()
release.wait()
_ = await release.wait()
} else if source == .chinaIPv4 && count == 1 {
throw RuleSnapshotError.missing
}
ShadowsocksX-NG2/Tests/Rules/RuleEnablementControllerTests.swift
+4 -4
View File
@@ -124,7 +124,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
store: store,
settings: ProxyRuntimeSettings(
listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true),
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) }))
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { await loader.load($0) }))
try await controller.activate(seeded.server)
let runtimeStore = RuntimeFileStore(fileURL: runtime.contract)
let previous = try XCTUnwrap(runtimeStore.loadDocument())
@@ -164,7 +164,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
store: store,
settings: ProxyRuntimeSettings(
listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true),
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) }))
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { await loader.load($0) }))
try await controller.activate(seeded.server)
let modeSwitch = Task { await controller.setProxyMode(.rule) }
try await withTestTask(
@@ -187,7 +187,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
let started = expectation(description: "background source preparation")
let loader = PausedRuntimeRuleSource(started: started)
defer { loader.resume() }
let snapshots = BuiltinRuleSnapshots(loader: { loader.load($0) })
let snapshots = BuiltinRuleSnapshots(loader: { await loader.load($0) })
let settings = ProxyRuntimeSettings(
listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true)
let controller = makeRuleEnablementController(
@@ -237,7 +237,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
store: store,
settings: ProxyRuntimeSettings(
listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true),
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) }))
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { await loader.load($0) }))
try await controller.activate(seeded.server)
let old = Task { await controller.setProxyMode(.rule) }
try await withTestTask(
ShadowsocksX-NG2/Tests/Runtime/FirewallStatusCheckerTests.swift
+8 -5
View File
@@ -3,7 +3,7 @@ import XCTest
@testable import ShadowsocksX_NG2
final class FirewallStatusCheckerTests: XCTestCase {
func testGetAppBlockedOutputMapsToBlockedStatus() {
func testGetAppBlockedOutputMapsToBlockedStatus() async {
var capturedExecutable: URL?
var capturedArguments: [String] = []
let checker = SocketFilterFirewallChecker { executable, arguments in
@@ -13,14 +13,15 @@ final class FirewallStatusCheckerTests: XCTestCase {
}
let target = URL(fileURLWithPath: "/bundle/Helpers/sslocal")
XCTAssertEqual(checker.status(for: target), .blocked)
let blocked = await checker.status(for: target)
XCTAssertEqual(blocked, .blocked)
XCTAssertEqual(
capturedExecutable?.path,
"/usr/libexec/ApplicationFirewall/socketfilterfw")
XCTAssertEqual(capturedArguments, ["--getappblocked", target.path])
}
func testPermittedAndUnknownResultsDegradeWithoutFalseBlock() {
func testPermittedAndUnknownResultsDegradeWithoutFalseBlock() async {
let permitted = SocketFilterFirewallChecker { _, _ in
"Incoming connection to /bundle/Helpers/sslocal is permitted.\n"
}
@@ -29,7 +30,9 @@ final class FirewallStatusCheckerTests: XCTestCase {
}
let target = URL(fileURLWithPath: "/bundle/Helpers/sslocal")
XCTAssertEqual(permitted.status(for: target), .permitted)
XCTAssertEqual(unavailable.status(for: target), .unknown)
let permittedStatus = await permitted.status(for: target)
let unavailableStatus = await unavailable.status(for: target)
XCTAssertEqual(permittedStatus, .permitted)
XCTAssertEqual(unavailableStatus, .unknown)
}
}
ShadowsocksX-NG2/Tests/Settings/ListenerOccupancyLifecycleTests.swift
+2 -2
View File
@@ -155,7 +155,7 @@ struct ListenerOccupancyLifecycleTests {
private final class SuggestionProbe: PortOccupancyProbing, Sendable {
let gate = GatedOccupancyProbe(gatedAnswer: .free, passThroughAnswer: .free)
func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy {
request.port >= PortSuggestion.range.lowerBound ? gate.occupancy(for: request) : .free
func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy {
request.port >= PortSuggestion.range.lowerBound ? await gate.occupancy(for: request) : .free
}
}
ShadowsocksX-NG2/Tests/Support/BoundedTestWait.swift
+58 -12
View File
@@ -46,11 +46,16 @@ func withTestTask<Value: Sendable>(
await finish()
}
/// Broadcast release is remembered, including when cleanup runs before a worker enters.
final class TestBlockingGate: @unchecked Sendable {
private let group = DispatchGroup()
/// Broadcast async release is remembered, and timed-out waiters do not occupy a worker thread.
final class TestAsyncGate: @unchecked Sendable {
private struct Waiter {
let continuation: CheckedContinuation<Bool, Never>
var timeoutTask: Task<Void, Never>?
}
private let lock = NSLock()
private var released = false
private var waiters: [UUID: Waiter] = [:]
private let onTimeout: @Sendable () -> Void
init(
@@ -61,7 +66,6 @@ final class TestBlockingGate: @unchecked Sendable {
if let onTimeout {
self.onTimeout = onTimeout
} else if let test = Test.current {
// Detached workers have no current test; retain the creating test and source location.
let message = Comment(
rawValue: "Timed out waiting for \(description) (created by \(test.name))")
let location = SourceLocation(
@@ -70,22 +74,64 @@ final class TestBlockingGate: @unchecked Sendable {
} else {
self.onTimeout = { XCTFail("Timed out waiting for \(description)", file: file, line: line) }
}
group.enter()
}
func wait(timeout: TimeInterval = 10) -> Bool {
guard group.wait(timeout: .now() + timeout) == .success else {
onTimeout()
return false
func wait(timeout: Duration = .seconds(10)) async -> Bool {
let id = UUID()
return await withCheckedContinuation { continuation in
lock.lock()
guard !released else {
lock.unlock()
continuation.resume(returning: true)
return
}
waiters[id] = Waiter(continuation: continuation)
lock.unlock()
let timeoutTask = Task { [weak self] in
do {
try await Task.sleep(for: timeout)
self?.expire(id)
} catch {
// Release cancels the timer after resuming the waiter.
}
}
lock.lock()
if var waiter = waiters[id] {
waiter.timeoutTask = timeoutTask
waiters[id] = waiter
lock.unlock()
} else {
lock.unlock()
timeoutTask.cancel()
}
}
return true
}
func release() {
lock.lock()
let shouldRelease = !released
guard !released else {
lock.unlock()
return
}
released = true
let pending = Array(waiters.values)
waiters.removeAll()
lock.unlock()
if shouldRelease { group.leave() }
for waiter in pending {
waiter.timeoutTask?.cancel()
waiter.continuation.resume(returning: true)
}
}
private func expire(_ id: UUID) {
lock.lock()
let waiter = waiters.removeValue(forKey: id)
lock.unlock()
guard let waiter else { return }
onTimeout()
waiter.continuation.resume(returning: false)
}
}
ShadowsocksX-NG2/Tests/Support/ProxyRuntimeBlockingFixtures.swift
+86
View File
@@ -0,0 +1,86 @@
import Foundation
import XCTest
@testable import ShadowsocksX_NG2
extension ProxyRuntimeFixture {
/// 可暂停探测替身：arm 后的下一次探测异步等待测试放行，在启动健康门的
/// await 窗口内确定性插入交错命令，不占用线程池工作线程。
final class BlockingProbe: EndpointProbing, @unchecked Sendable {
private let lock = NSLock()
private var gate: TestAsyncGate?
private var started: XCTestExpectation?
func arm(started: XCTestExpectation) {
lock.lock()
defer { lock.unlock() }
precondition(gate == nil, "上一次 arm 尚未 release")
gate = TestAsyncGate("runtime fixture release")
self.started = started
}
func release() {
lock.lock()
let gate = self.gate
self.gate = nil
self.started = nil
lock.unlock()
gate?.release()
}
func probe(host: String, port: Int, timeout: TimeInterval) async
-> EndpointHealthProbe.Outcome
{
let (gate, started) = armedGate()
if let gate {
started?.fulfill()
_ = await gate.wait()
}
return .reachable
}
private func armedGate() -> (TestAsyncGate?, XCTestExpectation?) {
lock.lock()
let gate = self.gate
let started = self.started
lock.unlock()
return (gate, started)
}
}
/// 按调用序返回预设结果，并可在指定调用处异步等待测试放行。
final class BlockingFirewallChecker: FirewallStatusChecking, @unchecked Sendable {
private let lock = NSLock()
private var callCount = 0
private let outcomes: [FirewallBlockStatus]
private let blockOnCall: Int
private let started: XCTestExpectation
private let gate = TestAsyncGate("runtime fixture release")
init(outcomes: [FirewallBlockStatus], blockOnCall: Int, started: XCTestExpectation) {
precondition(!outcomes.isEmpty)
self.outcomes = outcomes
self.blockOnCall = blockOnCall
self.started = started
}
func release() { gate.release() }
func status(for executableURL: URL) async -> FirewallBlockStatus {
let call = nextCall()
if call == blockOnCall {
started.fulfill()
_ = await gate.wait()
}
return outcomes[min(call, outcomes.count) - 1]
}
private func nextCall() -> Int {
lock.lock()
callCount += 1
let call = callCount
lock.unlock()
return call
}
}
}
ShadowsocksX-NG2/Tests/Support/ProxyRuntimeFixture.swift
+12 -71
View File
@@ -206,7 +206,13 @@ enum ProxyRuntimeFixture {
return requestedPorts
}
func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome {
func probe(host: String, port: Int, timeout: TimeInterval) async
-> EndpointHealthProbe.Outcome
{
recordProbe(port: port)
}
private func recordProbe(port: Int) -> EndpointHealthProbe.Outcome {
lock.lock()
defer { lock.unlock() }
callCount += 1
@@ -239,7 +245,11 @@ enum ProxyRuntimeFixture {
self.outcomes = outcomes
}
func status(for executableURL: URL) -> FirewallBlockStatus {
func status(for executableURL: URL) async -> FirewallBlockStatus {
recordStatus(for: executableURL)
}
private func recordStatus(for executableURL: URL) -> FirewallBlockStatus {
lock.lock()
defer { lock.unlock() }
recordedURLs.append(executableURL)
@@ -249,75 +259,6 @@ enum ProxyRuntimeFixture {
}
}
/// 可暂停探测替身：arm 后的下一次探测挂起在信号量上，直到测试放行——
/// 在启动健康门的 await 窗口内确定性插入交错命令（探测运行在 detached
/// 线程，不占主 actor）。放行后恢复直通。
final class BlockingProbe: EndpointProbing, @unchecked Sendable {
private let lock = NSLock()
private var gate: TestBlockingGate?
private var started: XCTestExpectation?
func arm(started: XCTestExpectation) {
lock.lock()
defer { lock.unlock() }
precondition(gate == nil, "上一次 arm 尚未 release")
gate = TestBlockingGate("runtime fixture release")
self.started = started
}
func release() {
lock.lock()
let gate = self.gate
self.gate = nil
self.started = nil
lock.unlock()
gate?.release()
}
func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome {
lock.lock()
let gate = self.gate
let started = self.started
lock.unlock()
if let gate {
started?.fulfill()
_ = gate.wait()
}
return .reachable
}
}
/// 按调用序返回预设结果、并可在第 N 次调用处挂起的防火墙检查替身：
/// 钉住「轮询 await 恢复后代际已推进时不得再写状态」的回归。
final class BlockingFirewallChecker: FirewallStatusChecking, @unchecked Sendable {
private let lock = NSLock()
private var callCount = 0
private let outcomes: [FirewallBlockStatus]
private let blockOnCall: Int
private let started: XCTestExpectation
private let gate = TestBlockingGate("runtime fixture release")
init(outcomes: [FirewallBlockStatus], blockOnCall: Int, started: XCTestExpectation) {
precondition(!outcomes.isEmpty)
self.outcomes = outcomes
self.blockOnCall = blockOnCall
self.started = started
}
func release() { gate.release() }
func status(for executableURL: URL) -> FirewallBlockStatus {
lock.lock()
callCount += 1
let call = callCount
lock.unlock()
if call == blockOnCall {
started.fulfill()
_ = gate.wait()
}
return outcomes[min(call, outcomes.count) - 1]
}
}
}
/// 每个夹具使用独立 defaults 域，最后一个引用释放时清理持久值。
ShadowsocksX-NG2/Tests/Support/RuleRuntimeDelayGate.swift
+3 -3
View File
@@ -71,13 +71,13 @@ func withRuleApplicationTask<Value: Sendable>(
final class PausedRuntimeRuleSource: @unchecked Sendable {
let started: XCTestExpectation
private let release = TestBlockingGate("rule source release")
private let release = TestAsyncGate("rule source release")
init(started: XCTestExpectation) { self.started = started }
func resume() { release.release() }
func load(_ source: RulesSource) -> RuleSnapshot {
func load(_ source: RulesSource) async -> RuleSnapshot {
if source == .geolocationCN {
started.fulfill()
_ = release.wait()
_ = await release.wait()
}
return rulesFixture(source)
}
ShadowsocksX-NG2/Tests/Support/SettingsWorkflowTestSupport.swift
+23 -8
View File
@@ -163,7 +163,11 @@ final class FakeOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
answerOverride = answer
}
func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy {
func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy {
occupancySynchronously(for: request)
}
private func occupancySynchronously(for request: PortOccupancyProbeRequest) -> PortOccupancy {
lock.lock()
defer { lock.unlock() }
requestedPorts.append(request.port)
@@ -199,8 +203,14 @@ final class FakeOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
/// 被扣住，`passThrough` 之后的调用立即返回新应答。用于把「旧一轮」结果钉在
/// 新一轮之后落盘，验证过期结果不得覆盖。
final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
private struct Invocation {
let isFirstEntry: Bool
let isGated: Bool
let answer: PortOccupancy
}
private let lock = NSLock()
private let gate = TestBlockingGate("occupancy probe release")
private let gate = TestAsyncGate("occupancy probe release")
private var isGated = true
private var enteredCount = 0
private let entry = XCTestExpectation(description: "occupancy probe entered")
@@ -239,17 +249,22 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
gate.release()
}
func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy {
func occupancy(for request: PortOccupancyProbeRequest) async -> PortOccupancy {
let invocation = record(request)
if invocation.isFirstEntry { entry.fulfill() }
if invocation.isGated, !(await gate.wait()) {
return .unknown(detail: "Test fixture release timed out")
}
return invocation.answer
}
private func record(_ request: PortOccupancyProbeRequest) -> Invocation {
lock.lock()
enteredCount += 1
let firstEntry = enteredCount == 1
let gated = isGated
let answer = gated ? gatedAnswer : passThroughAnswer
lock.unlock()
if firstEntry { entry.fulfill() }
if gated, !gate.wait() {
return .unknown(detail: "Test fixture release timed out")
}
return answer
return Invocation(isFirstEntry: firstEntry, isGated: gated, answer: answer)
}
}
ShadowsocksX-NG2/Tests/TestSupport/BoundedTestWaitTests.swift
+9 -10
View File
@@ -40,27 +40,26 @@ struct BoundedTestWaitTests {
await #expect(throws: CancellationError.self) { try await task.value }
}
@Test func blockingGateTimesOutInsteadOfRetainingAWorker() async {
let gate = TestBlockingGate(onTimeout: {})
let worker = Task.detached { gate.wait(timeout: 0.02) }
#expect(await worker.value == false)
@Test func asyncGateTimesOutWithoutBlockingAWorker() async {
let gate = TestAsyncGate(onTimeout: {})
#expect(await gate.wait(timeout: .milliseconds(20)) == false)
}
@Test func blockingGateRemembersEarlyAndRepeatedRelease() {
let gate = TestBlockingGate()
@Test func asyncGateRemembersEarlyAndRepeatedRelease() async {
let gate = TestAsyncGate()
gate.release()
gate.release()
#expect(gate.wait(timeout: 0))
#expect(await gate.wait(timeout: .zero))
}
@Test func blockingGateReleasesEveryWorker() async throws {
let gate = TestBlockingGate()
@Test func asyncGateReleasesEveryWorker() async throws {
let gate = TestAsyncGate()
let entered = XCTestExpectation(description: "workers entered")
entered.expectedFulfillmentCount = 4
let workers = (0..<4).map { _ in
Task.detached {
entered.fulfill()
return gate.wait()
return await gate.wait()
}
}
try await requireTestPhase([entered])