use super::*;
use FullAudioState::{AwaitingLabels, NeedsAttention, Processing};
use OverallState::{Completed, Confirmed, Failed, Queued, Running};
use kcode_k1_audio_classification::{FragmentStageV1, StageState, StageStatus};
use std::cell::{Cell, RefCell};
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn id(value: u8) -> FragmentId {
FragmentId::from_bytes([value; 12])
}
fn entry(value: u8, start: u64, end: u64) -> ManifestEntry {
ManifestEntry {
fragment_id: id(value),
start_sample_48k: start,
end_sample_48k: end,
}
}
fn pending(value: u8, start: u64, end: u64) -> PendingFragment {
PendingFragment {
start_sample_48k: start,
end_sample_48k: end,
ogg_bytes: vec![value],
}
}
fn stage(stage: FragmentStageV1) -> StageStatus {
StageStatus {
stage,
state: StageState::Pending,
}
}
fn classification_status(state: OverallState, transcript: Option<&str>) -> FragmentStatus {
FragmentStatus {
state,
queue: stage(FragmentStageV1::Queue),
transcript: stage(FragmentStageV1::Transcript),
speaker_labels: stage(FragmentStageV1::SpeakerLabels),
speaker_features: stage(FragmentStageV1::SpeakerFeatures),
structuring: stage(FragmentStageV1::Structuring),
label_confirmation: stage(FragmentStageV1::LabelConfirmation),
attempt_count: 0,
jobs: Vec::new(),
interim_txid: None,
analysis: None,
confirmed_labels: Vec::new(),
final_transcript: transcript.map(str::to_owned),
errors: Vec::new(),
errors_truncated: false,
}
}
fn object_with_data(data: Vec<u8>) -> Object {
Object {
filename: String::new(),
file_type: MANIFEST_FILE_TYPE.to_owned(),
description: String::new(),
data,
}
}
fn fragment_object(data: &[u8]) -> Object {
Object {
filename: String::new(),
file_type: FRAGMENT_FILE_TYPE.to_owned(),
description: String::new(),
data: data.to_vec(),
}
}
fn status_result(statuses: &[FragmentStatus]) -> Result<FullAudioStatus, String> {
let entries = statuses
.iter()
.enumerate()
.map(|(index, _)| {
let value = u8::try_from(index + 1).expect("test fragment ID");
let start = u64::try_from(index).expect("test index") * 80;
entry(value, start, start + 100)
})
.collect::<Vec<_>>();
let object = object_with_data(encode_manifest(&entries).expect("manifest"));
status_orchestration(
id(90),
|_| Ok(Some(object.clone())),
|fragment_id| {
Ok(entries
.iter()
.position(|entry| entry.fragment_id == fragment_id)
.map(|index| statuses[index].clone()))
},
)
}
fn failing_submission(
fragments: Vec<PendingFragment>,
fail_on: u8,
) -> (Result<FullAudioId, String>, usize, usize) {
let submissions = Cell::new(0);
let saves = Cell::new(0);
let result = submit_orchestration(
b"input",
|_| Ok(vec![1]),
|_| Ok(fragments),
|fragment| {
submissions.set(submissions.get() + 1);
if fragment[0] == fail_on {
Err("submission failed".to_owned())
} else {
Ok(id(fragment[0]))
}
},
|_| {
saves.set(saves.get() + 1);
Ok(id(9))
},
);
(result, submissions.get(), saves.get())
}
#[test]
fn submission_contract_is_preserved() {
let events = RefCell::new(Vec::new());
let returned = submit_orchestration(
b"input",
|audio| {
assert_eq!(audio, b"input");
events.borrow_mut().push("convert");
Ok(b"ogg".to_vec())
},
|ogg| {
assert_eq!(ogg, b"ogg");
events.borrow_mut().push("split");
Ok(vec![pending(10, 0, 100), pending(20, 80, 180)])
},
|fragment| {
events.borrow_mut().push(if fragment[0] == 10 {
"submit-10"
} else {
"submit-20"
});
Ok(id(fragment[0] / 10))
},
|manifest| {
events.borrow_mut().push("save");
assert_eq!(
decode_manifest(manifest).expect("saved manifest"),
vec![entry(1, 0, 100), entry(2, 80, 180)]
);
Ok(id(9))
},
)
.expect("submit");
assert_eq!(returned, id(9));
assert_eq!(
events.borrow().as_slice(),
["convert", "split", "submit-10", "submit-20", "save"]
);
let partial = failing_submission(vec![pending(1, 0, 100), pending(2, 80, 180)], 2);
assert!(partial.0.is_err());
assert_eq!((partial.1, partial.2), (2, 0));
let overlong = failing_submission(vec![pending(1, 0, 7_200_001)], 0);
assert_eq!(overlong.0.unwrap_err(), "invalid fragment interval");
assert_eq!((overlong.1, overlong.2), (0, 0));
}
#[test]
fn status_contract_and_identity_text_are_preserved() {
for (first, second, transcript, expected) in [
(Failed, Queued, None, NeedsAttention),
(Completed, Running, None, Processing),
(Completed, Confirmed, Some("ready"), AwaitingLabels),
] {
let status = status_result(&[
classification_status(first, None),
classification_status(second, transcript),
])
.expect("derived state");
assert_eq!((status.state, status.final_transcript), (expected, None));
}
let complete = status_result(&[
classification_status(
Confirmed,
Some("[high] abababababababababababab: one two three four"),
),
classification_status(Confirmed, Some("[medium] Unknown: two three four five")),
])
.expect("complete");
assert_eq!(complete.state, FullAudioState::Complete);
assert_eq!(
complete.final_transcript.as_deref(),
Some("[high] abababababababababababab: one two three four\n[medium] Unknown: five")
);
assert_eq!(complete.fragments[0].fragment_id, id(1));
assert_eq!(complete.fragments[1].start_sample_48k, 80);
for status in [
classification_status(Confirmed, None),
classification_status(Completed, Some("not confirmed")),
] {
assert!(status_result(&[status]).is_err());
}
let entries = [entry(1, 0, 100)];
let valid = object_with_data(encode_manifest(&entries).expect("manifest"));
let mut wrong_type = valid.clone();
wrong_type.file_type = "wrong".to_owned();
let mut wrong_filename = valid.clone();
wrong_filename.filename = "named".to_owned();
for object in [
None,
Some(valid),
Some(wrong_type),
Some(wrong_filename),
Some(object_with_data(vec![1])),
] {
assert!(status_orchestration(id(90), |_| Ok(object.clone()), |_| Ok(None)).is_err());
}
let statuses = [
classification_status(Confirmed, Some("Alpha, beta gamma delta")),
classification_status(Confirmed, Some("BETA GAMMA DELTA epsilon")),
];
let first = status_result(&statuses).expect("first status");
assert_eq!(first, status_result(&statuses).expect("second status"));
assert_eq!(
first.final_transcript.as_deref(),
Some("Alpha, beta gamma delta\nepsilon")
);
}
#[test]
fn fragment_audio_requires_membership_and_exact_object_metadata() {
let manifest_id = id(90);
let fragment_id = id(1);
let manifest = object_with_data(
encode_manifest(&[entry(1, 0, 100), entry(2, 80, 180)]).expect("manifest"),
);
let fragment = fragment_object(b"exact ogg bytes");
let loads = RefCell::new(Vec::new());
let bytes = fragment_audio_orchestration(manifest_id, fragment_id, |object_id| {
loads.borrow_mut().push(object_id);
if object_id == manifest_id {
Ok(Some(manifest.clone()))
} else if object_id == fragment_id {
Ok(Some(fragment.clone()))
} else {
Ok(None)
}
})
.expect("authorized fragment read");
assert_eq!(bytes, b"exact ogg bytes");
assert_eq!(loads.into_inner(), vec![manifest_id, fragment_id]);
let loads = Cell::new(0);
let foreign = fragment_audio_orchestration(manifest_id, id(3), |_| {
loads.set(loads.get() + 1);
Ok(Some(manifest.clone()))
});
assert_eq!(
foreign.unwrap_err(),
"fragment does not belong to full audio"
);
assert_eq!(loads.get(), 1);
let missing = fragment_audio_orchestration(manifest_id, fragment_id, |object_id| {
if object_id == manifest_id {
Ok(Some(manifest.clone()))
} else {
Ok(None)
}
});
assert!(missing.is_err());
let mut wrong_type = fragment.clone();
wrong_type.file_type = "not-audio".to_owned();
let mut named = fragment.clone();
named.filename = "fragment.ogg".to_owned();
let mut described = fragment;
described.description = "description".to_owned();
for invalid in [wrong_type, named, described] {
let result = fragment_audio_orchestration(manifest_id, fragment_id, |object_id| {
if object_id == manifest_id {
Ok(Some(manifest.clone()))
} else {
Ok(Some(invalid.clone()))
}
});
assert_eq!(result.unwrap_err(), "object is not an audio fragment");
}
}
#[test]
fn independent_calls_and_path_validation_are_preserved() {
let (entered_sender, entered_receiver) = mpsc::sync_channel(1);
let (release_sender, release_receiver) = mpsc::channel();
let blocked = thread::spawn(move || {
submit_orchestration(
b"input",
|_| Ok(vec![1]),
|_| Ok(vec![pending(1, 0, 100)]),
|_| {
entered_sender.send(()).expect("signal blocked submit");
release_receiver.recv().expect("release blocked submit");
Ok(id(1))
},
|_| Ok(id(10)),
)
});
entered_receiver
.recv_timeout(Duration::from_secs(2))
.expect("blocked operation entered");
let independent = submit_orchestration(
b"input",
|_| Ok(vec![1]),
|_| Ok(vec![pending(2, 0, 100)]),
|_| Ok(id(2)),
|_| Ok(id(20)),
);
assert_eq!(independent.expect("independent submit"), id(20));
release_sender.send(()).expect("release blocked operation");
assert_eq!(
blocked.join().expect("join").expect("blocked submit"),
id(10)
);
assert_eq!(
validate_ffmpeg_path(Path::new("/definitely/not/present")).expect("absolute path"),
PathBuf::from("/definitely/not/present")
);
assert!(validate_ffmpeg_path(Path::new("relative/ffmpeg")).is_err());
}