async fn run_invocations(
    suite: &ftest::SuiteProxy,
    invocations: Vec<Invocation>,
    run_options: fidl_fuchsia_test::RunOptions,
    counter: &AtomicU32,
    sender: &mut mpsc::Sender<Result<FidlSuiteEvent, LaunchError>>,
    timeout_fut: futures::future::Shared<fasync::Timer>,
) -> Result<SuiteStatus, anyhow::Error> {
    let (run_listener_client, mut run_listener) =
        fidl::endpoints::create_request_stream().expect("cannot create request stream");
    suite.run(&mut invocations.into_iter().map(|i| i.into()), run_options, run_listener_client)?;

    let tasks = Arc::new(lock::Mutex::new(vec![]));
    let running_test_cases = Arc::new(lock::Mutex::new(HashSet::new()));
    let tasks_clone = tasks.clone();
    let initial_suite_status: SuiteStatus;
    let mut sender_clone = sender.clone();
    let test_fut = async {
        let mut initial_suite_status = SuiteStatus::DidNotFinish;
        while let Some(result_event) =
            run_listener.try_next().await.context("error waiting for listener")?
        {
            match result_event {
                ftest::RunListenerRequest::OnTestCaseStarted {
                    invocation,
                    primary_log,
                    listener,
                    control_handle: _,
                } => {
                    let name =
                        invocation.name.ok_or(format_err!("cannot find name in invocation"))?;
                    let identifier = counter.fetch_add(1, Ordering::Relaxed);
                    let events = vec![
                        Ok(SuiteEvents::case_found(identifier, name).into()),
                        Ok(SuiteEvents::case_started(identifier).into()),
                        Ok(SuiteEvents::case_stdout(identifier, primary_log).into()),
                    ];
                    for event in events {
                        sender_clone.send(event).await.unwrap();
                    }
                    let listener =
                        listener.into_stream().context("Cannot convert listener to stream")?;
                    running_test_cases.lock().await.insert(identifier);
                    let running_test_cases = running_test_cases.clone();
                    let sender = sender_clone.clone();
                    let task = fasync::Task::spawn(async move {
                        let status = listen_for_completion(listener, identifier, sender).await;
                        running_test_cases.lock().await.remove(&identifier);
                        status
                    });
                    tasks_clone.lock().await.push(task);
                }
                ftest::RunListenerRequest::OnFinished { .. } => {
                    initial_suite_status = SuiteStatus::Passed;
                    break;
                }
            }
        }
        Ok(initial_suite_status)
    }
    .fuse();

    futures::pin_mut!(test_fut);
    let timeout_fut = timeout_fut.fuse();
    futures::pin_mut!(timeout_fut);

    futures::select! {
        () = timeout_fut => {
                let mut all_tasks = vec![];
                let mut tasks = tasks.lock().await;
                all_tasks.append(&mut tasks);
                drop(tasks);
                for t in all_tasks {
                    t.cancel().await;
                }
                let running_test_cases = running_test_cases.lock().await;
                for i in &*running_test_cases {
                    sender
                        .send(Ok(SuiteEvents::case_stopped(*i, CaseStatus::TimedOut).into()))
                        .await
                        .unwrap();
                    sender
                        .send(Ok(SuiteEvents::case_finished(*i).into()))
                        .await
                        .unwrap();
                }
                return Ok(SuiteStatus::TimedOut);
            }
        r = test_fut => {
            initial_suite_status = match r {
                Err(e) => {
                    return Err(e);
                }
                Ok(s) => s,
            };
        }
    }

    let mut tasks = tasks.lock().await;
    let mut all_tasks = vec![];
    all_tasks.append(&mut tasks);
    // await for all invocations to complete for which test case never completed.
    let suite_status = join_all(all_tasks)
        .await
        .into_iter()
        .fold(initial_suite_status, get_suite_status_from_case_status);
    Ok(suite_status)
}
/// Runs the test component using `suite` and collects stdout logs and results.
Runs the test component using `suite` and collects stdout logs and results.
Runs the test component using ` suite ` and collects stdout logs and results .
async fn run_invocations(
    suite: &ftest::SuiteProxy,
    invocations: Vec<Invocation>,
    run_options: fidl_fuchsia_test::RunOptions,
    counter: &AtomicU32,
    sender: &mut mpsc::Sender<Result<FidlSuiteEvent, LaunchError>>,
    timeout_fut: futures::future::Shared<fasync::Timer>,
) -> Result<SuiteStatus, anyhow::Error> {
    let (run_listener_client, mut run_listener) =
        fidl::endpoints::create_request_stream().expect("cannot create request stream");
    suite.run(&mut invocations.into_iter().map(|i| i.into()), run_options, run_listener_client)?;
    let tasks = Arc::new(lock::Mutex::new(vec![]));
    let running_test_cases = Arc::new(lock::Mutex::new(HashSet::new()));
    let tasks_clone = tasks.clone();
    let initial_suite_status: SuiteStatus;
    let mut sender_clone = sender.clone();
    let test_fut = async {
        let mut initial_suite_status = SuiteStatus::DidNotFinish;
        while let Some(result_event) =
            run_listener.try_next().await.context("error waiting for listener")?
        {
            match result_event {
                ftest::RunListenerRequest::OnTestCaseStarted {
                    invocation,
                    primary_log,
                    listener,
                    control_handle: _,
                } => {
                    let name =
                        invocation.name.ok_or(format_err!("cannot find name in invocation"))?;
                    let identifier = counter.fetch_add(1, Ordering::Relaxed);
                    let events = vec![
                        Ok(SuiteEvents::case_found(identifier, name).into()),
                        Ok(SuiteEvents::case_started(identifier).into()),
                        Ok(SuiteEvents::case_stdout(identifier, primary_log).into()),
                    ];
                    for event in events {
                        sender_clone.send(event).await.unwrap();
                    }
                    let listener =
                        listener.into_stream().context("Cannot convert listener to stream")?;
                    running_test_cases.lock().await.insert(identifier);
                    let running_test_cases = running_test_cases.clone();
                    let sender = sender_clone.clone();
                    let task = fasync::Task::spawn(async move {
                        let status = listen_for_completion(listener, identifier, sender).await;
                        running_test_cases.lock().await.remove(&identifier);
                        status
                    });
                    tasks_clone.lock().await.push(task);
                }
                ftest::RunListenerRequest::OnFinished { .. } => {
                    initial_suite_status = SuiteStatus::Passed;
                    break;
                }
            }
        }
        Ok(initial_suite_status)
    }
    .fuse();
    futures::pin_mut!(test_fut);
    let timeout_fut = timeout_fut.fuse();
    futures::pin_mut!(timeout_fut);
    futures::select! {
        () = timeout_fut => {
                let mut all_tasks = vec![];
                let mut tasks = tasks.lock().await;
                all_tasks.append(&mut tasks);
                drop(tasks);
                for t in all_tasks {
                    t.cancel().await;
                }
                let running_test_cases = running_test_cases.lock().await;
                for i in &*running_test_cases {
                    sender
                        .send(Ok(SuiteEvents::case_stopped(*i, CaseStatus::TimedOut).into()))
                        .await
                        .unwrap();
                    sender
                        .send(Ok(SuiteEvents::case_finished(*i).into()))
                        .await
                        .unwrap();
                }
                return Ok(SuiteStatus::TimedOut);
            }
        r = test_fut => {
            initial_suite_status = match r {
                Err(e) => {
                    return Err(e);
                }
                Ok(s) => s,
            };
        }
    }
    let mut tasks = tasks.lock().await;
    let mut all_tasks = vec![];
    all_tasks.append(&mut tasks);
    // await for all invocations to complete for which test case never completed.
    let suite_status = join_all(all_tasks)
        .await
        .into_iter()
        .fold(initial_suite_status, get_suite_status_from_case_status);
    Ok(suite_status)
}
async fn run_invocations ( suite : & ftest :: SuiteProxy , invocations : Vec < Invocation > , run_options : fidl_fuchsia_test :: RunOptions , counter : & AtomicU32 , sender : & mut mpsc :: Sender < Result < FidlSuiteEvent , LaunchError > > , timeout_fut : futures :: future :: Shared < fasync :: Timer > , ) -> Result < SuiteStatus , anyhow :: Error > { let ( run_listener_client , mut run_listener ) = fidl :: endpoints :: create_request_stream ( ) . expect ( " " ) ; suite . run ( & mut invocations . into_iter ( ) . map ( | i | i . into ( ) ) , run_options , run_listener_client ) ? ; let tasks = Arc :: new ( lock :: Mutex :: new ( vec ! [ ] ) ) ; let running_test_cases = Arc :: new ( lock :: Mutex :: new ( HashSet :: new ( ) ) ) ; let tasks_clone = tasks . clone ( ) ; let initial_suite_status : SuiteStatus ; let mut sender_clone = sender . clone ( ) ; let test_fut = async { let mut initial_suite_status = SuiteStatus :: DidNotFinish ; while let Some ( result_event ) = run_listener . try_next ( ) . await . context ( " " ) ? { match result_event { ftest :: RunListenerRequest :: OnTestCaseStarted { invocation , primary_log , listener , control_handle : _ , } => { let name = invocation . name . ok_or ( format_err ! ( " " ) ) ? ; let identifier = counter . fetch_add ( 1 , Ordering :: Relaxed ) ; let events = vec ! [ Ok ( SuiteEvents case_found ( identifier name ) into ( ) ) Ok ( SuiteEvents case_started ( identifier ) into ( ) ) Ok ( SuiteEvents case_stdout ( identifier primary_log ) into ( ) ) ] ; for event in events { sender_clone . send ( event ) . await . unwrap ( ) ; } let listener = listener . into_stream ( ) . context ( " " ) ? ; running_test_cases . lock ( ) . await . insert ( identifier ) ; let running_test_cases = running_test_cases . clone ( ) ; let sender = sender_clone . clone ( ) ; let task = fasync :: Task :: spawn ( async move { let status = listen_for_completion ( listener , identifier , sender ) . await ; running_test_cases . lock ( ) . await . remove ( & identifier ) ; status } ) ; tasks_clone . lock ( ) . await . push ( task ) ; } ftest :: RunListenerRequest :: OnFinished { .. } => { initial_suite_status = SuiteStatus :: Passed ; break ; } } } Ok ( initial_suite_status ) } . fuse ( ) ; futures :: pin_mut ! ( test_fut ) ; let timeout_fut = timeout_fut . fuse ( ) ; futures :: pin_mut ! ( timeout_fut ) ; futures :: select ! { ( ) timeout_fut { let mut all_tasks vec [ ] let mut tasks tasks lock ( ) await all_tasks append ( mut tasks ) drop ( tasks ) for t in all_tasks { t cancel ( ) await } let running_test_cases running_test_cases lock ( ) await for i in running_test_cases { sender send ( Ok ( SuiteEvents case_stopped ( i CaseStatus TimedOut ) into ( ) ) ) await unwrap ( ) sender send ( Ok ( SuiteEvents case_finished ( i ) into ( ) ) ) await unwrap ( ) } return Ok ( SuiteStatus TimedOut ) } r test_fut { initial_suite_status match r { Err ( e ) { return Err ( e ) } Ok ( s ) s } } } let mut tasks = tasks . lock ( ) . await ; let mut all_tasks = vec ! [ ] ; all_tasks . append ( & mut tasks ) ; // await for all invocations to complete for which test case never completed. let suite_status = join_all ( all_tasks ) . await . into_iter ( ) . fold ( initial_suite_status , get_suite_status_from_case_status ) ; Ok ( suite_status ) }
Runs the test component using `suite` and collects stdout logs and results.
Runs the test component using ` suite ` and collects stdout logs and results .
suite
&ftest::SuiteProxy
invocations
Vec<Invocation>
run_options
fidl_fuchsia_test::RunOptions
counter
&AtomicU32
sender
&mut mpsc::Sender<Result<FidlSuiteEvent, LaunchError>>
timeout_fut
futures::future::Shared<fasync::Timer>
suite
&ftest::SuiteProxy
invocations
Vec<Invocation>
run_options
fidl_fuchsia_test::RunOptions
counter
&AtomicU32
sender
&mut mpsc::Sender<Result<FidlSuiteEvent, LaunchError>>
timeout_fut
futures::future::Shared<fasync::Timer>