test: bound asynchronous fixture waits

This commit is contained in:
Qiu Yuzhou
2026-10-19 04:02:05 +08:00
parent e5b418761e
commit f1214c4cf5
10 changed files with 483 additions and 173 deletions
+3
View File
@@ -34,6 +34,7 @@ ShadowsocksX-NG 的现代化重写版本。新工程的全部源码与构建配
跨进程契约按**文件**编入其它 target:`Runtime/` 的 10 个文件同时编入 wrapper(`RuntimePaths`、`SslocalRuntimeDocument`、`RuntimeLog` 等),`SystemProxy/` 全部 6 个文件编入特权 helper。`project.yml` 里这 16 条是显式路径,在 `Domain/` 内移动文件必须同步,漏改会让 `xcodegen generate` 以缺文件直接失败。布局守卫(`Tests/Architecture/SkeletonTests.swift` 的 `DomainLayoutTests`)断言根下无平铺 `.swift`、树下只有 `.swift`(非源码文件会被 XcodeGen 收进 Resources 阶段并封进签名后的 bundle)。
- `Tests/` — 单元测试 target,随 `ShadowsocksX-NG2` scheme 运行。目录即主题,根下不留平铺测试文件:
- `Architecture/` — 跨切面工程约束守卫(分层与布局、平台效应 seam、开窗落点、清单完整性),扫源码或构建产物而不驱动行为。
- `TestSupport/` — 测试等待与清理 helper 的故障回归用例。
- `Support/` — 夹具、替身与 helper,不含用例(`CatalogFixtures`、`ProxyRuntimeFixture`、`SystemProxyFakes`、`AppArtifact`、`TestSourceTree` 等)。
- `Agent/`、`Activation/`、`Catalog/`(工作流用例在 `Catalog/Workflow/`)、`Subscription/`、`Rules/`(自定义规则在 `Rules/Custom/`)、`ProxyRuntime/`、`ProxyControl/`、`Runtime/`、`Settings/`、`SystemProxy/`、`Plugin/`、`Diagnostics/`、`Credentials/`、`Workspace/`、`Servers/` — 按功能领域的用例,与生产代码的领域主题和 `*Workflow` 同名同界。
- `Smoke/` — 真实进程冒烟,独立 target 与 scheme(`task test:smoke`)。
@@ -97,6 +98,8 @@ task --exit-code --taskfile ShadowsocksX-NG2/Taskfile.dist.yml validate -- lint
统一入口成功表示所选关卡的命令通过,仍需核对证据是否完整;hostless、真实进程 smoke、原生 UI 和真实系统代理的验收边界仍分别成立,发布打包门槛见下文。
`task test` 与 `task test:smoke` 显式启用 Xcode 单测试超时:默认 120 秒,上限 300 秒。这是 runner 兜底,不保证测试任务正常清理,也不是性能门槛。规则阶段等待失败会抛错退出测试体;受保护的测试任务先释放夹具,再取消并等待结束。规则延迟 gate 和同步阻塞夹具的默认等待上限为 10 秒,超时必须报告失败。共享 `AsyncGate` 保留由用例控制的取消语义。Swift Testing 的 detached worker 超时报告保留创建用例名称和源码位置;应同时核对命令退出码与测试计数,不能仅凭单例通过数判断成功。
在本目录内,日常格式化、静态检查、工程生成、构建和测试优先使用 `Taskfile.dist.yml` 定义的任务:
```bash
+3
View File
@@ -5,6 +5,7 @@ vars:
PROJECT_NAME: "ShadowsocksX-NG2"
BUILD_DIR: "{{ .TASKFILE_DIR }}/.build"
DERIVED_DATA_DIR: "{{ .BUILD_DIR }}/derivedData"
TEST_TIMEOUT_ARGS: '-test-timeouts-enabled YES -default-test-execution-time-allowance 120 -maximum-test-execution-time-allowance 300'
tasks:
validate:
@@ -167,6 +168,7 @@ tasks:
-destination 'platform=macOS,arch=arm64'
-derivedDataPath {{ .DERIVED_DATA_DIR }}
{{if .RESULT_BUNDLE_PATH}}-resultBundlePath {{.RESULT_BUNDLE_PATH | shellQuote}}{{end}}
{{.TEST_TIMEOUT_ARGS}}
test
test:binaries:
@@ -192,6 +194,7 @@ tasks:
-destination 'platform=macOS,arch=arm64'
-derivedDataPath {{ .DERIVED_DATA_DIR }}
{{if .RESULT_BUNDLE_PATH}}-resultBundlePath {{.RESULT_BUNDLE_PATH | shellQuote}}{{end}}
{{.TEST_TIMEOUT_ARGS}}
test
clean:
@@ -38,23 +38,28 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
store: store,
settings: ProxySettings(
listen: ActivationFixture.listen, preferredMode: .rule, agentEnabled: true),
proxyMode: .rule, ruleApplicationDelay: { await applicationGate.wait() })
proxyMode: .rule, ruleApplicationDelay: { try await applicationGate.wait() })
try await controller.activate(seeded.server)
let before = agent.registerCount
let one = await controller.commitRuleDocument(CustomRuleDocument(rules: [first]))
await fulfillment(of: [applicationStarted], timeout: 3)
let two = await controller.commitRuleDocument(CustomRuleDocument(rules: [latest]))
XCTAssertEqual(one.outcome, .saved)
XCTAssertEqual(two.outcome, .saved)
XCTAssertEqual(try store.load(), [latest])
XCTAssertEqual(
agent.registerCount, before, "Saving does not wait for or start deployment inline")
applicationGate.resume()
await controller.ruleApplicationTask?.value
XCTAssertEqual(agent.registerCount, before + 1)
let content = try activeACLContent(RuntimeFileStore(fileURL: runtime.contract))
XCTAssertTrue(content.contains("latest-save.example"))
XCTAssertFalse(content.contains("first-save.example"))
try await withRuleApplicationTask(
try XCTUnwrap(controller.ruleApplicationTask), controller: controller,
release: applicationGate.resume,
body: {
try await requireTestPhase([applicationStarted], timeout: 3)
let two = await controller.commitRuleDocument(CustomRuleDocument(rules: [latest]))
XCTAssertEqual(one.outcome, .saved)
XCTAssertEqual(two.outcome, .saved)
XCTAssertEqual(try store.load(), [latest])
XCTAssertEqual(
agent.registerCount, before, "Saving does not wait for or start deployment inline")
applicationGate.resume()
await controller.ruleApplicationTask?.value
XCTAssertEqual(agent.registerCount, before + 1)
let content = try activeACLContent(RuntimeFileStore(fileURL: runtime.contract))
XCTAssertTrue(content.contains("latest-save.example"))
XCTAssertFalse(content.contains("first-save.example"))
})
}
func testSaveDuringDeploymentIsAcceptedAndLatestDocumentConvergesAfterIt() async throws {
@@ -70,7 +75,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
settings: ProxySettings(
listen: ActivationFixture.listen, preferredMode: .rule, agentEnabled: true),
proxyMode: .rule, launchHealthTimeoutSeconds: 0.3,
launchHealthRetryDelay: { await healthGate.wait() })
launchHealthRetryDelay: { try await healthGate.wait() })
try await controller.activate(seeded.server)
let runtimeStore = RuntimeFileStore(fileURL: runtime.contract)
let started = expectation(description: "first deployment awaiting receipt")
@@ -88,20 +93,26 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
}
let one = await controller.commitRuleDocument(CustomRuleDocument(rules: [first]))
XCTAssertEqual(one.outcome, .saved)
await fulfillment(of: [started, healthWaiting], timeout: 3)
let two = await controller.commitRuleDocument(CustomRuleDocument(rules: [latest]))
XCTAssertEqual(two.outcome, .saved)
XCTAssertEqual(try store.load(), [latest])
XCTAssertEqual(agent.registerCount, before + 1, "A new save cannot overlap rule deployments")
try runtimeStore.writeRuntimeReceipt(for: XCTUnwrap(firstDocument), processID: 42)
healthGate.resume()
await controller.ruleApplicationTask?.value
XCTAssertEqual(try store.load(), [latest])
XCTAssertEqual(agent.registerCount, before + 2)
let content = try activeACLContent(runtimeStore)
XCTAssertTrue(content.contains("after-flight.example"))
XCTAssertFalse(content.contains("in-flight.example"))
XCTAssertEqual(controller.state, .running)
try await withRuleApplicationTask(
try XCTUnwrap(controller.ruleApplicationTask), controller: controller,
release: healthGate.resume,
body: {
try await requireTestPhase([started, healthWaiting], timeout: 3)
let two = await controller.commitRuleDocument(CustomRuleDocument(rules: [latest]))
XCTAssertEqual(two.outcome, .saved)
XCTAssertEqual(try store.load(), [latest])
XCTAssertEqual(
agent.registerCount, before + 1, "A new save cannot overlap rule deployments")
try runtimeStore.writeRuntimeReceipt(for: XCTUnwrap(firstDocument), processID: 42)
healthGate.resume()
await controller.ruleApplicationTask?.value
XCTAssertEqual(try store.load(), [latest])
XCTAssertEqual(agent.registerCount, before + 2)
let content = try activeACLContent(runtimeStore)
XCTAssertTrue(content.contains("after-flight.example"))
XCTAssertFalse(content.contains("in-flight.example"))
XCTAssertEqual(controller.state, .running)
})
}
func testNewPreparationStopsRemainingActionsOfAnAwaitingRuntimeOperation() async throws {
@@ -109,7 +120,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
let (store, _) = try makeCustomRuleStore()
let started = expectation(description: "new source preparation")
let loader = PausedRuntimeRuleSource(started: started)
defer { loader.release.signal() }
defer { loader.resume() }
let controller = makeRuleEnablementController(
store: store,
settings: ProxySettings(
@@ -122,17 +133,26 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
let unregistered = expectation(description: "old operation waiting for wrapper exit")
agent.onUnregister = { unregistered.fulfill() }
let oldOperation = Task { await controller.execute(.stop, document: nil) }
await fulfillment(of: [unregistered], timeout: 3)
let newMode = Task { await controller.setProxyMode(.rule) }
await fulfillment(of: [started], timeout: 3)
try FileManager.default.removeItem(at: runtimeStore.pidFileURL)
let completed = await oldOperation.value
XCTAssertFalse(completed.succeeded)
XCTAssertEqual(runtimeStore.loadDocument(), previous, "Stale stop must not delete the contract")
loader.release.signal()
await newMode.value
XCTAssertEqual(controller.lastDocument?.aclRuntime?.summary, "rule-proxy-default")
XCTAssertEqual(controller.state, .running)
try await withTestTask(
oldOperation, release: { loader.resume() },
body: {
try await requireTestPhase([unregistered], timeout: 3)
let newMode = Task { await controller.setProxyMode(.rule) }
try await withTestTask(
newMode, release: { loader.resume() },
body: {
try await requireTestPhase([started], timeout: 3)
try FileManager.default.removeItem(at: runtimeStore.pidFileURL)
let completed = await oldOperation.value
XCTAssertFalse(completed.succeeded)
XCTAssertEqual(
runtimeStore.loadDocument(), previous, "Stale stop must not delete the contract")
loader.resume()
await newMode.value
XCTAssertEqual(controller.lastDocument?.aclRuntime?.summary, "rule-proxy-default")
XCTAssertEqual(controller.state, .running)
})
})
}
func testSystemProxyToggleDoesNotDiscardPendingRuleModePreparation() async throws {
@@ -140,7 +160,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
let (store, _) = try makeCustomRuleStore()
let started = expectation(description: "background source preparation")
let loader = PausedRuntimeRuleSource(started: started)
defer { loader.release.signal() }
defer { loader.resume() }
let controller = makeRuleEnablementController(
store: store,
settings: ProxySettings(
@@ -148,14 +168,18 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) }))
try await controller.activate(seeded.server)
let modeSwitch = Task { await controller.setProxyMode(.rule) }
await fulfillment(of: [started], timeout: 3)
await controller.setSystemProxyEnabled(true)
loader.release.signal()
await modeSwitch.value
XCTAssertEqual(controller.proxyMode, .rule)
XCTAssertTrue(controller.settings.systemProxyEnabled)
XCTAssertEqual(controller.lastDocument?.aclRuntime?.summary, "rule-proxy-default")
XCTAssertEqual(controller.state, .running)
try await withTestTask(
modeSwitch, release: { loader.resume() },
body: {
try await requireTestPhase([started], timeout: 3)
await controller.setSystemProxyEnabled(true)
loader.resume()
await modeSwitch.value
XCTAssertEqual(controller.proxyMode, .rule)
XCTAssertTrue(controller.settings.systemProxyEnabled)
XCTAssertEqual(controller.lastDocument?.aclRuntime?.summary, "rule-proxy-default")
XCTAssertEqual(controller.state, .running)
})
}
func testOffIntentDuringSharedSourcePreparationCannotBeOverwritten() async throws {
@@ -163,7 +187,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
let (store, _) = try makeCustomRuleStore()
let started = expectation(description: "background source preparation")
let loader = PausedRuntimeRuleSource(started: started)
defer { loader.release.signal() }
defer { loader.resume() }
let snapshots = BuiltinRuleSnapshots(loader: { loader.load($0) })
let settings = ProxySettings(
listen: ActivationFixture.listen, preferredMode: .global, agentEnabled: true)
@@ -171,29 +195,37 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
store: store, settings: settings, proxyMode: .global, ruleSnapshots: snapshots)
try await controller.activate(seeded.server)
let modeSwitch = Task { await controller.setProxyMode(.rule) }
await fulfillment(of: [started], timeout: 3)
let identity = RuleIdentity(action: .proxy, match: .domainExact("kept.example"))
let ruleCommit = Task {
await controller.commitRuleDocument(
CustomRuleDocument(rules: [], disabledIdentities: [identity]))
}
// Persistence happens before the awaited source. Observe the owner, not timing.
while controller.ruleDocuments.current?.disabledIdentities.contains(identity) != true {
await Task.yield()
}
await controller.setAgentEnabled(false)
let registrationsAfterStop = agent.registerCount
loader.release.signal()
await modeSwitch.value
let result = await ruleCommit.value
XCTAssertEqual(result.outcome, .saved)
await controller.ruleApplicationTask?.value
XCTAssertEqual(result.document?.disabledIdentities, [identity])
XCTAssertEqual(try store.loadDocument().disabledIdentities, [identity])
XCTAssertEqual(controller.state, .off)
XCTAssertFalse(controller.settings.agentEnabled)
XCTAssertEqual(agent.registerCount, registrationsAfterStop)
XCTAssertNil(RuntimeFileStore(fileURL: runtime.contract).loadDocument())
try await withTestTask(
modeSwitch, release: { loader.resume() },
body: {
try await requireTestPhase([started], timeout: 3)
let identity = RuleIdentity(action: .proxy, match: .domainExact("kept.example"))
let ruleCommit = Task {
await controller.commitRuleDocument(
CustomRuleDocument(rules: [], disabledIdentities: [identity]))
}
try await withRuleApplicationTask(
ruleCommit, controller: controller, release: loader.resume,
body: {
// Persistence happens before the awaited source. Observe the owner, not timing.
try await requireTestCondition("rule document persisted") {
controller.ruleDocuments.current?.disabledIdentities.contains(identity) == true
}
await controller.setAgentEnabled(false)
let registrationsAfterStop = agent.registerCount
loader.resume()
await modeSwitch.value
let result = await ruleCommit.value
XCTAssertEqual(result.outcome, .saved)
await controller.ruleApplicationTask?.value
XCTAssertEqual(result.document?.disabledIdentities, [identity])
XCTAssertEqual(try store.loadDocument().disabledIdentities, [identity])
XCTAssertEqual(controller.state, .off)
XCTAssertFalse(controller.settings.agentEnabled)
XCTAssertEqual(agent.registerCount, registrationsAfterStop)
XCTAssertNil(RuntimeFileStore(fileURL: runtime.contract).loadDocument())
})
})
}
func testNewModeSupersedesRulePreparationWithoutRestoringOldMode() async throws {
@@ -201,7 +233,7 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
let (store, _) = try makeCustomRuleStore()
let started = expectation(description: "background source preparation")
let loader = PausedRuntimeRuleSource(started: started)
defer { loader.release.signal() }
defer { loader.resume() }
let controller = makeRuleEnablementController(
store: store,
settings: ProxySettings(
@@ -209,14 +241,18 @@ final class RuleEnablementControllerTests: ProxyRuntimeControllerTests {
proxyMode: .global, ruleSnapshots: BuiltinRuleSnapshots(loader: { loader.load($0) }))
try await controller.activate(seeded.server)
let old = Task { await controller.setProxyMode(.rule) }
await fulfillment(of: [started], timeout: 3)
await controller.setProxyMode(.direct)
loader.release.signal()
await old.value
XCTAssertEqual(controller.proxyMode, .direct)
XCTAssertEqual(controller.state, .running)
XCTAssertEqual(
RuntimeFileStore(fileURL: runtime.contract).loadDocument()?.aclRuntime?.summary, "direct")
try await withTestTask(
old, release: { loader.resume() },
body: {
try await requireTestPhase([started], timeout: 3)
await controller.setProxyMode(.direct)
loader.resume()
await old.value
XCTAssertEqual(controller.proxyMode, .direct)
XCTAssertEqual(controller.state, .running)
XCTAssertEqual(
RuntimeFileStore(fileURL: runtime.contract).loadDocument()?.aclRuntime?.summary, "direct")
})
}
func testOffSaveUsesOwnedDocumentAndNewSessionRejectsCorruption() async throws {
@@ -268,7 +304,7 @@ extension RuleEnablementControllerTests {
launchHealthTimeoutSeconds: 0.05,
launchHealthRetryDelay: {
if deploymentIsRunning {
await deploymentGate.wait()
try await deploymentGate.wait()
} else {
try await Task.sleep(for: .milliseconds(1))
}
@@ -281,27 +317,31 @@ extension RuleEnablementControllerTests {
}
let disabledDocument = CustomRuleDocument(rules: [rule], disabledIdentities: [rule.identity])
let pending = Task { await controller.updateRuleDocument(disabledDocument) }
await fulfillment(of: [deploymentWaiting], timeout: 5)
if switchMode {
await controller.setProxyMode(.global)
} else {
await controller.setAgentEnabled(false)
}
let registrations = agent.registerCount
deploymentGate.resume()
let outcome = await pending.value
XCTAssertEqual(outcome, .saved)
await controller.ruleApplicationTask?.value
XCTAssertEqual(try store.loadDocument().disabledIdentities, [rule.identity])
XCTAssertEqual(agent.registerCount, registrations)
if switchMode {
XCTAssertEqual(controller.proxyMode, .global)
XCTAssertEqual(runtimeStore.loadDocument()?.aclRuntime?.summary, "global")
} else {
XCTAssertFalse(controller.settings.agentEnabled)
XCTAssertEqual(controller.state, .off)
XCTAssertNil(runtimeStore.loadDocument())
}
try await withRuleApplicationTask(
pending, controller: controller, release: deploymentGate.resume,
body: {
try await requireTestPhase([deploymentWaiting], timeout: 5)
if switchMode {
await controller.setProxyMode(.global)
} else {
await controller.setAgentEnabled(false)
}
let registrations = agent.registerCount
deploymentGate.resume()
let outcome = await pending.value
XCTAssertEqual(outcome, .saved)
await controller.ruleApplicationTask?.value
XCTAssertEqual(try store.loadDocument().disabledIdentities, [rule.identity])
XCTAssertEqual(agent.registerCount, registrations)
if switchMode {
XCTAssertEqual(controller.proxyMode, .global)
XCTAssertEqual(runtimeStore.loadDocument()?.aclRuntime?.summary, "global")
} else {
XCTAssertFalse(controller.settings.agentEnabled)
XCTAssertEqual(controller.state, .off)
XCTAssertNil(runtimeStore.loadDocument())
}
})
}
private func configureDeploymentReceipts(
@@ -345,43 +385,3 @@ extension RuleEnablementControllerTests {
XCTAssertNotEqual(controller.state, .running)
}
}
private final class PausedRuntimeRuleSource: @unchecked Sendable {
let started: XCTestExpectation
let release = DispatchSemaphore(value: 0)
init(started: XCTestExpectation) { self.started = started }
func load(_ source: RulesSource) -> RuleSnapshot {
if source == .geolocationCN {
started.fulfill()
release.wait()
}
return rulesFixture(source)
}
}
/// Pause one retry/debounce phase; subsequent passes remain asynchronous without wall-clock delay.
@MainActor
private final class RuleRuntimeDelayGate {
private let started: XCTestExpectation
private var didPause = false
private var continuation: CheckedContinuation<Void, Never>?
init(started: XCTestExpectation) { self.started = started }
func wait() async {
guard !didPause else {
await Task.yield()
return
}
didPause = true
await withCheckedContinuation {
continuation = $0
started.fulfill()
}
}
func resume() {
continuation?.resume()
continuation = nil
}
}
@@ -0,0 +1,59 @@
import Testing
import XCTest
@MainActor
struct RuleRuntimeDelayGateTests {
@Test func releaseBeforeWaitIsRemembered() async throws {
let gate = RuleRuntimeDelayGate(started: XCTestExpectation(description: "entered"))
gate.resume()
gate.resume()
try await gate.wait()
}
@Test func missingReleaseTimesOutAndReportsFailure() async {
var reported = false
let gate = RuleRuntimeDelayGate(
started: XCTestExpectation(description: "entered"), onTimeout: { reported = true })
await #expect(throws: TestWaitError.self) {
try await gate.wait(timeout: .milliseconds(20))
}
#expect(reported)
gate.resume()
gate.resume()
}
@Test func cancellationBeforeEntryDoesNotSuspend() async {
let gate = RuleRuntimeDelayGate(started: XCTestExpectation(description: "entered"))
let task = Task {
withUnsafeCurrentTask { $0?.cancel() }
try await gate.wait()
}
await #expect(throws: CancellationError.self) { try await task.value }
}
@Test func cancellationWhileWaitingDrainsTheWait() async throws {
let entered = XCTestExpectation(description: "entered")
let gate = RuleRuntimeDelayGate(started: entered)
let task = Task { try await gate.wait() }
try await requireTestPhase([entered])
task.cancel()
await #expect(throws: CancellationError.self) { try await task.value }
gate.resume()
}
@Test func releaseAndCancellationCanRaceWithoutDoubleResume() async throws {
for _ in 0..<100 {
let entered = XCTestExpectation(description: "entered")
let gate = RuleRuntimeDelayGate(started: entered)
let task = Task { try await gate.wait() }
try await requireTestPhase([entered])
let cancellation = Task.detached { task.cancel() }
gate.resume()
gate.resume()
do { try await task.value } catch {
#expect(error is CancellationError)
}
await cancellation.value
}
}
}
@@ -175,13 +175,17 @@ struct LANListeningWorkflowTests {
draft.isEnabled = true
draft.mode = .allIPv4Interfaces
let submission = Task { await workflow.saveLANListening(draft) }
await probe.waitUntilEntered()
#expect(workflow.isCommitting)
#expect(await workflow.saveProxyExceptions("example.com") == .rejected(.inProgress))
probe.passThrough(.free)
probe.releaseGatedCalls()
#expect(await submission.value == .persisted)
#expect(!workflow.isCommitting)
try await withTestTask(
submission, release: { probe.releaseGatedCalls() },
body: {
try await probe.waitUntilEntered()
#expect(workflow.isCommitting)
#expect(await workflow.saveProxyExceptions("example.com") == .rejected(.inProgress))
probe.passThrough(.free)
probe.releaseGatedCalls()
#expect(await submission.value == .persisted)
#expect(!workflow.isCommitting)
})
}
@Test(arguments: [
@@ -0,0 +1,91 @@
import Foundation
import Testing
import XCTest
enum TestWaitError: Error {
case timedOut(String)
}
/// A phase failure must leave the test body before it can issue another gated command.
@MainActor
func requireTestPhase(_ expectations: [XCTestExpectation], timeout: TimeInterval = 5) async throws {
guard await XCTWaiter.fulfillment(of: expectations, timeout: timeout) == .completed else {
throw TestWaitError.timedOut(expectations.map(\.expectationDescription).joined(separator: ", "))
}
}
@MainActor
func requireTestCondition(
_ description: String, timeout: Duration = .seconds(5), condition: () -> Bool
) async throws {
let deadline = ContinuousClock.now.advanced(by: timeout)
while !condition() {
try Task.checkCancellation()
guard ContinuousClock.now < deadline else { throw TestWaitError.timedOut(description) }
try await Task.sleep(for: .milliseconds(10))
}
}
/// Release fixtures before joining a cancelled task, including on a phase failure.
@MainActor
func withTestTask<Value: Sendable>(
_ task: Task<Value, Never>, release: () -> Void, finish: () async -> Void = {},
body: () async throws -> Void
) async throws {
do {
try await body()
} catch {
release()
task.cancel()
_ = await task.value
await finish()
throw error
}
release()
_ = await task.value
await finish()
}
/// Broadcast release is remembered, including when cleanup runs before a worker enters.
final class TestBlockingGate: @unchecked Sendable {
private let group = DispatchGroup()
private let lock = NSLock()
private var released = false
private let onTimeout: @Sendable () -> Void
init(
_ description: String = "test worker release", file: StaticString = #filePath,
line: UInt = #line, fileID: String = #fileID, column: Int = #column,
onTimeout: (@Sendable () -> Void)? = nil
) {
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(
fileID: fileID, filePath: String(describing: file), line: Int(line), column: column)
self.onTimeout = { Issue.record(message, sourceLocation: location) }
} 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
}
return true
}
func release() {
lock.lock()
let shouldRelease = !released
released = true
lock.unlock()
if shouldRelease { group.leave() }
}
}
@@ -254,14 +254,14 @@ enum ProxyRuntimeFixture {
/// 线程,不占主 actor)。放行后恢复直通。
final class BlockingProbe: EndpointProbing, @unchecked Sendable {
private let lock = NSLock()
private var gate: DispatchSemaphore?
private var gate: TestBlockingGate?
private var started: XCTestExpectation?
func arm(started: XCTestExpectation) {
lock.lock()
defer { lock.unlock() }
precondition(gate == nil, "上一次 arm 尚未 release")
gate = DispatchSemaphore(value: 0)
gate = TestBlockingGate("runtime fixture release")
self.started = started
}
@@ -271,7 +271,7 @@ enum ProxyRuntimeFixture {
self.gate = nil
self.started = nil
lock.unlock()
gate?.signal()
gate?.release()
}
func probe(host: String, port: Int, timeout: TimeInterval) -> EndpointHealthProbe.Outcome {
@@ -281,7 +281,7 @@ enum ProxyRuntimeFixture {
lock.unlock()
if let gate {
started?.fulfill()
gate.wait()
_ = gate.wait()
}
return .reachable
}
@@ -295,7 +295,7 @@ enum ProxyRuntimeFixture {
private let outcomes: [FirewallBlockStatus]
private let blockOnCall: Int
private let started: XCTestExpectation
private let gate = DispatchSemaphore(value: 0)
private let gate = TestBlockingGate("runtime fixture release")
init(outcomes: [FirewallBlockStatus], blockOnCall: Int, started: XCTestExpectation) {
precondition(!outcomes.isEmpty)
@@ -304,7 +304,7 @@ enum ProxyRuntimeFixture {
self.started = started
}
func release() { gate.signal() }
func release() { gate.release() }
func status(for executableURL: URL) -> FirewallBlockStatus {
lock.lock()
@@ -313,7 +313,7 @@ enum ProxyRuntimeFixture {
lock.unlock()
if call == blockOnCall {
started.fulfill()
gate.wait()
_ = gate.wait()
}
return outcomes[min(call, outcomes.count) - 1]
}
@@ -0,0 +1,84 @@
import XCTest
@testable import ShadowsocksX_NG2
/// Pause one retry/debounce phase; subsequent passes remain asynchronous without wall-clock delay.
@MainActor
final class RuleRuntimeDelayGate {
private let started: XCTestExpectation
private let onTimeout: () -> Void
private var didPause = false
private let stream: AsyncStream<Void>
private let continuation: AsyncStream<Void>.Continuation
init(
started: XCTestExpectation,
onTimeout: @escaping () -> Void = {
XCTFail("Timed out waiting for rule runtime delay release")
}
) {
self.started = started
self.onTimeout = onTimeout
(stream, continuation) = AsyncStream<Void>.makeStream()
}
func wait(timeout: Duration = .seconds(10)) async throws {
try Task.checkCancellation()
guard !didPause else {
await Task.yield()
return
}
didPause = true
started.fulfill()
// AsyncStream ends its iterator on cancellation, so the timer race can drain both children.
do {
try await withThrowingTaskGroup(of: Void.self) { group in
group.addTask { [stream] in
for await _ in stream {}
try Task.checkCancellation()
}
group.addTask {
try await Task.sleep(for: timeout)
throw TestWaitError.timedOut("rule runtime delay release")
}
defer { group.cancelAll() }
try await group.next()
}
} catch let error as TestWaitError {
onTimeout()
throw error
}
}
func resume() {
continuation.finish()
}
}
/// Rule application can replace its task handle while an earlier save is finishing.
@MainActor
func withRuleApplicationTask<Value: Sendable>(
_ task: Task<Value, Never>, controller: ProxyRuntimeController, release: () -> Void,
body: () async throws -> Void
) async throws {
try await withTestTask(
task, release: release,
finish: {
controller.ruleApplicationTask?.cancel()
await controller.ruleApplicationTask?.value
}, body: body)
}
final class PausedRuntimeRuleSource: @unchecked Sendable {
let started: XCTestExpectation
private let release = TestBlockingGate("rule source release")
init(started: XCTestExpectation) { self.started = started }
func resume() { release.release() }
func load(_ source: RulesSource) -> RuleSnapshot {
if source == .geolocationCN {
started.fulfill()
_ = release.wait()
}
return rulesFixture(source)
}
}
@@ -191,18 +191,16 @@ final class FakeOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
/// 新一轮之后落盘,验证过期结果不得覆盖。
final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
private let lock = NSLock()
private let gate = DispatchGroup()
private let gate = TestBlockingGate("occupancy probe release")
private var isGated = true
private var enteredCount = 0
private let entry = AsyncGate()
private var released = false
private let entry = XCTestExpectation(description: "occupancy probe entered")
private var gatedAnswer: PortOccupancy
private var passThroughAnswer: PortOccupancy
init(gatedAnswer: PortOccupancy, passThroughAnswer: PortOccupancy) {
self.gatedAnswer = gatedAnswer
self.passThroughAnswer = passThroughAnswer
gate.enter()
}
/// 已进入探测的调用数(按端口标识序,socks 先于 http)。
@@ -212,8 +210,8 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
return enteredCount
}
func waitUntilEntered() async {
await entry.wait()
func waitUntilEntered() async throws {
try await requireTestPhase([entry])
}
/// 之后的调用立即返回 `answer`;已进入且被扣住的调用仍持有各自应答。
@@ -227,24 +225,21 @@ final class GatedOccupancyProbe: PortOccupancyProbing, @unchecked Sendable {
/// 放行全部被扣住的调用(广播门闩)。
func releaseGatedCalls() {
lock.lock()
let shouldRelease = !released
released = true
isGated = false
lock.unlock()
if shouldRelease {
gate.leave()
}
gate.release()
}
func occupancy(for request: PortOccupancyProbeRequest) -> PortOccupancy {
lock.lock()
enteredCount += 1
let firstEntry = enteredCount == 1
let gated = isGated
let answer = gated ? gatedAnswer : passThroughAnswer
lock.unlock()
entry.release()
if gated {
gate.wait()
if firstEntry { entry.fulfill() }
if gated, !gate.wait() {
return .unknown(detail: "Test fixture release timed out")
}
return answer
}
@@ -0,0 +1,71 @@
import Testing
import XCTest
@MainActor
struct BoundedTestWaitTests {
@Test func missingPhaseStopsTheBodyAndDrainsItsTask() async throws {
let entered = XCTestExpectation(description: "worker entered")
let gate = RuleRuntimeDelayGate(started: entered)
var exited = false
var continued = false
let task = Task {
try? await gate.wait()
exited = true
}
try await requireTestPhase([entered])
await #expect(throws: TestWaitError.self) {
try await withTestTask(
task, release: { gate.resume() },
body: {
try await requireTestPhase(
[XCTestExpectation(description: "phase never reached")], timeout: 0.02)
continued = true
})
}
#expect(!continued)
#expect(exited)
}
@Test func missingConditionHasADeadline() async {
await #expect(throws: TestWaitError.self) {
try await requireTestCondition("never true", timeout: .milliseconds(20)) { false }
}
}
@Test func cancelledConditionWaitStops() async {
let task = Task {
withUnsafeCurrentTask { $0?.cancel() }
try await requireTestCondition("never true") { false }
}
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 blockingGateRemembersEarlyAndRepeatedRelease() {
let gate = TestBlockingGate()
gate.release()
gate.release()
#expect(gate.wait(timeout: 0))
}
@Test func blockingGateReleasesEveryWorker() async throws {
let gate = TestBlockingGate()
let entered = XCTestExpectation(description: "workers entered")
entered.expectedFulfillmentCount = 4
let workers = (0..<4).map { _ in
Task.detached {
entered.fulfill()
return gate.wait()
}
}
try await requireTestPhase([entered])
gate.release()
gate.release()
for worker in workers { #expect(await worker.value) }
}
}