1045 lines
58 KiB
Swift
1045 lines
58 KiB
Swift
import Foundation
|
|
@preconcurrency import Network
|
|
import Security
|
|
|
|
#if CAST_SESSION_CHECKS
|
|
struct CastTextTrack: Sendable {
|
|
let id: Int
|
|
let url: URL
|
|
let language: String
|
|
let label: String
|
|
}
|
|
struct CastMediaDescription: Sendable {
|
|
let url: URL
|
|
let contentType: String
|
|
let duration: Double
|
|
let live: Bool
|
|
let tracks: [CastTextTrack]
|
|
}
|
|
#endif
|
|
|
|
private func require(_ condition: @autoclosure () -> Bool, _ message: String) {
|
|
if !condition() { fatalError("FAIL: " + message) }
|
|
}
|
|
|
|
@MainActor
|
|
private func until(_ message: String, attempts: Int = 500, _ predicate: () -> Bool) async throws {
|
|
// An I/O watchdog, not a wall-clock date used to decide playback behavior.
|
|
for _ in 0..<attempts {
|
|
if predicate() { return }
|
|
try await Task.sleep(for: .milliseconds(10))
|
|
}
|
|
require(predicate(), message)
|
|
}
|
|
@MainActor
|
|
private final class CastCheckClock {
|
|
private struct Waiter {
|
|
let delay: Duration
|
|
let deadline: Duration
|
|
let continuation: CheckedContinuation<Void, any Error>
|
|
}
|
|
private var now: Duration = .zero
|
|
private var waiters: [UUID: Waiter] = [:]
|
|
var pendingCount: Int { waiters.count }
|
|
func hasDelay(_ delay: Duration) -> Bool { waiters.values.contains { $0.delay == delay } }
|
|
|
|
func sleep(_ delay: Duration) async throws {
|
|
try Task.checkCancellation()
|
|
let id = UUID()
|
|
try await withTaskCancellationHandler {
|
|
try await withCheckedThrowingContinuation { continuation in
|
|
waiters[id] = Waiter(delay: delay, deadline: now + delay, continuation: continuation)
|
|
}
|
|
} onCancel: {
|
|
Task { @MainActor [weak self] in
|
|
self?.waiters.removeValue(forKey: id)?.continuation.resume(throwing: CancellationError())
|
|
}
|
|
}
|
|
try Task.checkCancellation()
|
|
}
|
|
|
|
func advance(_ duration: Duration) {
|
|
now += duration
|
|
let due = waiters.filter { $0.value.deadline <= now }.map(\.key)
|
|
for id in due { waiters.removeValue(forKey: id)?.continuation.resume() }
|
|
}
|
|
}
|
|
|
|
|
|
@MainActor
|
|
private final class Receiver {
|
|
private let listener: NWListener
|
|
private var socket: NWConnection?
|
|
private var stream = CastWire.Stream()
|
|
private var sender = ""
|
|
private let transport = "receiver-app"
|
|
private var media: [String: Any] = [:]
|
|
private var session = 1
|
|
private var position = 0.0
|
|
private var playing = false
|
|
private var activeTracks: [Int] = []
|
|
var commands: [(String, [String: Any])] = []
|
|
var rejectLoad = false
|
|
var rejectLaunch = false
|
|
var dropLaunch = false
|
|
var dropLoad = false
|
|
var dropCommands = false
|
|
var rejectCommand = false
|
|
var holdSeek = false
|
|
var holdVolume = false
|
|
var holdLaunch = false
|
|
var holdLoad = false
|
|
var holdPlayback = false
|
|
private var heldLaunch: Int?
|
|
private var heldLoad: [String: Any]?
|
|
var ignoreHeartbeat = false
|
|
var verifyRelay = false
|
|
var relayVerified = false
|
|
var failure: String?
|
|
private var heldSeek: [String: Any]?
|
|
private var started = false
|
|
private var volume = 1.0
|
|
var muted = false
|
|
var port: UInt16 { listener.port!.rawValue }
|
|
var device: CastDevice { .init(id: "loopback", name: "Loopback TV", host: "127.0.0.1", port: port) }
|
|
|
|
init(identity: SecIdentity) throws {
|
|
let tls = NWProtocolTLS.Options()
|
|
guard let identity = sec_identity_create(identity) else { throw CastError.protocolError }
|
|
sec_protocol_options_set_local_identity(tls.securityProtocolOptions, identity)
|
|
sec_protocol_options_set_min_tls_protocol_version(tls.securityProtocolOptions, .TLSv12)
|
|
let parameters = NWParameters(tls: tls, tcp: NWProtocolTCP.Options())
|
|
parameters.requiredLocalEndpoint = .hostPort(host: "127.0.0.1", port: .any)
|
|
listener = try NWListener(using: parameters)
|
|
}
|
|
|
|
func start() async throws {
|
|
listener.stateUpdateHandler = { [weak self] state in
|
|
Task { @MainActor in
|
|
switch state {
|
|
case .ready: self?.started = true
|
|
case .failed: self?.failure = "TLS listener failed"
|
|
default: break
|
|
}
|
|
}
|
|
}
|
|
listener.newConnectionHandler = { [weak self] socket in
|
|
Task { @MainActor in
|
|
guard let self else { socket.cancel(); return }
|
|
self.socket?.cancel()
|
|
self.socket = socket
|
|
self.stream = CastWire.Stream()
|
|
socket.stateUpdateHandler = { [weak self] state in
|
|
Task { @MainActor in
|
|
if case .ready = state { self?.receive(socket) }
|
|
}
|
|
}
|
|
socket.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
}
|
|
}
|
|
listener.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
try await until("TLS listener starts") { started || failure != nil }
|
|
require(failure == nil, "TLS listener succeeds")
|
|
}
|
|
|
|
func stop() { socket?.cancel(); listener.cancel() }
|
|
func disconnect() { socket?.cancel() }
|
|
|
|
func releaseLaunch() throws {
|
|
guard let requestID = heldLaunch else { return }
|
|
heldLaunch = nil
|
|
try receiverStatus(requestID)
|
|
}
|
|
|
|
func releaseLoad(reject: Bool = false) async throws {
|
|
guard let command = heldLoad else { return }
|
|
heldLoad = nil
|
|
if reject {
|
|
try send(["type": "LOAD_FAILED", "requestId": command["requestId"] as? Int ?? 0], namespace: CastWire.media)
|
|
} else {
|
|
try await load(command)
|
|
}
|
|
}
|
|
|
|
private func load(_ json: [String: Any]) async throws {
|
|
media = json["media"] as! [String: Any]
|
|
session += 1
|
|
playing = json["autoplay"] as? Bool ?? true
|
|
position = json["currentTime"] as? Double ?? 0
|
|
activeTracks = json["activeTrackIds"] as? [Int] ?? []
|
|
if verifyRelay { try await fetchRelay() }
|
|
try status(requestID: json["requestId"] as? Int ?? 0)
|
|
}
|
|
|
|
private func receive(_ socket: NWConnection) {
|
|
socket.receive(minimumIncompleteLength: 1, maximumLength: 16384) { [weak self] data, _, ended, error in
|
|
Task { @MainActor in
|
|
guard let self else { return }
|
|
do {
|
|
if let data {
|
|
for message in try self.stream.append(data) { try await self.handle(message) }
|
|
}
|
|
if !ended, error == nil { self.receive(socket) }
|
|
} catch { self.failure = "Simulated receiver protocol/relay failure: \(error.localizedDescription)" }
|
|
}
|
|
}
|
|
}
|
|
|
|
private func send(_ json: [String: Any], namespace: String, source: String? = nil, fragmented: Bool = false) throws {
|
|
let payload = try JSONSerialization.data(withJSONObject: json)
|
|
let frame = try CastWire.frame(.init(source: source ?? (namespace == CastWire.media ? transport : "receiver-0"),
|
|
destination: sender, namespace: namespace, payload: payload))
|
|
if fragmented {
|
|
// Split both the length prefix and protobuf, exercising genuine receive framing.
|
|
for chunk in [frame.prefix(2), frame.dropFirst(2).prefix(5), frame.dropFirst(7)] {
|
|
socket?.send(content: Data(chunk), completion: .contentProcessed { _ in })
|
|
}
|
|
} else { socket?.send(content: frame, completion: .contentProcessed { _ in }) }
|
|
}
|
|
|
|
func receiverStatus(_ requestID: Int? = nil, appID: String? = "CC1AD845",
|
|
level: Double? = nil, source: String? = nil) throws {
|
|
let apps: [[String: Any]] = appID.map { [["appId": $0, "sessionId": "app-session", "transportId": transport]] } ?? []
|
|
var json: [String: Any] = ["type": "RECEIVER_STATUS",
|
|
"status": ["applications": apps, "volume": ["level": level ?? volume, "muted": muted]]]
|
|
if let requestID { json["requestId"] = requestID }
|
|
try send(json, namespace: CastWire.receiver, source: source, fragmented: true)
|
|
}
|
|
|
|
func receiverPing() throws {
|
|
try send(["type": "PING"], namespace: CastWire.heartbeat)
|
|
}
|
|
|
|
func status(requestID: Int = 0, session override: Int? = nil, content: String? = nil,
|
|
state: String? = nil, idle: String? = nil) throws {
|
|
var media = self.media
|
|
if let content { media["contentId"] = content }
|
|
try send(["type": "MEDIA_STATUS", "requestId": requestID,
|
|
"status": [["mediaSessionId": override ?? session, "media": media,
|
|
"playerState": state ?? (playing ? "PLAYING" : "PAUSED"), "currentTime": position,
|
|
"idleReason": idle ?? "", "supportedMediaCommands": 4099, "activeTrackIds": activeTracks]]],
|
|
namespace: CastWire.media, fragmented: true)
|
|
}
|
|
|
|
func releaseSeek() throws {
|
|
guard let command = heldSeek else { return }
|
|
heldSeek = nil
|
|
position = command["currentTime"] as? Double ?? position
|
|
try status(requestID: command["requestId"] as? Int ?? 0)
|
|
}
|
|
|
|
func invalidFrame() {
|
|
socket?.send(content: Data([0, 1, 0, 1]), completion: .contentProcessed { _ in }) // 65537 > bound
|
|
}
|
|
|
|
func takeOverApplication() throws {
|
|
try send(["type": "RECEIVER_STATUS", "status": ["applications": []]], namespace: CastWire.receiver)
|
|
}
|
|
|
|
private func handle(_ message: CastWire.Message) async throws {
|
|
sender = message.source
|
|
let json = try JSONSerialization.jsonObject(with: message.payload) as! [String: Any]
|
|
let type = json["type"] as! String
|
|
let requestID = json["requestId"] as? Int ?? 0
|
|
commands.append((type, json))
|
|
if message.namespace == CastWire.heartbeat {
|
|
if type == "PING", !ignoreHeartbeat { try send(["type": "PONG"], namespace: CastWire.heartbeat) }
|
|
return
|
|
}
|
|
switch type {
|
|
case "CONNECT": break
|
|
case "CLOSE": break
|
|
case "LAUNCH":
|
|
require(json["appId"] as? String == "CC1AD845", "Default Media Receiver, no registration")
|
|
if holdLaunch { heldLaunch = requestID; return }
|
|
if dropLaunch { return }
|
|
if rejectLaunch { try send(["type": "LAUNCH_ERROR", "requestId": requestID], namespace: CastWire.receiver) }
|
|
else { try receiverStatus(requestID) }
|
|
case "LOAD":
|
|
if holdLoad { heldLoad = json; return }
|
|
if rejectLoad { try send(["type": "LOAD_FAILED", "requestId": requestID], namespace: CastWire.media); return }
|
|
if dropLoad { return }
|
|
try await load(json)
|
|
case "PLAY", "PAUSE", "SEEK", "GET_STATUS", "EDIT_TRACKS_INFO":
|
|
if holdPlayback, type == "PLAY" || type == "PAUSE" { return }
|
|
if dropCommands { return }
|
|
if rejectCommand {
|
|
try send(["type": "INVALID_PLAYER_STATE", "requestId": requestID], namespace: CastWire.media)
|
|
return
|
|
}
|
|
require(json["mediaSessionId"] as? Int == session, "Controls address the acquired media session")
|
|
if type == "PLAY" { playing = true }
|
|
if type == "PAUSE" { playing = false }
|
|
if type == "EDIT_TRACKS_INFO" { activeTracks = json["activeTrackIds"] as? [Int] ?? [] }
|
|
if type == "SEEK" {
|
|
if holdSeek { heldSeek = json; try status(); return }
|
|
position = json["currentTime"] as? Double ?? position
|
|
}
|
|
try status(requestID: requestID)
|
|
case "SET_VOLUME":
|
|
require(message.namespace == CastWire.receiver, "Volume controls device, not stream")
|
|
if rejectCommand {
|
|
try send(["type": "INVALID_REQUEST", "requestId": requestID], namespace: CastWire.receiver)
|
|
return
|
|
}
|
|
volume = (json["volume"] as? [String: Any])?["level"] as? Double ?? volume
|
|
muted = (json["volume"] as? [String: Any])?["muted"] as? Bool ?? muted
|
|
if !holdVolume { try receiverStatus(requestID) }
|
|
case "STOP":
|
|
require(message.namespace == CastWire.media, "Disconnect does not stop another sender's application")
|
|
require(json["mediaSessionId"] as? Int == session, "Disconnect stops only owned media")
|
|
default: failure = "Unexpected command \(type)"
|
|
}
|
|
}
|
|
|
|
private func fetchRelay() async throws {
|
|
let url = URL(string: media["contentId"] as! String)!
|
|
let config = URLSessionConfiguration.ephemeral
|
|
config.timeoutIntervalForRequest = 30
|
|
let session = URLSession(configuration: config)
|
|
defer { session.invalidateAndCancel() }
|
|
func fetch(_ url: URL) async throws -> Data {
|
|
let (data, response) = try await session.data(from: url)
|
|
require((response as? HTTPURLResponse)?.statusCode == 200, "Receiver fetch succeeds")
|
|
require(!data.isEmpty, "Relay returns real bytes")
|
|
return data
|
|
}
|
|
var playlistURL = url
|
|
var playlist = try await fetch(playlistURL)
|
|
var body = String(data: playlist, encoding: .utf8) ?? ""
|
|
require(body.hasPrefix("#EXTM3U"), "LOAD is playable HLS")
|
|
for _ in 0..<3 {
|
|
guard let uri = body.split(separator: "\n").first(where: { !$0.hasPrefix("#") && !$0.isEmpty }),
|
|
let next = URL(string: String(uri).trimmingCharacters(in: .whitespacesAndNewlines), relativeTo: playlistURL)?.absoluteURL else {
|
|
fatalError("FAIL: playlist has a media resource")
|
|
}
|
|
let data = try await fetch(next)
|
|
if let text = String(data: data, encoding: .utf8), text.hasPrefix("#EXTM3U") {
|
|
playlistURL = next; playlist = data; body = text; continue
|
|
}
|
|
require(data.count >= 188 && data.first == 0x47 || data.count >= 12 &&
|
|
["ftyp", "styp", "moof"].contains(String(data: data[4..<8], encoding: .ascii) ?? ""),
|
|
"Relay serves MPEG-TS or fragmented MP4, not placeholder content")
|
|
for track in media["tracks"] as? [[String: Any]] ?? [] {
|
|
if let textURL = (track["trackContentId"] as? String).flatMap(URL.init(string:)) {
|
|
let text = try await fetch(textURL)
|
|
require(String(data: text, encoding: .utf8)?.hasPrefix("WEBVTT") == true, "Receiver subtitles are WebVTT")
|
|
}
|
|
}
|
|
relayVerified = true
|
|
return
|
|
}
|
|
fatalError("FAIL: receiver playlist resolves to real media")
|
|
}
|
|
}
|
|
|
|
@MainActor
|
|
private func testIdentity() throws -> (SecIdentity, URL) {
|
|
let directory = FileManager.default.temporaryDirectory.appendingPathComponent("terebi-cast-check-" + UUID().uuidString)
|
|
try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
|
|
func openssl(_ arguments: [String]) throws {
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: "/usr/bin/openssl")
|
|
process.arguments = arguments
|
|
process.standardOutput = FileHandle.nullDevice
|
|
process.standardError = FileHandle.nullDevice
|
|
try process.run()
|
|
process.waitUntilExit()
|
|
require(process.terminationStatus == 0, "Creates ephemeral self-signed TLS test identity")
|
|
}
|
|
let key = directory.appendingPathComponent("key.pem").path
|
|
let certificate = directory.appendingPathComponent("cert.pem").path
|
|
let bundle = directory.appendingPathComponent("identity.p12")
|
|
try openssl(["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", key,
|
|
"-out", certificate, "-days", "1", "-subj", "/CN=Cast loopback"])
|
|
try openssl(["pkcs12", "-export", "-inkey", key, "-in", certificate, "-out", bundle.path,
|
|
"-passout", "pass:cast-check", "-keypbe", "PBE-SHA1-3DES", "-certpbe", "PBE-SHA1-3DES", "-macalg", "sha1"])
|
|
let data = try Data(contentsOf: bundle)
|
|
var result: CFArray?
|
|
let status = SecPKCS12Import(data as CFData,
|
|
[kSecImportExportPassphrase as String: "cast-check"] as CFDictionary, &result)
|
|
require(status == errSecSuccess, "Imports ephemeral TLS identity")
|
|
let item = (result! as! [[String: Any]])[0]
|
|
return (item[kSecImportItemIdentity as String] as! SecIdentity, directory)
|
|
}
|
|
|
|
@main
|
|
@MainActor
|
|
struct CastSessionCheck {
|
|
static func discovery() async throws {
|
|
let prefix = "terebi-discovery-" + UUID().uuidString
|
|
let firstID = prefix + "-first", secondID = prefix + "-second"
|
|
let discovery = CastDiscovery()
|
|
var listeners: [NWListener] = []
|
|
var accepted: [NWConnection] = []
|
|
defer {
|
|
discovery.stop()
|
|
for listener in listeners { listener.cancel() }
|
|
for connection in accepted { connection.cancel() }
|
|
}
|
|
func advertise(_ suffix: String, id: String, name: String) throws -> NWListener {
|
|
let listener = try NWListener(using: .tcp)
|
|
listener.service = .init(name: prefix + suffix, type: "_googlecast._tcp",
|
|
txtRecord: NWTXTRecord(["id": id, "fn": name]))
|
|
listener.newConnectionHandler = { connection in
|
|
MainActor.assumeIsolated {
|
|
accepted.append(connection)
|
|
connection.start(queue: .main)
|
|
}
|
|
}
|
|
listener.start(queue: .main)
|
|
listeners.append(listener)
|
|
return listener
|
|
}
|
|
let first = try advertise("-a", id: firstID, name: "TV 10")
|
|
let second = try advertise("-b", id: secondID, name: "TV 2")
|
|
let duplicate = try advertise("-c", id: firstID, name: "TV 10")
|
|
let fallback = try advertise("-fallback", id: "", name: "")
|
|
try await until("Bonjour publishers are ready") { listeners.allSatisfy { $0.state == .ready } }
|
|
let duplicatePort = duplicate.port!.rawValue
|
|
discovery.start()
|
|
discovery.start() // Repeated menu requests must not replace a running search.
|
|
func ownDevices() -> [CastDevice] {
|
|
discovery.devices.filter { $0.id == firstID || $0.id == secondID }
|
|
}
|
|
try await until("Bonjour TXT identity/name, deduplication and natural device order", attempts: 1500) {
|
|
ownDevices().map(\.id) == [secondID, firstID] &&
|
|
ownDevices().map(\.name) == ["TV 2", "TV 10"]
|
|
}
|
|
try await until("Empty TXT values fall back to the service name and identity", attempts: 1500) {
|
|
discovery.devices.contains { $0.name == prefix + "-fallback" && $0.port == fallback.port?.rawValue }
|
|
}
|
|
let fallbackID = discovery.devices.first { $0.name == prefix + "-fallback" }!.id
|
|
let device = ownDevices()[0]
|
|
require(device.port == second.port?.rawValue, "Discovery preserves the advertised Cast port")
|
|
let connection = NWConnection(host: NWEndpoint.Host(device.host),
|
|
port: NWEndpoint.Port(rawValue: device.port)!, using: .tcp)
|
|
defer { connection.cancel() }
|
|
connection.start(queue: .main)
|
|
try await until("Resolved host/port connects to the advertised receiver") {
|
|
connection.state == .ready && !accepted.isEmpty
|
|
}
|
|
second.service = .init(name: prefix + "-b", type: "_googlecast._tcp",
|
|
txtRecord: NWTXTRecord(["id": secondID, "fn": "TV 1"]))
|
|
try await until("TXT changes update the device without changing its identity", attempts: 1500) {
|
|
ownDevices().map(\.id) == [secondID, firstID] && ownDevices()[0].name == "TV 1"
|
|
}
|
|
first.cancel()
|
|
try await until("A duplicate advertisement keeps the receiver available", attempts: 1500) {
|
|
ownDevices().first(where: { $0.id == firstID })?.port == duplicatePort
|
|
}
|
|
duplicate.cancel()
|
|
try await until("Removing the last advertisement removes the receiver", attempts: 1500) {
|
|
ownDevices().map(\.id) == [secondID]
|
|
}
|
|
discovery.stop()
|
|
require(discovery.devices.isEmpty, "Stop clears published receivers")
|
|
discovery.start()
|
|
try await until("Restart ignores cancelled browser/resolver callbacks and rediscovers the receiver", attempts: 1500) {
|
|
ownDevices().map(\.id) == [secondID] && ownDevices()[0].name == "TV 1"
|
|
}
|
|
try await until("Service identity without TXT stays stable across restart", attempts: 1500) {
|
|
discovery.devices.contains { $0.id == fallbackID && $0.name == prefix + "-fallback" }
|
|
}
|
|
second.cancel()
|
|
try await until("Receiver removal after restart", attempts: 1500) { ownDevices().isEmpty }
|
|
require(discovery.error == nil, "Successful browsing leaves no permission/network error")
|
|
print("PASS: Bonjour discovery TXT identity/name/update, natural order, duplicate removal, routable host/port, stop/restart")
|
|
}
|
|
|
|
static func codec() throws {
|
|
let payload = Data("{\"type\":\"PING\"}".utf8)
|
|
let message = CastWire.Message(source: "sender-0", destination: "receiver-0",
|
|
namespace: CastWire.heartbeat, payload: payload)
|
|
let frame = try CastWire.frame(message)
|
|
var stream = CastWire.Stream()
|
|
var decoded: [CastWire.Message] = []
|
|
for byte in frame { decoded += try stream.append(Data([byte])) }
|
|
require(decoded.count == 1 && decoded[0].payload == payload, "Byte-fragmented frame round trips")
|
|
let pair = try stream.append(frame + frame)
|
|
require(pair.count == 2, "Coalesced frames decode")
|
|
for malformed in [Data([0, 0, 0, 0]), Data([0, 1, 0, 1]), Data([255, 255, 255, 255])] {
|
|
do { var parser = CastWire.Stream(); _ = try parser.append(malformed); fatalError("FAIL: invalid length accepted") }
|
|
catch CastError.protocolError { }
|
|
}
|
|
for malformed in [Data([8, 128]), Data([8, 0, 18, 127]), Data([15]), Data(repeating: 255, count: 12)] {
|
|
do { _ = try CastWire.decode(malformed); fatalError("FAIL: malformed protobuf accepted") }
|
|
catch CastError.protocolError { }
|
|
}
|
|
do {
|
|
_ = try CastWire.frame(.init(source: "s", destination: "r", namespace: "m",
|
|
payload: Data(repeating: 65, count: CastWire.maximumFrame)))
|
|
fatalError("FAIL: oversized outbound frame accepted")
|
|
} catch CastError.protocolError { }
|
|
}
|
|
|
|
static func media(_ suffix: String = "first") -> CastMediaDescription {
|
|
.init(url: URL(string: "http://127.0.0.1:9123/session-token/\(suffix).m3u8")!,
|
|
contentType: "application/x-mpegURL", duration: 120, live: false,
|
|
tracks: [.init(id: 17, url: URL(string: "http://127.0.0.1:9123/session-token/subtitle.vtt")!,
|
|
language: "en", label: "English")])
|
|
}
|
|
|
|
static func seekCoalescing(identity: SecIdentity) async throws {
|
|
let peer = try Receiver(identity: identity)
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(sleep: { try await clock.sleep($0) })
|
|
var status: CastStatus?
|
|
var errors: [String] = []
|
|
session.onStatus = { status = $0 }
|
|
session.onError = { errors.append($0) }
|
|
try await session.connect(to: peer.device)
|
|
try await session.load(media(), title: "Seek burst", position: 0, playing: false, subtitle: nil)
|
|
let before = peer.commands.count
|
|
for target in 1...64 { require(session.seek(Double(target)), "Seek burst exceeds ordinary queue capacity") }
|
|
require(session.setPlaying(true), "PLAY separates seek runs")
|
|
for target in 65...96 { require(session.seek(Double(target)), "Second seek run accepted") }
|
|
require(session.setPlaying(false), "PAUSE separates seek runs")
|
|
for target in (48...63).reversed() { require(session.seek(Double(target)), "Reverse seek run accepted") }
|
|
require(status?.seekPending == true, "Latest queued seek remains unacknowledged")
|
|
try await until("Final seek and playback acknowledged") {
|
|
status?.position == 48 && status?.seekPending == false &&
|
|
status?.playing == false && status?.playbackPending == false
|
|
}
|
|
let controls = peer.commands.dropFirst(before).filter { ["SEEK", "PLAY", "PAUSE"].contains($0.0) }
|
|
let targets = controls.filter { $0.0 == "SEEK" }.compactMap { $0.1["currentTime"] as? Double }
|
|
require(targets == [64, 96, 48] || targets == [1, 64, 96, 48],
|
|
"Only unsent adjacent seeks are replaced")
|
|
require(Array(controls.suffix(5)).map(\.0) == ["SEEK", "PLAY", "SEEK", "PAUSE", "SEEK"],
|
|
"Playback controls remain seek ordering barriers")
|
|
require(errors.isEmpty && peer.failure == nil, "Superseded requests do not fail or leak queue capacity")
|
|
session.disconnect()
|
|
}
|
|
|
|
static func behavior(identity: SecIdentity) async throws {
|
|
let peer = try Receiver(identity: identity)
|
|
peer.muted = true
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(sleep: { try await clock.sleep($0) })
|
|
var statuses: [CastStatus] = []
|
|
var errors: [String] = []
|
|
var disconnected = false
|
|
session.onStatus = { statuses.append($0) }
|
|
session.onError = { errors.append($0) }
|
|
session.onDisconnect = { _ in disconnected = true }
|
|
try await session.connect(to: peer.device)
|
|
require(peer.commands.first?.0 == "CONNECT", "TLS platform channel connects before launching")
|
|
require(!session.setPlaying(true), "No media control before load acknowledgement")
|
|
try await session.load(media(), title: "Check", position: 9.25, playing: false, subtitle: 0)
|
|
require(statuses.last?.position == 9.25 && statuses.last?.playing == false && statuses.last?.subtitle == 0,
|
|
"LOAD is receiver-authoritative and subtitle IDs map to indices")
|
|
require(statuses.last?.volume == 0, "Pre-existing device mute is reflected in controls")
|
|
require(session.setPlaying(true) && session.setPlaying(false), "Rapid opposite commands accepted")
|
|
try await until("PLAY and PAUSE acknowledged in order") {
|
|
peer.commands.filter { ["PLAY", "PAUSE"].contains($0.0) }.count == 2 && statuses.last?.playing == false
|
|
}
|
|
require(peer.commands.filter { ["PLAY", "PAUSE"].contains($0.0) }.map(\.0) == ["PLAY", "PAUSE"], "Ordered controls")
|
|
let beforeBurst = statuses.count
|
|
require(session.setPlaying(true) && session.setPlaying(false) && session.setPlaying(true),
|
|
"Three rapid intents are accepted")
|
|
try await until("Latest playback intent is acknowledged") {
|
|
statuses.last?.playing == true && statuses.last?.playbackPending == false
|
|
}
|
|
require(statuses.dropFirst(beforeBurst).contains { $0.playing && $0.playbackPending },
|
|
"An older matching PLAY cannot settle the latest playback intent")
|
|
require(session.setPlaying(false), "Pause before seek checks")
|
|
try await until("Pause settles") { statuses.last?.playing == false && statuses.last?.playbackPending == false }
|
|
peer.holdSeek = true
|
|
let beforeSeek = statuses.count
|
|
require(session.seek(24.25), "SEEK accepted")
|
|
try await until("Old status arrives while SEEK pending") {
|
|
peer.commands.contains { $0.0 == "SEEK" } && statuses.count >= beforeSeek + 2
|
|
}
|
|
require(statuses.last?.seekPending == true && statuses.last?.position == 9.25,
|
|
"Old status cannot clear an unacknowledged seek or invent position")
|
|
try await until("Heartbeat is waiting on the controlled clock") { clock.hasDelay(.seconds(5)) }
|
|
clock.advance(.seconds(5))
|
|
try await until("Newer status poll replies before held SEEK") {
|
|
peer.commands.contains { $0.0 == "GET_STATUS" } && statuses.last?.seekPending == true
|
|
}
|
|
try peer.releaseSeek()
|
|
try await until("SEEK request correlation") { statuses.last?.seekPending == false && statuses.last?.position == 24.25 }
|
|
require(session.setVolume(0.3), "Device volume accepted")
|
|
try await until("Device volume acknowledged") { statuses.last?.volume == 0.3 }
|
|
require(session.setSubtitle(nil), "Subtitle off accepted")
|
|
try await until("Subtitle off acknowledged") { statuses.last?.subtitle == nil }
|
|
let oldSession = 2
|
|
try await session.load(media("replacement"), title: "Replacement", position: 3, playing: true, subtitle: nil)
|
|
let statusCount = statuses.count
|
|
try peer.status(session: oldSession, content: media().url.absoluteString)
|
|
try peer.status()
|
|
try await until("Valid status after stale frame provides an ordered receiver barrier") {
|
|
statuses.count >= statusCount + 1
|
|
}
|
|
require(statuses.count == statusCount + 1 && !disconnected, "Only the valid barrier status is published")
|
|
try peer.status(state: "IDLE", idle: "FINISHED")
|
|
try await until("Receiver reports natural end") { statuses.last?.ended == true }
|
|
let beforeReplay = peer.commands.filter { $0.0 == "LOAD" }.count
|
|
require(session.setPlaying(true) && session.setPlaying(false), "Ended replay and rapid pause accepted")
|
|
try await until("Ended replay uses LOAD from zero then honors pause") {
|
|
peer.commands.filter { $0.0 == "LOAD" }.count == beforeReplay + 1 &&
|
|
statuses.last?.position == 0 && statuses.last?.playing == false && statuses.last?.ended == false
|
|
}
|
|
require(peer.commands.last(where: { $0.0 == "LOAD" })?.1["currentTime"] as? Double == 0,
|
|
"Default receiver ended media reloaded, not invalid PLAY")
|
|
require(!session.seek(.nan) && !session.setVolume(2) && !session.setSubtitle(99), "Invalid command input rejected")
|
|
session.disconnect()
|
|
try await until("Owned media STOP arrives") { peer.commands.contains { $0.0 == "STOP" } }
|
|
require(disconnected && errors.isEmpty && peer.failure == nil, "Safe disconnect without spurious cancellation errors")
|
|
}
|
|
|
|
static func launchRegressions(identity: SecIdentity) async throws {
|
|
for mode in ["cold", "asynchronous", "intermediate"] {
|
|
let peer = try Receiver(identity: identity)
|
|
peer.dropLaunch = true
|
|
try await peer.start()
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(sleep: { try await clock.sleep($0) })
|
|
var outcome: String?
|
|
let task = Task {
|
|
do { try await session.connect(to: peer.device); outcome = "connected" }
|
|
catch { outcome = error.localizedDescription }
|
|
}
|
|
try await until("LAUNCH arrives before delayed receiver status") {
|
|
peer.commands.contains { $0.0 == "LAUNCH" } && clock.hasDelay(.seconds(5))
|
|
}
|
|
let launchID = peer.commands.first { $0.0 == "LAUNCH" }!.1["requestId"] as! Int
|
|
if mode == "cold" {
|
|
clock.advance(.seconds(9))
|
|
try await until("Cold launch advances past ordinary request deadline") {
|
|
outcome != nil || peer.commands.contains { $0.0 == "PING" } && clock.hasDelay(.seconds(5))
|
|
}
|
|
require(outcome == nil, "Cold receiver launch survives the ordinary eight-second control deadline")
|
|
}
|
|
let pongCount = peer.commands.filter { $0.0 == "PONG" }.count
|
|
try peer.receiverStatus(appID: "unrelated-app")
|
|
try peer.receiverStatus(source: "unrelated-transport")
|
|
if mode == "intermediate" { try peer.receiverStatus(launchID, appID: nil) }
|
|
try peer.receiverPing()
|
|
try await until("Ordered receiver barrier after unrelated and intermediate launch statuses") {
|
|
outcome != nil || peer.commands.filter { $0.0 == "PONG" }.count == pongCount + 1
|
|
}
|
|
require(outcome == nil, "Launch waits for Default Media Receiver rather than adopting another app or rejecting intermediate status")
|
|
if mode == "cold" {
|
|
clock.advance(.seconds(11))
|
|
try await until("Cold receiver remains pending beyond the TLS connection deadline") {
|
|
outcome != nil || peer.commands.filter { $0.0 == "PING" }.count >= 2 && clock.hasDelay(.seconds(5))
|
|
}
|
|
require(outcome == nil, "Successful TLS connection does not impose a twelve-second receiver launch deadline")
|
|
}
|
|
try peer.receiverStatus() // Real receivers can broadcast readiness without the LAUNCH request ID.
|
|
try await until("Asynchronous receiver launch status completes connection") { outcome != nil }
|
|
require(outcome == "connected", "Default Media Receiver readiness completes launch without the original acknowledgement")
|
|
await task.value
|
|
try peer.receiverStatus(launchID) // The original acknowledgement can arrive after readiness.
|
|
try await session.load(media(), title: "Delayed launch", position: 0, playing: false, subtitle: nil)
|
|
require(peer.failure == nil, "Late launch acknowledgement does not disturb acquired receiver ownership")
|
|
session.disconnect(stopReceiver: false)
|
|
try await until("Delayed launch cancels all clock handles") { clock.pendingCount == 0 }
|
|
peer.stop()
|
|
}
|
|
}
|
|
|
|
static func volumeRegressions(identity: SecIdentity) async throws {
|
|
let peer = try Receiver(identity: identity)
|
|
peer.holdVolume = true
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(sleep: { try await clock.sleep($0) })
|
|
var statuses: [CastStatus] = []
|
|
var errors: [String] = []
|
|
session.onStatus = { statuses.append($0) }
|
|
session.onError = { errors.append($0) }
|
|
try await session.connect(to: peer.device)
|
|
try await session.load(media(), title: "Volume feedback", position: 0, playing: false, subtitle: nil)
|
|
func volumeID(_ index: Int) -> Int {
|
|
peer.commands.filter { $0.0 == "SET_VOLUME" }[index].1["requestId"] as! Int
|
|
}
|
|
func barrier(_ message: String) async throws {
|
|
let pongs = peer.commands.filter { $0.0 == "PONG" }.count
|
|
try peer.status()
|
|
try peer.receiverPing()
|
|
try await until(message) { peer.commands.filter { $0.0 == "PONG" }.count == pongs + 1 }
|
|
}
|
|
require(session.setVolume(0.2) && session.setVolume(0.8), "Rapid slider commands accepted")
|
|
try await until("Both slider commands reach receiver") { peer.commands.filter { $0.0 == "SET_VOLUME" }.count == 2 }
|
|
let beforeBurst = statuses.count
|
|
try peer.receiverStatus(volumeID(0), level: 0.2)
|
|
let launchID = peer.commands.first { $0.0 == "LAUNCH" }!.1["requestId"] as! Int
|
|
try peer.receiverStatus(launchID, level: 0.1)
|
|
try peer.receiverStatus(level: 0.8)
|
|
try peer.receiverStatus(level: 0.45)
|
|
try await barrier("Media status provides barrier after stale volume feedback")
|
|
require(statuses.last?.volume == 0.8 && statuses.dropFirst(beforeBurst).allSatisfy { $0.volume == 0.8 },
|
|
"Older command and unsolicited receiver feedback cannot pull the slider away from latest user intent")
|
|
try peer.receiverStatus(volumeID(1), level: 0.79)
|
|
try await barrier("Latest slider acknowledgement supplies receiver-authoritative volume")
|
|
require(statuses.last?.volume == 0.79, "Latest acknowledgement restores the receiver's actual quantized volume")
|
|
try peer.receiverStatus(volumeID(0), level: 0.2)
|
|
try peer.receiverStatus(launchID, level: 0.1)
|
|
try await barrier("Late duplicate old volume acknowledgement is consumed")
|
|
require(statuses.last?.volume == 0.79, "Old acknowledged request cannot roll volume backward after latest command settles")
|
|
try peer.receiverStatus(level: 0.4)
|
|
try await barrier("External TV volume arrives while idle")
|
|
require(statuses.last?.volume == 0.4, "External TV changes remain authoritative when no latest slider intent is pending")
|
|
require(session.setVolume(0.6) && session.setVolume(0.9), "Second overlapping slider burst accepted")
|
|
try await until("Second slider burst reaches receiver") { peer.commands.filter { $0.0 == "SET_VOLUME" }.count == 4 }
|
|
try peer.receiverStatus(volumeID(3), level: 0.88)
|
|
try peer.receiverStatus(volumeID(2), level: 0.6)
|
|
try peer.receiverStatus(level: 0.5)
|
|
try await barrier("Reverse acknowledgement order and external change processed")
|
|
require(statuses.last?.volume == 0.5, "Settled latest intent allows external updates even after an older overlapping request")
|
|
require(session.setVolume(0.7), "Unanswered volume command accepted")
|
|
try await until("Volume timeout waits on controlled clock") {
|
|
peer.commands.filter { $0.0 == "SET_VOLUME" }.count == 5 && clock.hasDelay(.seconds(8))
|
|
}
|
|
clock.advance(.seconds(8))
|
|
try await until("Volume timeout releases latest user intent") { !errors.isEmpty }
|
|
require(statuses.last?.volume == 0.5, "Timed-out volume intent restores the last authoritative receiver value")
|
|
try peer.receiverStatus(level: 0.7)
|
|
try await barrier("External volume becomes authoritative after timeout")
|
|
require(statuses.last?.volume == 0.7, "Unsolicited TV volume resumes after bounded command failure")
|
|
peer.rejectCommand = true
|
|
let beforeRejection = errors.count
|
|
require(session.setVolume(0.4), "Rejected volume command is sent")
|
|
try await until("Explicit volume rejection releases latest intent") { errors.count > beforeRejection }
|
|
require(statuses.last?.volume == 0.7, "Rejected slider intent restores authoritative receiver volume")
|
|
session.disconnect(stopReceiver: false)
|
|
try await until("Volume regression cancels all clock handles") { clock.pendingCount == 0 }
|
|
}
|
|
|
|
static func connectionDeadline() async throws {
|
|
let listener = try NWListener(using: .tcp, on: .any)
|
|
@MainActor final class ListenerState {
|
|
var accepted: [NWConnection] = []
|
|
var listening = false
|
|
}
|
|
let state = ListenerState()
|
|
listener.stateUpdateHandler = { event in
|
|
Task { @MainActor in if case .ready = event { state.listening = true } }
|
|
}
|
|
listener.newConnectionHandler = { socket in
|
|
Task { @MainActor in
|
|
state.accepted.append(socket)
|
|
// Accept TCP but never answer the client's TLS handshake.
|
|
socket.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
}
|
|
}
|
|
listener.start(queue: DispatchQueue.global(qos: .userInitiated))
|
|
defer { for socket in state.accepted { socket.cancel() }; listener.cancel() }
|
|
try await until("Incomplete TLS listener starts") { state.listening }
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(sleep: { try await clock.sleep($0) })
|
|
let task = Task {
|
|
try await session.connect(to: .init(id: "stall", name: "Stalled TLS", host: "127.0.0.1",
|
|
port: listener.port!.rawValue))
|
|
}
|
|
try await until("Connection handshake waits on frozen clock") {
|
|
!state.accepted.isEmpty && clock.hasDelay(.seconds(12))
|
|
}
|
|
clock.advance(.seconds(12))
|
|
do { try await task.value; fatalError("FAIL: stalled TLS handshake succeeded") }
|
|
catch CastError.timeout { }
|
|
try await until("Connection deadline cancels all timer handles") { clock.pendingCount == 0 }
|
|
}
|
|
|
|
static func failures(identity: SecIdentity) async throws {
|
|
for mode in ["launch", "launch-timeout", "load", "deadline", "command", "rejection", "cancel", "frame",
|
|
"ownership", "media-error", "media-cancel", "media-interrupt", "heartbeat", "queue", "pending-disconnect"] {
|
|
let peer = try Receiver(identity: identity)
|
|
try await peer.start()
|
|
let clock = CastCheckClock()
|
|
let session = CastSession(requestTimeout: .milliseconds(400), loadTimeout: .milliseconds(400),
|
|
heartbeatInterval: mode == "heartbeat" ? .milliseconds(100) : .seconds(5),
|
|
sleep: { try await clock.sleep($0) })
|
|
var disconnected = false
|
|
var errors: [String] = []
|
|
session.onDisconnect = { _ in disconnected = true }
|
|
session.onError = { errors.append($0) }
|
|
if mode == "launch" { peer.rejectLaunch = true }
|
|
if mode == "launch-timeout" { peer.dropLaunch = true }
|
|
if mode == "heartbeat" { peer.ignoreHeartbeat = true }
|
|
if mode == "launch" {
|
|
do { try await session.connect(to: peer.device); fatalError("FAIL: rejected launch succeeded") }
|
|
catch CastError.rejected { }
|
|
} else if mode == "launch-timeout" {
|
|
let task = Task { try await session.connect(to: peer.device) }
|
|
try await until("Receiver launch waits on frozen clock") {
|
|
peer.commands.contains { $0.0 == "LAUNCH" } && clock.hasDelay(.seconds(30))
|
|
}
|
|
clock.advance(.seconds(30))
|
|
do { try await task.value; fatalError("FAIL: unanswered launch succeeded") }
|
|
catch CastError.timeout { require(disconnected, "Launch deadline releases channel") }
|
|
} else {
|
|
try await session.connect(to: peer.device)
|
|
try await until("Handshake deadline handles cancelled") { clock.pendingCount == 1 }
|
|
peer.rejectLoad = mode == "load"
|
|
peer.dropLoad = mode == "deadline" || mode == "cancel"
|
|
if mode == "load" || mode == "deadline" {
|
|
let task = Task { try await session.load(media(), title: "Failure", position: 0, playing: true, subtitle: nil) }
|
|
if mode == "deadline" {
|
|
try await until("LOAD waits on frozen clock") {
|
|
peer.commands.contains { $0.0 == "LOAD" } && clock.hasDelay(.milliseconds(400))
|
|
}
|
|
clock.advance(.milliseconds(400))
|
|
}
|
|
do { try await task.value; fatalError("FAIL: failed load succeeded") }
|
|
catch CastError.rejected { require(mode == "load", "LOAD rejection correlated") }
|
|
catch CastError.timeout { require(mode == "deadline", "LOAD deadline bounded") }
|
|
} else if mode == "cancel" {
|
|
let task = Task { try await session.load(media(), title: "Cancel", position: 0, playing: true, subtitle: nil) }
|
|
try await until("LOAD sent before cancellation") { peer.commands.contains { $0.0 == "LOAD" } }
|
|
task.cancel()
|
|
do { try await task.value; fatalError("FAIL: cancelled LOAD succeeded") }
|
|
catch { require(disconnected, "LOAD cancellation releases channel") }
|
|
} else {
|
|
try await session.load(media(), title: "Check", position: 0, playing: false, subtitle: nil)
|
|
switch mode {
|
|
case "command":
|
|
peer.dropCommands = true
|
|
require(session.seek(12), "Unanswered SEEK sent")
|
|
try await until("SEEK waits on frozen clock") {
|
|
peer.commands.contains { $0.0 == "SEEK" } && clock.hasDelay(.milliseconds(400))
|
|
}
|
|
clock.advance(.milliseconds(400))
|
|
try await until("Command deadline reports recoverable error") { !errors.isEmpty }
|
|
require(!disconnected, "Command timeout is not channel failure")
|
|
case "rejection":
|
|
peer.rejectCommand = true
|
|
require(session.setPlaying(true), "Rejected command sent")
|
|
try await until("Command rejection reports recoverable error") { !errors.isEmpty }
|
|
require(!disconnected, "Command rejection is not channel failure")
|
|
case "pending-disconnect":
|
|
peer.dropCommands = true
|
|
require(session.setPlaying(true) && session.seek(12), "Playback and seek accepted before disconnect")
|
|
try await until("Pending controls reach receiver before disconnect") {
|
|
peer.commands.contains { $0.0 == "PLAY" } && peer.commands.contains { $0.0 == "SEEK" }
|
|
}
|
|
peer.disconnect()
|
|
try await until("Terminal failure is reported through disconnect") { disconnected }
|
|
require(errors.isEmpty, "Terminal failure does not masquerade as recoverable command rejection")
|
|
case "media-error": try peer.status(state: "IDLE", idle: "ERROR")
|
|
case "media-cancel": try peer.status(state: "IDLE", idle: "CANCELLED")
|
|
case "media-interrupt": try peer.status(state: "IDLE", idle: "INTERRUPTED")
|
|
case "frame": peer.invalidFrame()
|
|
case "ownership": try peer.takeOverApplication()
|
|
case "heartbeat":
|
|
try await until("Heartbeat tick is scheduled on frozen clock") { clock.hasDelay(.milliseconds(100)) }
|
|
clock.advance(.milliseconds(100))
|
|
try await until("PING sent and next heartbeat tick scheduled") {
|
|
peer.commands.contains { $0.0 == "PING" } && clock.hasDelay(.milliseconds(100))
|
|
}
|
|
clock.advance(.milliseconds(100))
|
|
case "queue":
|
|
peer.dropCommands = true
|
|
var accepted = 0
|
|
for _ in 0..<40 { if session.setPlaying(true) { accepted += 1 } }
|
|
require(accepted == 32 && !errors.isEmpty, "Pending command resource bound enforced")
|
|
default: break
|
|
}
|
|
if ["frame", "ownership", "media-error", "media-cancel", "media-interrupt", "heartbeat"].contains(mode) {
|
|
try await until("\(mode) causes bounded channel failure") { disconnected }
|
|
}
|
|
}
|
|
}
|
|
session.disconnect(stopReceiver: false)
|
|
try await until("Cancelled timer handles released") { clock.pendingCount == 0 }
|
|
peer.stop()
|
|
}
|
|
}
|
|
|
|
static func playback(identity: SecIdentity, executable: String, file: String) async throws {
|
|
let peer = try Receiver(identity: identity)
|
|
peer.verifyRelay = true
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: executable)
|
|
process.arguments = [file, String(peer.port)]
|
|
try process.run()
|
|
defer { if process.isRunning { process.terminate() } }
|
|
try await until("Integrated playback subprocess completes", attempts: 18000) { !process.isRunning || peer.failure != nil }
|
|
require(peer.failure == nil, peer.failure ?? "Simulated receiver accepts integration flow")
|
|
require(process.terminationStatus == 0, "Player model integration check succeeds")
|
|
try await until("Disconnect STOP reaches receiver") { peer.commands.contains { $0.0 == "STOP" } }
|
|
let load = peer.commands.first { $0.0 == "LOAD" }!.1
|
|
require(load["currentTime"] as? Double == 9.25 && load["autoplay"] as? Bool == false, "Acknowledged paused handoff")
|
|
require(peer.relayVerified, "Receiver fetches real relay HLS and media bytes")
|
|
}
|
|
|
|
static func queuedPlayback(identity: SecIdentity, executable: String, file: String) async throws {
|
|
let peer = try Receiver(identity: identity)
|
|
peer.verifyRelay = true
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let gates = FileManager.default.temporaryDirectory.appendingPathComponent("terebi-cast-queue-" + UUID().uuidString)
|
|
try FileManager.default.createDirectory(at: gates, withIntermediateDirectories: true)
|
|
defer { try? FileManager.default.removeItem(at: gates) }
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: executable)
|
|
process.arguments = [file, String(peer.port), "queue", gates.path]
|
|
try process.run()
|
|
defer { if process.isRunning { process.terminate() } }
|
|
for stage in ["next", "advanced"] {
|
|
try await until("queue consumer reaches \(stage)", attempts: 18000) {
|
|
FileManager.default.fileExists(atPath: gates.appendingPathComponent(stage).path) ||
|
|
!process.isRunning || peer.failure != nil
|
|
}
|
|
require(peer.failure == nil && process.isRunning, peer.failure ?? "queue consumer remains active")
|
|
require(FileManager.default.fileExists(atPath: gates.appendingPathComponent(stage).path),
|
|
"queue consumer acknowledges \(stage)")
|
|
try peer.status(state: "IDLE", idle: "FINISHED")
|
|
}
|
|
try await until("queue consumer completes", attempts: 18000) { !process.isRunning || peer.failure != nil }
|
|
require(peer.failure == nil && !process.isRunning && process.terminationStatus == 0,
|
|
peer.failure ?? "queue navigation and EOF retain receiver")
|
|
require(peer.commands.filter { $0.0 == "LOAD" }.count == 3 && peer.relayVerified,
|
|
"initial, manual-next and automatic-next LOADs fetch real relay media")
|
|
}
|
|
|
|
static func interruptedPlayback(identity: SecIdentity, executable: String, file: String) async throws {
|
|
for scenario in ["cancel-connect", "cancel-load", "reject-load", "replace-load", "close-load",
|
|
"replace-active", "close-active", "shutdown-load", "shutdown-active", "disconnect-active"] {
|
|
let peer = try Receiver(identity: identity)
|
|
let active = scenario.hasSuffix("-active")
|
|
peer.holdLaunch = scenario == "cancel-connect"
|
|
peer.holdLoad = !active && scenario != "cancel-connect"
|
|
peer.verifyRelay = active
|
|
try await peer.start()
|
|
defer { peer.stop() }
|
|
let gates = FileManager.default.temporaryDirectory.appendingPathComponent("terebi-cast-interruption-" + UUID().uuidString)
|
|
try FileManager.default.createDirectory(at: gates, withIntermediateDirectories: true)
|
|
defer { try? FileManager.default.removeItem(at: gates) }
|
|
func exists(_ name: String) -> Bool {
|
|
FileManager.default.fileExists(atPath: gates.appendingPathComponent(name).path)
|
|
}
|
|
func signal(_ name: String) throws {
|
|
try Data().write(to: gates.appendingPathComponent(name), options: .atomic)
|
|
}
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: executable)
|
|
process.arguments = [file, String(peer.port), scenario, gates.path]
|
|
try process.run()
|
|
defer { if process.isRunning { process.terminate() } }
|
|
func awaitStage(_ message: String, _ predicate: () -> Bool) async throws {
|
|
try await until("\(scenario): \(message)", attempts: 18000) {
|
|
predicate() || !process.isRunning || peer.failure != nil
|
|
}
|
|
require(peer.failure == nil, peer.failure ?? "Receiver accepts interrupted playback")
|
|
require(process.isRunning, "\(scenario): child remains alive until fixture stage")
|
|
require(predicate(), "\(scenario): \(message)")
|
|
}
|
|
if active {
|
|
try await awaitStage("remote playback active") { exists("active") }
|
|
require(peer.relayVerified, "\(scenario): active receiver fetched real relay bytes")
|
|
if scenario == "disconnect-active" {
|
|
peer.holdPlayback = true
|
|
peer.holdSeek = true
|
|
}
|
|
} else {
|
|
let command = scenario == "cancel-connect" ? "LAUNCH" : "LOAD"
|
|
try await awaitStage("handoff reaches held \(command)") { peer.commands.contains { $0.0 == command } }
|
|
}
|
|
try signal("held")
|
|
if scenario == "reject-load" {
|
|
try await awaitStage("consumer observes paused pending LOAD") { exists("observed") }
|
|
try await peer.releaseLoad(reject: true)
|
|
} else {
|
|
try await awaitStage("consumer interrupts handoff") { exists("interrupted") }
|
|
if scenario == "disconnect-active" {
|
|
try await awaitStage("latest playback and seek reach receiver") {
|
|
peer.commands.filter { ["PLAY", "PAUSE"].contains($0.0) }.count == 4 &&
|
|
peer.commands.contains { $0.0 == "SEEK" }
|
|
}
|
|
peer.disconnect()
|
|
} else if scenario == "cancel-connect" {
|
|
try peer.releaseLaunch()
|
|
} else if !active {
|
|
// A cancelled LOAD can still produce a late receiver acknowledgement.
|
|
// Skip relay fetching: the sender has intentionally stopped that relay.
|
|
try await peer.releaseLoad()
|
|
} else {
|
|
try peer.status(state: "PLAYING")
|
|
try peer.receiverStatus(level: 0.9)
|
|
}
|
|
}
|
|
try signal("released")
|
|
try await until("\(scenario): subprocess completes", attempts: 18000) {
|
|
!process.isRunning || peer.failure != nil
|
|
}
|
|
require(peer.failure == nil, peer.failure ?? "Receiver accepts interrupted playback")
|
|
require(!process.isRunning && process.terminationStatus == 0, "\(scenario): consumer-visible restoration succeeds")
|
|
}
|
|
}
|
|
|
|
static func realPeer(_ arguments: [String]) async throws {
|
|
guard let port = UInt16(arguments[3]), let url = URL(string: arguments[4]) else {
|
|
fatalError("Usage: cast-session-check --peer HOST PORT MEDIA_URL CONTENT_TYPE")
|
|
}
|
|
let session = CastSession()
|
|
var status: CastStatus?
|
|
var error: String?
|
|
session.onStatus = { status = $0 }
|
|
session.onDisconnect = { error = $0 }
|
|
session.onError = { error = $0 }
|
|
defer { session.disconnect() }
|
|
try await session.connect(to: .init(id: "smoke", name: "Smoke receiver", host: arguments[2], port: port))
|
|
try await session.load(.init(url: url, contentType: arguments[5], duration: 0, live: false, tracks: []),
|
|
title: "Terebi Cast smoke", position: 0, playing: false, subtitle: nil)
|
|
require(status != nil, "Real receiver acknowledges LOAD")
|
|
require(session.setPlaying(true), "Real receiver PLAY accepted")
|
|
try await until("Real receiver plays", attempts: 1500) { status?.playing == true || error != nil }
|
|
require(error == nil && status?.playing == true, "Real receiver acknowledges PLAY")
|
|
require(session.setPlaying(false), "Real receiver PAUSE accepted")
|
|
try await until("Real receiver pauses", attempts: 1500) { status?.playing == false || error != nil }
|
|
require(error == nil && status?.playing == false, "Real receiver acknowledges PAUSE")
|
|
print("PASS: real peer TLS Default Media Receiver LOAD/PLAY/PAUSE (URLs intentionally omitted)")
|
|
}
|
|
|
|
static func main() async throws {
|
|
if CommandLine.arguments.count == 2, CommandLine.arguments[1] == "--discovery" {
|
|
try await discovery()
|
|
return
|
|
}
|
|
if CommandLine.arguments.count == 6, CommandLine.arguments[1] == "--peer" {
|
|
try await realPeer(CommandLine.arguments)
|
|
return
|
|
}
|
|
try await discovery()
|
|
try codec()
|
|
let (identity, directory) = try testIdentity()
|
|
defer { try? FileManager.default.removeItem(at: directory) }
|
|
if CommandLine.arguments.count == 2, CommandLine.arguments[1] == "--launch-regressions" {
|
|
try await launchRegressions(identity: identity)
|
|
print("PASS: cold and asynchronous Cast launch readiness, unrelated app protection, late acknowledgement")
|
|
return
|
|
}
|
|
if CommandLine.arguments.count == 2, CommandLine.arguments[1] == "--volume-regressions" {
|
|
try await volumeRegressions(identity: identity)
|
|
print("PASS: latest slider intent, stale volume feedback, authoritative acknowledgement, external TV changes, timeout")
|
|
return
|
|
}
|
|
if CommandLine.arguments.count == 4, CommandLine.arguments[1] == "--playback" {
|
|
try await playback(identity: identity, executable: CommandLine.arguments[2], file: CommandLine.arguments[3])
|
|
print("PASS: native TLS receiver fetches relay and acknowledges PlayerModel handoff/control/disconnect")
|
|
try await interruptedPlayback(identity: identity, executable: CommandLine.arguments[2], file: CommandLine.arguments[3])
|
|
print("PASS: interrupted Cast connect/LOAD, rejection, replacement/close, latest intent and acknowledged-position recovery")
|
|
try await queuedPlayback(identity: identity, executable: CommandLine.arguments[2], file: CommandLine.arguments[3])
|
|
print("PASS: Cast queue next, automatic advance, final EOF and local restoration")
|
|
} else {
|
|
try await behavior(identity: identity)
|
|
try await seekCoalescing(identity: identity)
|
|
try await launchRegressions(identity: identity)
|
|
try await volumeRegressions(identity: identity)
|
|
try await failures(identity: identity)
|
|
try await connectionDeadline()
|
|
print("PASS: bounded Cast framing, self-signed TLS launch/load/control, correlation, stale ownership, queue, deadlines, heartbeat, cancellation")
|
|
}
|
|
}
|
|
}
|