Buckets:

localhost1978
/

localhost

localhost1978/localhost / LOCALHOST1978 /Modules /SSK_Orchestrator /Tests /SSK_OrchestratorTests /AudioIngestionRuntimeTests.swift

localhost1978

3 months ago

download

raw

10.2 kB

import Foundation
import SSK_Abstractions
@testable import SSK_Orchestrator
import UniformTypeIdentifiers
import XCTest

@available(macOS 27.0, *)
final class AudioIngestionRuntimeTests: XCTestCase {
    func testVaultFailureIsFailClosed() async throws {
        let store = RuntimeJobStore(vaultAvailable: false)
        let runner = RuntimePipelineRunner(mode: .success)
        let service = AudioIngestionRuntimeService(
            jobStore: store,
            pipelineRunner: runner,
            sourceResolver: makeResolver(fileURL: makeAudioFile())
        )

        do {
            _ = try await service.startAudioIngestion(makeRequest())
            XCTFail("The request must fail without a mounted vault.")
        } catch let failure as AudioIngestionFailure {
            XCTAssertEqual(failure.code, .vaultUnavailable)
        }
        let invocationCount = await runner.invocationCount()
        XCTAssertEqual(invocationCount, 0)
    }

    func testCompletedJobPersistsAcrossServiceRestartAndBecomesCacheHit() async throws {
        let store = RuntimeJobStore(vaultAvailable: true)
        let runner = RuntimePipelineRunner(mode: .success)
        let audioURL = makeAudioFile()
        let request = makeRequest()
        let firstService = AudioIngestionRuntimeService(
            jobStore: store,
            pipelineRunner: runner,
            sourceResolver: makeResolver(fileURL: audioURL)
        )

        let accepted = try await firstService.startAudioIngestion(request)
        let completed = try await waitForState(
            .completed,
            jobID: accepted.record.id,
            service: firstService
        )
        XCTAssertNotNil(completed.manifest)
        let firstInvocationCount = await runner.invocationCount()
        XCTAssertEqual(firstInvocationCount, 1)

        let restartedService = AudioIngestionRuntimeService(
            jobStore: store,
            pipelineRunner: runner,
            sourceResolver: makeResolver(fileURL: audioURL)
        )
        let restored = try await restartedService.audioIngestionSnapshot(
            jobID: accepted.record.id
        )
        XCTAssertEqual(restored.record.state, .completed)

        let cached = try await restartedService.startAudioIngestion(request)
        XCTAssertTrue(cached.record.cacheHit)
        XCTAssertTrue(cached.progress.cacheHit)
        XCTAssertTrue(cached.widget.cacheHit)
        let cachedInvocationCount = await runner.invocationCount()
        XCTAssertEqual(cachedInvocationCount, 1)
    }

    func testDuplicateActiveJobIsRejected() async throws {
        let store = RuntimeJobStore(vaultAvailable: true)
        let runner = RuntimePipelineRunner(mode: .suspended)
        let audioURL = makeAudioFile()
        let service = AudioIngestionRuntimeService(
            jobStore: store,
            pipelineRunner: runner,
            sourceResolver: makeResolver(fileURL: audioURL)
        )
        let request = makeRequest()
        let accepted = try await service.startAudioIngestion(request)

        do {
            _ = try await service.startAudioIngestion(request)
            XCTFail("An active duplicate must be rejected.")
        } catch let failure as AudioIngestionFailure {
            XCTAssertEqual(failure.code, .duplicateRequest)
        }
        _ = try await service.cancelAudioIngestion(jobID: accepted.record.id)
    }

    func testCancellationPersistsCancelledState() async throws {
        let store = RuntimeJobStore(vaultAvailable: true)
        let runner = RuntimePipelineRunner(mode: .suspended)
        let service = AudioIngestionRuntimeService(
            jobStore: store,
            pipelineRunner: runner,
            sourceResolver: makeResolver(fileURL: makeAudioFile())
        )
        let accepted = try await service.startAudioIngestion(makeRequest())
        let cancelled = try await service.cancelAudioIngestion(jobID: accepted.record.id)

        XCTAssertEqual(cancelled.record.state, .cancelled)
        XCTAssertEqual(cancelled.record.failure?.code, .cancelled)
        let stored = await store.loadAudioJobRecord(id: accepted.record.id)
        XCTAssertEqual(stored?.state, .cancelled)
    }

    private func waitForState(
        _ state: AudioIngestionState,
        jobID: AudioIngestionJobID,
        service: AudioIngestionRuntimeService
    ) async throws -> AudioIngestionSnapshot {
        for _ in 0..<100 {
            let snapshot = try await service.audioIngestionSnapshot(jobID: jobID)
            if snapshot.record.state == state { return snapshot }
            try await Task.sleep(for: .milliseconds(10))
        }
        XCTFail("The audio job did not reach \(state.rawValue).")
        return try await service.audioIngestionSnapshot(jobID: jobID)
    }

    private func makeAudioFile() -> URL {
        let url = FileManager.default.temporaryDirectory
            .appendingPathComponent(UUID().uuidString)
            .appendingPathExtension("wav")
        try! Data("deterministic-audio-fixture".utf8).write(to: url)
        return url
    }

    private func makeResolver(fileURL: URL) -> AudioIngestionSourceResolver {
        AudioIngestionSourceResolver(
            vaultResolver: ClosureAudioVaultFileResolver { _ in fileURL }
        )
    }

    private func makeRequest() -> AudioIngestionRequest {
        AudioIngestionRequest(
            clientRequestID: UUID(uuidString: "10000000-0000-0000-0000-000000000001")!,
            source: .vaultFile(
                uri: SSKURI(
                    authority: .cas(hash: SSKURI.ContentHash(rawValue: "fixture")),
                    domain: .media,
                    subdomain: "audio"
                ),
                displayName: "fixture.wav",
                contentTypeIdentifier: UTType.wav.identifier
            ),
            mode: .postProduction
        )
    }
}

@available(macOS 27.0, *)
private actor RuntimeJobStore: AudioIngestionJobPersisting {
    private let vaultAvailable: Bool
    private var records: [AudioIngestionJobID: AudioIngestionJobRecord] = [:]
    private var manifests: [AudioIngestionJobID: AudioArtifactManifest] = [:]
    private var widgets: [AudioIngestionJobID: AudioIngestionWidgetModel] = [:]

    init(vaultAvailable: Bool) {
        self.vaultAvailable = vaultAvailable
    }

    func isVaultAvailable() -> Bool { vaultAvailable }
    func saveAudioJobRecord(_ record: AudioIngestionJobRecord) { records[record.id] = record }
    func loadAudioJobRecord(id: AudioIngestionJobID) -> AudioIngestionJobRecord? { records[id] }
    func saveAudioArtifactManifest(_ manifest: AudioArtifactManifest) { manifests[manifest.jobID] = manifest }
    func loadAudioArtifactManifest(jobID: AudioIngestionJobID) -> AudioArtifactManifest? { manifests[jobID] }
    func saveAudioWidgetModel(_ model: AudioIngestionWidgetModel) { widgets[model.jobID] = model }
    func loadAudioWidgetModel(jobID: AudioIngestionJobID) -> AudioIngestionWidgetModel? { widgets[jobID] }
    func revealableAudioJobURL(jobID: AudioIngestionJobID) -> URL {
        FileManager.default.temporaryDirectory.appendingPathComponent(jobID.description)
    }
    func shareableAudioArchiveURL(jobID: AudioIngestionJobID) -> URL {
        FileManager.default.temporaryDirectory.appendingPathComponent("\(jobID.description).share")
    }
}

@available(macOS 27.0, *)
private actor RuntimePipelineRunner: AudioIngestionPipelineRunning {
    enum Mode { case success, suspended }
    private let mode: Mode
    private var calls = 0

    init(mode: Mode) { self.mode = mode }
    func invocationCount() -> Int { calls }

    func processAudio(
        url: URL,
        contentType: UTType,
        jobID: AudioIngestionJobID,
        sourceHash: String,
        sourceName: String,
        cacheKey: String,
        pipelineVersion: String,
        modelRevision: String,
        outputPolicy: AudioOutputPolicy,
        sharingPolicy: AudioSharingPolicy,
        progressSink: AudioPipelineProgressSink
    ) async throws -> AudioPipelineResult {
        calls += 1
        if mode == .suspended {
            try await Task.sleep(for: .seconds(60))
            throw CancellationError()
        }

        let sessionURI = SSKURI(
            authority: .node(nodeID: .init(uuid: UUID())),
            domain: .identity,
            subdomain: "session"
        )
        await progressSink.emit(AudioIngestionProgress(
            jobID: jobID,
            sessionURI: sessionURI,
            state: .analyzing,
            sourceName: sourceName,
            fractionCompleted: 0.5,
            modelReadiness: .ready,
            message: "Analyzing."
        ))
        await progressSink.emit(AudioIngestionProgress(
            jobID: jobID,
            sessionURI: sessionURI,
            state: .persisting,
            sourceName: sourceName,
            fractionCompleted: 0.8,
            modelReadiness: .ready,
            message: "Persisting."
        ))

        let artifacts = AudioArtifactManifest.requiredReadyArtifactKinds.map { kind in
            AudioArtifactReference(
                kind: kind,
                uri: SSKURI(
                    authority: .cas(hash: .init(rawValue: "hash-\(kind.rawValue)")),
                    domain: .media,
                    subdomain: "audio-artifact"
                ),
                contentHash: "hash-\(kind.rawValue)",
                byteCount: 1
            )
        }
        let manifest = AudioArtifactManifest(
            jobID: jobID,
            sessionURI: sessionURI,
            sourceContentHash: sourceHash,
            pipelineVersion: pipelineVersion,
            modelRevision: modelRevision,
            cacheKey: cacheKey,
            artifacts: artifacts,
            retentionClass: "test"
        )
        let widget = AudioIngestionWidgetModel(
            jobID: jobID,
            title: "Audio analysis",
            sourceName: sourceName,
            state: .ready,
            phaseMessage: "Ready.",
            fractionCompleted: 0.92,
            modelReadiness: .ready,
            availableArtifacts: Set(artifacts.map(\.kind))
        )
        return AudioPipelineResult(manifest: manifest, widget: widget)
    }
}

Xet Storage Details

Size:: 10.2 kB
Xet hash:: 046540e04967aec8cfdd8ead5c4c9e23e9f87bdd4780ca2fcc2137d2f5d23663

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	import Foundation
	import SSK_Abstractions
	@testable import SSK_Orchestrator
	import UniformTypeIdentifiers
	import XCTest

	@available(macOS 27.0, *)
	final class AudioIngestionRuntimeTests: XCTestCase {
	func testVaultFailureIsFailClosed() async throws {
	let store = RuntimeJobStore(vaultAvailable: false)
	let runner = RuntimePipelineRunner(mode: .success)
	let service = AudioIngestionRuntimeService(
	jobStore: store,
	pipelineRunner: runner,
	sourceResolver: makeResolver(fileURL: makeAudioFile())
	)

	do {
	_ = try await service.startAudioIngestion(makeRequest())
	XCTFail("The request must fail without a mounted vault.")
	} catch let failure as AudioIngestionFailure {
	XCTAssertEqual(failure.code, .vaultUnavailable)
	}
	let invocationCount = await runner.invocationCount()
	XCTAssertEqual(invocationCount, 0)
	}

	func testCompletedJobPersistsAcrossServiceRestartAndBecomesCacheHit() async throws {
	let store = RuntimeJobStore(vaultAvailable: true)
	let runner = RuntimePipelineRunner(mode: .success)
	let audioURL = makeAudioFile()
	let request = makeRequest()
	let firstService = AudioIngestionRuntimeService(
	jobStore: store,
	pipelineRunner: runner,
	sourceResolver: makeResolver(fileURL: audioURL)
	)

	let accepted = try await firstService.startAudioIngestion(request)
	let completed = try await waitForState(
	.completed,
	jobID: accepted.record.id,
	service: firstService
	)
	XCTAssertNotNil(completed.manifest)
	let firstInvocationCount = await runner.invocationCount()
	XCTAssertEqual(firstInvocationCount, 1)

	let restartedService = AudioIngestionRuntimeService(
	jobStore: store,
	pipelineRunner: runner,
	sourceResolver: makeResolver(fileURL: audioURL)
	)
	let restored = try await restartedService.audioIngestionSnapshot(
	jobID: accepted.record.id
	)
	XCTAssertEqual(restored.record.state, .completed)

	let cached = try await restartedService.startAudioIngestion(request)
	XCTAssertTrue(cached.record.cacheHit)
	XCTAssertTrue(cached.progress.cacheHit)
	XCTAssertTrue(cached.widget.cacheHit)
	let cachedInvocationCount = await runner.invocationCount()
	XCTAssertEqual(cachedInvocationCount, 1)
	}

	func testDuplicateActiveJobIsRejected() async throws {
	let store = RuntimeJobStore(vaultAvailable: true)
	let runner = RuntimePipelineRunner(mode: .suspended)
	let audioURL = makeAudioFile()
	let service = AudioIngestionRuntimeService(
	jobStore: store,
	pipelineRunner: runner,
	sourceResolver: makeResolver(fileURL: audioURL)
	)
	let request = makeRequest()
	let accepted = try await service.startAudioIngestion(request)

	do {
	_ = try await service.startAudioIngestion(request)
	XCTFail("An active duplicate must be rejected.")
	} catch let failure as AudioIngestionFailure {
	XCTAssertEqual(failure.code, .duplicateRequest)
	}
	_ = try await service.cancelAudioIngestion(jobID: accepted.record.id)
	}

	func testCancellationPersistsCancelledState() async throws {
	let store = RuntimeJobStore(vaultAvailable: true)
	let runner = RuntimePipelineRunner(mode: .suspended)
	let service = AudioIngestionRuntimeService(
	jobStore: store,
	pipelineRunner: runner,
	sourceResolver: makeResolver(fileURL: makeAudioFile())
	)
	let accepted = try await service.startAudioIngestion(makeRequest())
	let cancelled = try await service.cancelAudioIngestion(jobID: accepted.record.id)

	XCTAssertEqual(cancelled.record.state, .cancelled)
	XCTAssertEqual(cancelled.record.failure?.code, .cancelled)
	let stored = await store.loadAudioJobRecord(id: accepted.record.id)
	XCTAssertEqual(stored?.state, .cancelled)
	}

	private func waitForState(
	_ state: AudioIngestionState,
	jobID: AudioIngestionJobID,
	service: AudioIngestionRuntimeService
	) async throws -> AudioIngestionSnapshot {
	for _ in 0..<100 {
	let snapshot = try await service.audioIngestionSnapshot(jobID: jobID)
	if snapshot.record.state == state { return snapshot }
	try await Task.sleep(for: .milliseconds(10))
	}
	XCTFail("The audio job did not reach \(state.rawValue).")
	return try await service.audioIngestionSnapshot(jobID: jobID)
	}

	private func makeAudioFile() -> URL {
	let url = FileManager.default.temporaryDirectory
	.appendingPathComponent(UUID().uuidString)
	.appendingPathExtension("wav")
	try! Data("deterministic-audio-fixture".utf8).write(to: url)
	return url
	}

	private func makeResolver(fileURL: URL) -> AudioIngestionSourceResolver {
	AudioIngestionSourceResolver(
	vaultResolver: ClosureAudioVaultFileResolver { _ in fileURL }
	)
	}

	private func makeRequest() -> AudioIngestionRequest {
	AudioIngestionRequest(
	clientRequestID: UUID(uuidString: "10000000-0000-0000-0000-000000000001")!,
	source: .vaultFile(
	uri: SSKURI(
	authority: .cas(hash: SSKURI.ContentHash(rawValue: "fixture")),
	domain: .media,
	subdomain: "audio"
	),
	displayName: "fixture.wav",
	contentTypeIdentifier: UTType.wav.identifier
	),
	mode: .postProduction
	)
	}
	}

	@available(macOS 27.0, *)
	private actor RuntimeJobStore: AudioIngestionJobPersisting {
	private let vaultAvailable: Bool
	private var records: [AudioIngestionJobID: AudioIngestionJobRecord] = [:]
	private var manifests: [AudioIngestionJobID: AudioArtifactManifest] = [:]
	private var widgets: [AudioIngestionJobID: AudioIngestionWidgetModel] = [:]

	init(vaultAvailable: Bool) {
	self.vaultAvailable = vaultAvailable
	}

	func isVaultAvailable() -> Bool { vaultAvailable }
	func saveAudioJobRecord(_ record: AudioIngestionJobRecord) { records[record.id] = record }
	func loadAudioJobRecord(id: AudioIngestionJobID) -> AudioIngestionJobRecord? { records[id] }
	func saveAudioArtifactManifest(_ manifest: AudioArtifactManifest) { manifests[manifest.jobID] = manifest }
	func loadAudioArtifactManifest(jobID: AudioIngestionJobID) -> AudioArtifactManifest? { manifests[jobID] }
	func saveAudioWidgetModel(_ model: AudioIngestionWidgetModel) { widgets[model.jobID] = model }
	func loadAudioWidgetModel(jobID: AudioIngestionJobID) -> AudioIngestionWidgetModel? { widgets[jobID] }
	func revealableAudioJobURL(jobID: AudioIngestionJobID) -> URL {
	FileManager.default.temporaryDirectory.appendingPathComponent(jobID.description)
	}
	func shareableAudioArchiveURL(jobID: AudioIngestionJobID) -> URL {
	FileManager.default.temporaryDirectory.appendingPathComponent("\(jobID.description).share")
	}
	}

	@available(macOS 27.0, *)
	private actor RuntimePipelineRunner: AudioIngestionPipelineRunning {
	enum Mode { case success, suspended }
	private let mode: Mode
	private var calls = 0

	init(mode: Mode) { self.mode = mode }
	func invocationCount() -> Int { calls }

	func processAudio(
	url: URL,
	contentType: UTType,
	jobID: AudioIngestionJobID,
	sourceHash: String,
	sourceName: String,
	cacheKey: String,
	pipelineVersion: String,
	modelRevision: String,
	outputPolicy: AudioOutputPolicy,
	sharingPolicy: AudioSharingPolicy,
	progressSink: AudioPipelineProgressSink
	) async throws -> AudioPipelineResult {
	calls += 1
	if mode == .suspended {
	try await Task.sleep(for: .seconds(60))
	throw CancellationError()
	}

	let sessionURI = SSKURI(
	authority: .node(nodeID: .init(uuid: UUID())),
	domain: .identity,
	subdomain: "session"
	)
	await progressSink.emit(AudioIngestionProgress(
	jobID: jobID,
	sessionURI: sessionURI,
	state: .analyzing,
	sourceName: sourceName,
	fractionCompleted: 0.5,
	modelReadiness: .ready,
	message: "Analyzing."
	))
	await progressSink.emit(AudioIngestionProgress(
	jobID: jobID,
	sessionURI: sessionURI,
	state: .persisting,
	sourceName: sourceName,
	fractionCompleted: 0.8,
	modelReadiness: .ready,
	message: "Persisting."
	))

	let artifacts = AudioArtifactManifest.requiredReadyArtifactKinds.map { kind in
	AudioArtifactReference(
	kind: kind,
	uri: SSKURI(
	authority: .cas(hash: .init(rawValue: "hash-\(kind.rawValue)")),
	domain: .media,
	subdomain: "audio-artifact"
	),
	contentHash: "hash-\(kind.rawValue)",
	byteCount: 1
	)
	}
	let manifest = AudioArtifactManifest(
	jobID: jobID,
	sessionURI: sessionURI,
	sourceContentHash: sourceHash,
	pipelineVersion: pipelineVersion,
	modelRevision: modelRevision,
	cacheKey: cacheKey,
	artifacts: artifacts,
	retentionClass: "test"
	)
	let widget = AudioIngestionWidgetModel(
	jobID: jobID,
	title: "Audio analysis",
	sourceName: sourceName,
	state: .ready,
	phaseMessage: "Ready.",
	fractionCompleted: 0.92,
	modelReadiness: .ready,
	availableArtifacts: Set(artifacts.map(\.kind))
	)
	return AudioPipelineResult(manifest: manifest, widget: widget)
	}
	}