import Foundationimport Testing@testable import CodexBarCore private func antigravityBlockingSleep(_ interval: TimeInterval) { Thread.sleep(forTimeInterval: interval)} private final class AntigravityCLICounter: @unchecked Sendable { private let lock = NSLock() private var count = 0 @discardableResult func increment() -> Int { self.lock.lock() self.count += 1 let value = self.count self.lock.unlock() return value } var value: Int { self.lock.lock() let value = self.count self.lock.unlock() return value }} private final class AntigravityCLIPortRecorder: @unchecked Sendable { private let lock = NSLock() private var ports: [[Int]] = [] func append(_ value: [Int]) { self.lock.lock() self.ports.append(value) self.lock.unlock() } func snapshot() -> [[Int]] { self.lock.lock() let value = self.ports self.lock.unlock() return value }} private final class AntigravityCLITimeoutRecorder: @unchecked Sendable { private let lock = NSLock() private var timeouts: [TimeInterval] = [] func append(_ value: TimeInterval) { self.lock.lock() self.timeouts.append(value) self.lock.unlock() } func snapshot() -> [TimeInterval] { self.lock.lock() let value = self.timeouts self.lock.unlock() return value }} private final class AntigravityCLITestClock: @unchecked Sendable { private let lock = NSLock() private var date: Date init(date: Date) { self.date = date } func now() -> Date { self.lock.lock() let value = self.date self.date = self.date.addingTimeInterval(1) self.lock.unlock() return value }} private final class AntigravityCLIOutputSequence: @unchecked Sendable { private let lock = NSLock() private var values: [Data] init(_ values: [Data]) { self.values = values } func next() -> Data { self.lock.lock() let value = self.values.isEmpty ? Data() : self.values.removeFirst() self.lock.unlock() return value }} struct AntigravityCLIHTTPSFetchStrategyTests { @Test func `local strategy falls back to cli HTTPS in cli source mode`() { let strategy = AntigravityStatusFetchStrategy() let context = self.makeFetchContext(sourceMode: .cli) #expect(strategy.shouldFallback(on: AntigravityStatusProbeError.notRunning, context: context)) } @Test func `local strategy falls back to cli HTTPS in auto source mode`() { let strategy = AntigravityStatusFetchStrategy() let context = self.makeFetchContext(sourceMode: .auto) #expect(strategy.shouldFallback(on: AntigravityStatusProbeError.notRunning, context: context)) } @Test func `local strategy does not fallback for unrelated source modes`() { let strategy = AntigravityStatusFetchStrategy() #expect(!strategy.shouldFallback( on: AntigravityStatusProbeError.notRunning, context: self.makeFetchContext(sourceMode: .oauth))) #expect(!strategy.shouldFallback( on: AntigravityStatusProbeError.notRunning, context: self.makeFetchContext(sourceMode: .web))) #expect(!strategy.shouldFallback( on: AntigravityStatusProbeError.notRunning, context: self.makeFetchContext(sourceMode: .api))) } @Test func `strategy pipeline includes cli HTTPS fallback in cli and auto modes`() async { let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let cliStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .cli)) #expect(cliStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", ]) let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .auto)) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", ]) } @Test func `strategy pipeline keeps source mode authoritative with selected token account`() async { let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let accountID = UUID() let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .auto, selectedTokenAccountID: accountID)) let cliStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .cli, selectedTokenAccountID: accountID)) let oauthStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .oauth, selectedTokenAccountID: accountID)) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", "antigravity.oauth", ]) #expect(cliStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", ]) #expect(oauthStrategies.map(\.id) == ["antigravity.oauth"]) } @Test func `auto strategy pipeline includes oauth when credentials are injected`() async { let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext( sourceMode: .auto, env: self.accountEnv(email: "selected@example.com"))) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", "antigravity.oauth", ]) } @Test func `auto strategy pipeline preserves oauth fallback for shared credentials file`() async throws { let root = FileManager.default.temporaryDirectory .appendingPathComponent("antigravity-shared-auto-\(UUID().uuidString)", isDirectory: true) try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) defer { try? FileManager.default.removeItem(at: root) } let store = AntigravityOAuthCredentialsStore( fileURL: AntigravityOAuthCredentialsStore.defaultURL(home: root)) try store.save(AntigravityOAuthCredentials( accessToken: "access", refreshToken: "refresh", expiryDate: Date().addingTimeInterval(3600), email: "legacy@example.com")) let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .auto, env: ["HOME": root.path])) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", "antigravity.oauth", ]) } // MARK: - Selected-account guard @Test func `account guard ignores fetches without a selected account`() throws { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .auto, env: self.accountEnv(email: "selected@example.com")) try AntigravitySelectedAccountGuard.validate(usage, context: context) } @Test func `account guard accepts matching ambient snapshot in auto mode`() throws { let usage = self.makeUsage(accountEmail: "Selected@Example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) try AntigravitySelectedAccountGuard.validate(usage, context: context) } @Test func `account guard rejects mismatched ambient snapshot in auto mode`() { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) #expect(throws: AntigravityStatusProbeError.accountMismatch( expected: "selected@example.com", found: "ambient@example.com")) { try AntigravitySelectedAccountGuard.validate(usage, context: context) } } @Test func `account guard rejects snapshot without an identity email`() { let usage = self.makeUsage(accountEmail: nil) let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) #expect(throws: AntigravityStatusProbeError.accountMismatch( expected: "selected@example.com", found: nil)) { try AntigravitySelectedAccountGuard.validate(usage, context: context) } } @Test func `account guard rejects when selected account email cannot be resolved`() { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID()) #expect(throws: AntigravityStatusProbeError.accountMismatch( expected: nil, found: "ambient@example.com")) { try AntigravitySelectedAccountGuard.validate(usage, context: context) } } @Test func `account guard leaves explicit cli source mode authoritative`() throws { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .cli, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) try AntigravitySelectedAccountGuard.validate(usage, context: context) } @Test func `selected account email resolves from id_token when email field missing`() { let idToken = Self.makeIDToken(email: "jwt@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: nil, idToken: idToken)) #expect(AntigravitySelectedAccountGuard.selectedAccountEmail(context: context) == "jwt@example.com") } @Test func `selected account email prefers id_token over stored email field`() { let idToken = Self.makeIDToken(email: "jwt@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "stored@example.com", idToken: idToken)) #expect(AntigravitySelectedAccountGuard.selectedAccountEmail(context: context) == "jwt@example.com") } @Test func `cli HTTPS resets session only for one-shot CLI runtime`() { // One-shot CLI invocation: reset after fetch. #expect(AntigravityCLIHTTPSFetchStrategy.shouldResetSessionAfterFetch(self.makeFetchContext(runtime: .cli))) // App runtime keeps the warm session. #expect(!AntigravityCLIHTTPSFetchStrategy.shouldResetSessionAfterFetch(self.makeFetchContext(runtime: .app))) // Long-lived CLI host (codexbar serve) keeps the warm session even at .cli runtime. #expect(!AntigravityCLIHTTPSFetchStrategy.shouldResetSessionAfterFetch( self.makeFetchContext(runtime: .cli, persistsCLISessions: true))) } @Test func `cli HTTPS reports public source as cli`() { #expect(AntigravityCLIHTTPSFetchStrategy.sourceLabel == "cli") } @Test func `cli local strategy availability requires binary`() async throws { let binaryURL = FileManager.default.temporaryDirectory .appendingPathComponent("codexbar-antigravity-\(UUID().uuidString)") try Data("#!/bin/sh\n".utf8).write(to: binaryURL) try FileManager.default.setAttributes( [.posixPermissions: 0o755], ofItemAtPath: binaryURL.path) defer { try? FileManager.default.removeItem(at: binaryURL) } let strategy = AntigravityCLIHTTPSFetchStrategy() let context = self.makeFetchContext(env: ["ANTIGRAVITY_CLI_PATH": binaryURL.path]) let isAvailable = await strategy.isAvailable(context) #expect(isAvailable) } @Test func `cli local endpoints remain HTTPS only on macOS`() { #expect( AntigravityStatusProbe.cliEndpoints(ports: [55624]) == [ AntigravityStatusProbe.AntigravityConnectionEndpoint( scheme: "https", port: 55624, csrfToken: "", source: .cliHTTPS), ]) } @Test func `cli HTTPS falls back to command model configs when quota summary and user status fail`() async throws { let endpoints = [ AntigravityStatusProbe.AntigravityConnectionEndpoint( scheme: "https", port: 50080, csrfToken: "", source: .cliHTTPS), ] let attempts = AntigravityCLICounter() let snapshot = try await AntigravityStatusProbe.fetchSnapshot( context: AntigravityStatusProbe.RequestContext( endpoints: endpoints, timeout: 1, deadline: Date().addingTimeInterval(2)), send: { payload, _, _ in let attempt = attempts.increment() if attempt == 1 { #expect(payload.path == "/exa.language_server_pb.LanguageServerService/RetrieveUserQuotaSummary") throw AntigravityStatusProbeError.apiError("quota summary unavailable") } if attempt == 2 { #expect(payload.path == "/exa.language_server_pb.LanguageServerService/GetUserStatus") throw AntigravityStatusProbeError.apiError("user status unavailable") } #expect(payload.path == "/exa.language_server_pb.LanguageServerService/GetCommandModelConfigs") return Data(""" { "clientModelConfigs": [ { "label": "Claude Sonnet", "modelOrAlias": { "model": "claude-sonnet" }, "quotaInfo": { "remainingFraction": 0.5 } } ] } """.utf8) }) #expect(snapshot.modelQuotas.first?.label == "Claude Sonnet") #expect(attempts.value == 3) } @Test func `cli HTTPS waits for user status after ports appear`() async throws { let fetchAttempts = AntigravityCLICounter() let drainAttempts = AntigravityCLICounter() let fetchedPorts = AntigravityCLIPortRecorder() let snapshot = try await AntigravityCLIHTTPSFetchStrategy.waitForSnapshot( pid: 123, deadline: Date().addingTimeInterval(5), dependencies: AntigravityCLIHTTPSFetchStrategy.SnapshotWaitDependencies( pollIntervalNanoseconds: 0, listeningPorts: { _, _ in [50080, 50081] }, drainOutput: { drainAttempts.increment() return Data() }, fetchSnapshot: { ports in fetchedPorts.append(ports) if fetchAttempts.increment() == 1 { throw AntigravityStatusProbeError.apiError("HTTP 500: GetCascadeModelConfigData() is nil") } return AntigravityStatusSnapshot( modelQuotas: [ AntigravityModelQuota( label: "Claude Opus 4.6 (Thinking)", modelId: "claude-opus-4.6-thinking", remainingFraction: 1, resetTime: nil, resetDescription: nil), ], accountEmail: "user@example.com", accountPlan: "Pro", source: .local) })) #expect(snapshot.accountEmail == "user@example.com") #expect(fetchAttempts.value == 2) #expect(fetchedPorts.snapshot() == [[50080, 50081], [50080, 50081]]) #expect(drainAttempts.value == 4) } @Test func `cli HTTPS retries empty quota snapshots until usage is parseable`() async throws { let fetchAttempts = AntigravityCLICounter() let snapshot = try await AntigravityCLIHTTPSFetchStrategy.waitForSnapshot( pid: 123, deadline: Date().addingTimeInterval(5), dependencies: AntigravityCLIHTTPSFetchStrategy.SnapshotWaitDependencies( pollIntervalNanoseconds: 0, listeningPorts: { _, _ in [50080] }, drainOutput: { Data() }, fetchSnapshot: { _ in if fetchAttempts.increment() == 1 { return AntigravityStatusSnapshot( modelQuotas: [], accountEmail: nil, accountPlan: nil, source: .local) } return AntigravityStatusSnapshot( modelQuotas: [ AntigravityModelQuota( label: "Claude Sonnet", modelId: "claude-sonnet", remainingFraction: 0.5, resetTime: nil, resetDescription: nil), ], accountEmail: "user@example.com", accountPlan: "Pro", source: .local) })) #expect(fetchAttempts.value == 2) #expect(snapshot.modelQuotas.first?.modelId == "claude-sonnet") } @Test func `cli HTTPS drains output before ports appear`() async throws { let portPolls = AntigravityCLICounter() let drainAttempts = AntigravityCLICounter() let snapshot = try await AntigravityCLIHTTPSFetchStrategy.waitForSnapshot( pid: 123, deadline: Date().addingTimeInterval(5), dependencies: AntigravityCLIHTTPSFetchStrategy.SnapshotWaitDependencies( pollIntervalNanoseconds: 0, listeningPorts: { _, _ in portPolls.increment() == 1 ? [] : [50080]Lines 1–500 of 903. Load more
import Foundation import Testing @testable import CodexBarCore private func antigravityBlockingSleep(_ interval: TimeInterval) { Thread.sleep(forTimeInterval: interval) } private final class AntigravityCLICounter: @unchecked Sendable { private let lock = NSLock() private var count = 0 @discardableResult func increment() -> Int { self.lock.lock() self.count += 1 let value = self.count self.lock.unlock() return value } var value: Int { self.lock.lock() let value = self.count self.lock.unlock() return value } } private final class AntigravityCLIPortRecorder: @unchecked Sendable { private let lock = NSLock() private var ports: [[Int]] = [] func append(_ value: [Int]) { self.lock.lock() self.ports.append(value) self.lock.unlock() } func snapshot() -> [[Int]] { self.lock.lock() let value = self.ports self.lock.unlock() return value } } private final class AntigravityCLITimeoutRecorder: @unchecked Sendable { private let lock = NSLock() private var timeouts: [TimeInterval] = [] func append(_ value: TimeInterval) { self.lock.lock() self.timeouts.append(value) self.lock.unlock() } func snapshot() -> [TimeInterval] { self.lock.lock() let value = self.timeouts self.lock.unlock() return value } } private final class AntigravityCLITestClock: @unchecked Sendable { private let lock = NSLock() private var date: Date init(date: Date) { self.date = date } func now() -> Date { self.lock.lock() let value = self.date self.date = self.date.addingTimeInterval(1) self.lock.unlock() return value } } private final class AntigravityCLIOutputSequence: @unchecked Sendable { private let lock = NSLock() private var values: [Data] init(_ values: [Data]) { self.values = values } func next() -> Data { self.lock.lock() let value = self.values.isEmpty ? Data() : self.values.removeFirst() self.lock.unlock() return value } } struct AntigravityCLIHTTPSFetchStrategyTests { @Test func `local strategy falls back to cli HTTPS in cli source mode`() { let strategy = AntigravityStatusFetchStrategy() let context = self.makeFetchContext(sourceMode: .cli) #expect(strategy.shouldFallback(on: AntigravityStatusProbeError.notRunning, context: context)) } @Test func `local strategy falls back to cli HTTPS in auto source mode`() { let strategy = AntigravityStatusFetchStrategy() let context = self.makeFetchContext(sourceMode: .auto) #expect(strategy.shouldFallback(on: AntigravityStatusProbeError.notRunning, context: context)) } @Test func `local strategy does not fallback for unrelated source modes`() { let strategy = AntigravityStatusFetchStrategy() #expect(!strategy.shouldFallback( on: AntigravityStatusProbeError.notRunning, context: self.makeFetchContext(sourceMode: .oauth))) #expect(!strategy.shouldFallback( on: AntigravityStatusProbeError.notRunning, context: self.makeFetchContext(sourceMode: .web))) #expect(!strategy.shouldFallback( on: AntigravityStatusProbeError.notRunning, context: self.makeFetchContext(sourceMode: .api))) } @Test func `strategy pipeline includes cli HTTPS fallback in cli and auto modes`() async { let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let cliStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .cli)) #expect(cliStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", ]) let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .auto)) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", ]) } @Test func `strategy pipeline keeps source mode authoritative with selected token account`() async { let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let accountID = UUID() let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .auto, selectedTokenAccountID: accountID)) let cliStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .cli, selectedTokenAccountID: accountID)) let oauthStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .oauth, selectedTokenAccountID: accountID)) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", "antigravity.oauth", ]) #expect(cliStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", ]) #expect(oauthStrategies.map(\.id) == ["antigravity.oauth"]) } @Test func `auto strategy pipeline includes oauth when credentials are injected`() async { let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext( sourceMode: .auto, env: self.accountEnv(email: "selected@example.com"))) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", "antigravity.oauth", ]) } @Test func `auto strategy pipeline preserves oauth fallback for shared credentials file`() async throws { let root = FileManager.default.temporaryDirectory .appendingPathComponent("antigravity-shared-auto-\(UUID().uuidString)", isDirectory: true) try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) defer { try? FileManager.default.removeItem(at: root) } let store = AntigravityOAuthCredentialsStore( fileURL: AntigravityOAuthCredentialsStore.defaultURL(home: root)) try store.save(AntigravityOAuthCredentials( accessToken: "access", refreshToken: "refresh", expiryDate: Date().addingTimeInterval(3600), email: "legacy@example.com")) let descriptor = ProviderDescriptorRegistry.descriptor(for: .antigravity) let autoStrategies = await descriptor.fetchPlan.pipeline.resolveStrategies( self.makeFetchContext(sourceMode: .auto, env: ["HOME": root.path])) #expect(autoStrategies.map(\.id) == [ "antigravity.app-local", "antigravity.cli-https", "antigravity.ide-local", "antigravity.oauth", ]) } // MARK: - Selected-account guard @Test func `account guard ignores fetches without a selected account`() throws { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .auto, env: self.accountEnv(email: "selected@example.com")) try AntigravitySelectedAccountGuard.validate(usage, context: context) } @Test func `account guard accepts matching ambient snapshot in auto mode`() throws { let usage = self.makeUsage(accountEmail: "Selected@Example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) try AntigravitySelectedAccountGuard.validate(usage, context: context) } @Test func `account guard rejects mismatched ambient snapshot in auto mode`() { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) #expect(throws: AntigravityStatusProbeError.accountMismatch( expected: "selected@example.com", found: "ambient@example.com")) { try AntigravitySelectedAccountGuard.validate(usage, context: context) } } @Test func `account guard rejects snapshot without an identity email`() { let usage = self.makeUsage(accountEmail: nil) let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) #expect(throws: AntigravityStatusProbeError.accountMismatch( expected: "selected@example.com", found: nil)) { try AntigravitySelectedAccountGuard.validate(usage, context: context) } } @Test func `account guard rejects when selected account email cannot be resolved`() { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID()) #expect(throws: AntigravityStatusProbeError.accountMismatch( expected: nil, found: "ambient@example.com")) { try AntigravitySelectedAccountGuard.validate(usage, context: context) } } @Test func `account guard leaves explicit cli source mode authoritative`() throws { let usage = self.makeUsage(accountEmail: "ambient@example.com") let context = self.makeFetchContext( sourceMode: .cli, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "selected@example.com")) try AntigravitySelectedAccountGuard.validate(usage, context: context) } @Test func `selected account email resolves from id_token when email field missing`() { let idToken = Self.makeIDToken(email: "jwt@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: nil, idToken: idToken)) #expect(AntigravitySelectedAccountGuard.selectedAccountEmail(context: context) == "jwt@example.com") } @Test func `selected account email prefers id_token over stored email field`() { let idToken = Self.makeIDToken(email: "jwt@example.com") let context = self.makeFetchContext( sourceMode: .auto, selectedTokenAccountID: UUID(), env: self.accountEnv(email: "stored@example.com", idToken: idToken)) #expect(AntigravitySelectedAccountGuard.selectedAccountEmail(context: context) == "jwt@example.com") } @Test func `cli HTTPS resets session only for one-shot CLI runtime`() { // One-shot CLI invocation: reset after fetch. #expect(AntigravityCLIHTTPSFetchStrategy.shouldResetSessionAfterFetch(self.makeFetchContext(runtime: .cli))) // App runtime keeps the warm session. #expect(!AntigravityCLIHTTPSFetchStrategy.shouldResetSessionAfterFetch(self.makeFetchContext(runtime: .app))) // Long-lived CLI host (codexbar serve) keeps the warm session even at .cli runtime. #expect(!AntigravityCLIHTTPSFetchStrategy.shouldResetSessionAfterFetch( self.makeFetchContext(runtime: .cli, persistsCLISessions: true))) } @Test func `cli HTTPS reports public source as cli`() { #expect(AntigravityCLIHTTPSFetchStrategy.sourceLabel == "cli") } @Test func `cli local strategy availability requires binary`() async throws { let binaryURL = FileManager.default.temporaryDirectory .appendingPathComponent("codexbar-antigravity-\(UUID().uuidString)") try Data("#!/bin/sh\n".utf8).write(to: binaryURL) try FileManager.default.setAttributes( [.posixPermissions: 0o755], ofItemAtPath: binaryURL.path) defer { try? FileManager.default.removeItem(at: binaryURL) } let strategy = AntigravityCLIHTTPSFetchStrategy() let context = self.makeFetchContext(env: ["ANTIGRAVITY_CLI_PATH": binaryURL.path]) let isAvailable = await strategy.isAvailable(context) #expect(isAvailable) } @Test func `cli local endpoints remain HTTPS only on macOS`() { #expect( AntigravityStatusProbe.cliEndpoints(ports: [55624]) == [ AntigravityStatusProbe.AntigravityConnectionEndpoint( scheme: "https", port: 55624, csrfToken: "", source: .cliHTTPS), ]) } @Test func `cli HTTPS falls back to command model configs when quota summary and user status fail`() async throws { let endpoints = [ AntigravityStatusProbe.AntigravityConnectionEndpoint( scheme: "https", port: 50080, csrfToken: "", source: .cliHTTPS), ] let attempts = AntigravityCLICounter() let snapshot = try await AntigravityStatusProbe.fetchSnapshot( context: AntigravityStatusProbe.RequestContext( endpoints: endpoints, timeout: 1, deadline: Date().addingTimeInterval(2)), send: { payload, _, _ in let attempt = attempts.increment() if attempt == 1 { #expect(payload.path == "/exa.language_server_pb.LanguageServerService/RetrieveUserQuotaSummary") throw AntigravityStatusProbeError.apiError("quota summary unavailable") } if attempt == 2 { #expect(payload.path == "/exa.language_server_pb.LanguageServerService/GetUserStatus") throw AntigravityStatusProbeError.apiError("user status unavailable") } #expect(payload.path == "/exa.language_server_pb.LanguageServerService/GetCommandModelConfigs") return Data(""" { "clientModelConfigs": [ { "label": "Claude Sonnet", "modelOrAlias": { "model": "claude-sonnet" }, "quotaInfo": { "remainingFraction": 0.5 } } ] } """.utf8) }) #expect(snapshot.modelQuotas.first?.label == "Claude Sonnet") #expect(attempts.value == 3) } @Test func `cli HTTPS waits for user status after ports appear`() async throws { let fetchAttempts = AntigravityCLICounter() let drainAttempts = AntigravityCLICounter() let fetchedPorts = AntigravityCLIPortRecorder() let snapshot = try await AntigravityCLIHTTPSFetchStrategy.waitForSnapshot( pid: 123, deadline: Date().addingTimeInterval(5), dependencies: AntigravityCLIHTTPSFetchStrategy.SnapshotWaitDependencies( pollIntervalNanoseconds: 0, listeningPorts: { _, _ in [50080, 50081] }, drainOutput: { drainAttempts.increment() return Data() }, fetchSnapshot: { ports in fetchedPorts.append(ports) if fetchAttempts.increment() == 1 { throw AntigravityStatusProbeError.apiError("HTTP 500: GetCascadeModelConfigData() is nil") } return AntigravityStatusSnapshot( modelQuotas: [ AntigravityModelQuota( label: "Claude Opus 4.6 (Thinking)", modelId: "claude-opus-4.6-thinking", remainingFraction: 1, resetTime: nil, resetDescription: nil), ], accountEmail: "user@example.com", accountPlan: "Pro", source: .local) })) #expect(snapshot.accountEmail == "user@example.com") #expect(fetchAttempts.value == 2) #expect(fetchedPorts.snapshot() == [[50080, 50081], [50080, 50081]]) #expect(drainAttempts.value == 4) } @Test func `cli HTTPS retries empty quota snapshots until usage is parseable`() async throws { let fetchAttempts = AntigravityCLICounter() let snapshot = try await AntigravityCLIHTTPSFetchStrategy.waitForSnapshot( pid: 123, deadline: Date().addingTimeInterval(5), dependencies: AntigravityCLIHTTPSFetchStrategy.SnapshotWaitDependencies( pollIntervalNanoseconds: 0, listeningPorts: { _, _ in [50080] }, drainOutput: { Data() }, fetchSnapshot: { _ in if fetchAttempts.increment() == 1 { return AntigravityStatusSnapshot( modelQuotas: [], accountEmail: nil, accountPlan: nil, source: .local) } return AntigravityStatusSnapshot( modelQuotas: [ AntigravityModelQuota( label: "Claude Sonnet", modelId: "claude-sonnet", remainingFraction: 0.5, resetTime: nil, resetDescription: nil), ], accountEmail: "user@example.com", accountPlan: "Pro", source: .local) })) #expect(fetchAttempts.value == 2) #expect(snapshot.modelQuotas.first?.modelId == "claude-sonnet") } @Test func `cli HTTPS drains output before ports appear`() async throws { let portPolls = AntigravityCLICounter() let drainAttempts = AntigravityCLICounter() let snapshot = try await AntigravityCLIHTTPSFetchStrategy.waitForSnapshot( pid: 123, deadline: Date().addingTimeInterval(5), dependencies: AntigravityCLIHTTPSFetchStrategy.SnapshotWaitDependencies( pollIntervalNanoseconds: 0, listeningPorts: { _, _ in portPolls.increment() == 1 ? [] : [50080]