stream_processor_test/
test_spec.rs

1// Copyright 2019 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::elementary_stream::*;
6use crate::output_validator::*;
7use crate::stream::*;
8use crate::stream_runner::*;
9use crate::{FatalError, Result};
10use anyhow::Context as _;
11use fidl_fuchsia_media::StreamProcessorProxy;
12use futures::TryStreamExt;
13use futures::future::BoxFuture;
14use futures::stream::FuturesUnordered;
15use std::rc::Rc;
16
17pub enum OutputSize {
18    // Size of output in terms of packets.
19    PacketCount(usize),
20    // Size of output in terms of number of raw bytes.
21    RawBytesCount(usize),
22}
23
24const FIRST_FORMAT_DETAILS_VERSION_ORDINAL: u64 = 1;
25
26pub type TestCaseOutputs = Vec<Output>;
27
28pub trait StreamProcessorFactory {
29    fn connect_to_stream_processor(
30        &self,
31        stream: &dyn ElementaryStream,
32        format_details_version_ordinal: u64,
33    ) -> BoxFuture<'_, Result<StreamProcessorProxy>>;
34}
35
36/// A test spec describes all the cases that will run and the circumstances in which
37/// they will run.
38pub struct TestSpec {
39    pub cases: Vec<TestCase>,
40    pub relation: CaseRelation,
41    pub stream_processor_factory: Rc<dyn StreamProcessorFactory>,
42}
43
44/// A case relation describes the temporal relationship between two test cases.
45pub enum CaseRelation {
46    /// With serial relation, test cases will be run in sequence using the same codec server.
47    /// For serial relation, outputs from test cases will be returned.
48    Serial,
49    /// With concurrent relation, test cases will run concurrently using two or more codec servers.
50    /// For concurrent relation, outputs from test cases will not be returned.
51    Concurrent,
52}
53
54/// A test cases describes a sequence of elementary stream chunks that should be fed into a codec
55/// server, and a set of validators to check the output. To pass, all validations must pass for all
56/// output from the stream.
57pub struct TestCase {
58    pub name: &'static str,
59    pub stream: Rc<dyn ElementaryStream>,
60    pub validators: Vec<Rc<dyn OutputValidator>>,
61    pub stream_options: Option<StreamOptions>,
62}
63
64impl TestSpec {
65    pub async fn run(self) -> Result<Option<Vec<TestCaseOutputs>>> {
66        let res = match self.relation {
67            CaseRelation::Serial => {
68                Some(run_cases_serially(self.stream_processor_factory.as_ref(), self.cases).await?)
69            }
70            CaseRelation::Concurrent => {
71                run_cases_concurrently(self.stream_processor_factory.as_ref(), self.cases).await?;
72                None
73            }
74        };
75        Ok(res)
76    }
77}
78
79async fn run_cases_serially(
80    stream_processor_factory: &dyn StreamProcessorFactory,
81    cases: Vec<TestCase>,
82) -> Result<Vec<TestCaseOutputs>> {
83    let stream_processor =
84        if let Some(stream) = cases.first().as_ref().map(|case| case.stream.as_ref()) {
85            stream_processor_factory
86                .connect_to_stream_processor(stream, FIRST_FORMAT_DETAILS_VERSION_ORDINAL)
87                .await?
88        } else {
89            return Err(FatalError(String::from("No test cases provided.")).into());
90        };
91    let mut stream_runner = StreamRunner::new(stream_processor);
92
93    let mut all_outputs = Vec::new();
94    for case in cases {
95        let output = stream_runner
96            .run_stream(case.stream, case.stream_options.unwrap_or_default())
97            .await
98            .context(format!("Running case {}", case.name))?;
99        for validator in case.validators {
100            validator.validate(&output).await.context(format!("Validating case {}", case.name))?;
101        }
102        all_outputs.push(output);
103    }
104    Ok(all_outputs)
105}
106
107async fn run_cases_concurrently(
108    stream_processor_factory: &dyn StreamProcessorFactory,
109    cases: Vec<TestCase>,
110) -> Result<()> {
111    let mut unordered = FuturesUnordered::new();
112    for case in cases {
113        unordered.push(run_cases_serially(stream_processor_factory, vec![case]))
114    }
115
116    while let Some(_) = unordered.try_next().await? {}
117
118    Ok(())
119}
120
121pub fn with_large_stack(f: fn() -> Result<()>) -> Result<()> {
122    // The TestSpec futures are too big to fit on Fuchsia's default stack.
123    const MEGABYTE: usize = 1024 * 1024;
124    const STACK_SIZE: usize = 4 * MEGABYTE;
125    std::thread::Builder::new().stack_size(STACK_SIZE).spawn(f).unwrap().join().unwrap()
126}
stream_processor_test/test_spec.rs

stream_processor_test/
test_spec.rs
