// Copyright 2021 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use {
    anyhow::format_err,
    async_utils::hanging_get::client::HangingGetStream,
    bt_rfcomm::profile::server_channel_from_protocol,
    fidl_fuchsia_bluetooth_bredr as bredr,
    fidl_fuchsia_bluetooth_hfp::{NetworkInformation, PeerHandlerProxy},
    fuchsia_async::Task,
    fuchsia_bluetooth::{
        profile::{Attribute, ProtocolDescriptor},
        types::PeerId,
    },
    fuchsia_inspect::{self as inspect, Property},
    fuchsia_inspect_derive::{AttachError, Inspect},
    fuchsia_zircon as zx,
    futures::{
        channel::mpsc::{self, Sender},
        future::{self, Either, Future},
        select,
        stream::{empty, Empty},
        FutureExt, SinkExt, StreamExt,
    },
    parking_lot::Mutex,
    profile_client::ProfileEvent,
    std::{convert::TryInto, fmt, sync::Arc},
    tracing::{error, info, warn},
    vigil::{DropWatch, Vigil},
};

use super::{
    calls::{Call, CallAction, Calls},
    gain_control::GainControl,
    indicators::{AgIndicator, AgIndicators, HfIndicator},
    procedure::ProcedureMarker,
    ringer::Ringer,
    sco_state::{InspectableScoState, ScoActive, ScoState},
    service_level_connection::ServiceLevelConnection,
    slc_request::SlcRequest,
    update::AgUpdate,
    ConnectionBehavior, PeerRequest,
};

use crate::{
    a2dp,
    audio::AudioControl,
    config::AudioGatewayFeatureSupport,
    error::Error,
    features::CodecId,
    hfp,
    inspect::PeerTaskInspect,
    sco_connector::{ScoConnection, ScoConnector},
};

const CONNECTION_INIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

const DEFAULT_CODECS: &[CodecId] = &[CodecId::CVSD];

pub(super) struct PeerTask {
    id: PeerId,
    _local_config: AudioGatewayFeatureSupport,
    connection_behavior: ConnectionBehavior,
    profile_proxy: bredr::ProfileProxy,
    handler: Option<PeerHandlerProxy>,
    network: NetworkInformation,
    network_updates: Either<
        HangingGetStream<PeerHandlerProxy, NetworkInformation>,
        Empty<Result<NetworkInformation, fidl::Error>>,
    >,
    battery_level: u8,
    calls: Calls,
    gain_control: GainControl,
    connection: ServiceLevelConnection,
    a2dp_control: a2dp::Control,
    sco_connector: ScoConnector,
    sco_state: InspectableScoState,
    ringer: Ringer,
    audio_control: Arc<Mutex<Box<dyn AudioControl>>>,
    hfp_sender: Sender<hfp::Event>,
    manager_id: Option<hfp::ManagerConnectionId>,
    inspect: PeerTaskInspect,
}

impl Inspect for &mut PeerTask {
    fn iattach(self, parent: &inspect::Node, name: impl AsRef<str>) -> Result<(), AttachError> {
        let _ = self.inspect.iattach(parent, name)?;
        self.sco_state.iattach(self.inspect.node(), "sco_connection")?;
        Ok(())
    }
}

impl PeerTask {
    pub fn new(
        id: PeerId,
        profile_proxy: bredr::ProfileProxy,
        audio_control: Arc<Mutex<Box<dyn AudioControl>>>,
        local_config: AudioGatewayFeatureSupport,
        connection_behavior: ConnectionBehavior,
        hfp_sender: Sender<hfp::Event>,
    ) -> Result<Self, Error> {
        let connection = ServiceLevelConnection::with_init_timeout(fuchsia_async::Timer::new(
            CONNECTION_INIT_TIMEOUT,
        ));
        let sco_connector = ScoConnector::build(profile_proxy.clone());
        let a2dp_control = a2dp::Control::connect();
        Ok(Self {
            id,
            _local_config: local_config,
            connection_behavior,
            profile_proxy,
            handler: None,
            network: NetworkInformation::EMPTY,
            network_updates: empty().right_stream(),
            // Default to a battery level of 5 (max). This will be updated when we receive
            // a battery information update.
            battery_level: 5,
            calls: Calls::new(None),
            gain_control: GainControl::new()?,
            connection,
            a2dp_control,
            sco_connector,
            sco_state: InspectableScoState::default(),
            ringer: Ringer::default(),
            audio_control,
            hfp_sender,
            manager_id: None,
            inspect: PeerTaskInspect::new(id),
        })
    }

    pub fn spawn(
        id: PeerId,
        profile_proxy: bredr::ProfileProxy,
        audio_control: Arc<Mutex<Box<dyn AudioControl>>>,
        local_config: AudioGatewayFeatureSupport,
        connection_behavior: ConnectionBehavior,
        hfp_sender: Sender<hfp::Event>,
        inspect: &inspect::Node,
    ) -> Result<(Task<()>, Sender<PeerRequest>), Error> {
        let (sender, receiver) = mpsc::channel(0);
        let mut peer = Self::new(
            id,
            profile_proxy,
            audio_control,
            local_config,
            connection_behavior,
            hfp_sender,
        )?;
        if let Err(e) = peer.iattach(inspect, "task") {
            warn!("Failed to attach PeerTaskInspect to provided inspect node: {}", e)
        }
        let task = Task::local(peer.run(receiver).map(|_| ()));
        Ok((task, sender))
    }

    fn set_handler(&mut self, handler: Option<PeerHandlerProxy>) {
        self.handler = handler;
        self.inspect.connected_peer_handler.set(self.handler.is_some());
    }

    /// Always give preference to connection requests received from the peer device.
    async fn on_connection_request(
        &mut self,
        _protocol: Vec<ProtocolDescriptor>,
        channel: fuchsia_bluetooth::types::Channel,
    ) -> Result<(), Error> {
        if self.connection.connected() {
            info!("Overwriting existing connection to {}", self.id);
        } else {
            info!("Connection request from {}", self.id);
        }
        self.connection.connect(channel);
        if let Err(e) = self.connection.iattach(self.inspect.node(), "service_level_connection") {
            warn!("Failed to attach ServiceLevelConnection to PeerTaskInspect: {}", e)
        }
        if let Some(id) = self.manager_id {
            self.notify_peer_connected(id).await;
            self.setup_handler().await?;
        }
        Ok(())
    }

    /// Make a connection to the peer only if there is not an existing connection.
    async fn connect(&mut self, mut params: bredr::ConnectParameters) -> Result<(), anyhow::Error> {
        if self.connection.connected() {
            info!("Already connected to peer: {}.", self.id);
            return Ok(());
        }
        info!("Initiating connection to peer: {}", self.id);
        let channel = self
            .profile_proxy
            .connect(&mut self.id.into(), &mut params)
            .await?
            .map_err(|e| format_err!("Profile connection request error: {:?}", e))?;
        self.connection.connect(channel.try_into()?);
        if let Err(e) = self.connection.iattach(self.inspect.node(), "service_level_connection") {
            warn!("Failed to attach ServiceLevelConnection to PeerTaskInspect: {}", e)
        }
        if let Some(id) = self.manager_id {
            self.notify_peer_connected(id).await;
            self.setup_handler()
                .await
                .map_err(|e| format_err!("Error setting up peer handler: {}", e))?;
        }
        Ok(())
    }

    async fn on_search_result(
        &mut self,
        protocol: Option<Vec<ProtocolDescriptor>>,
        _attributes: Vec<Attribute>,
    ) {
        info!("Search results received for {}", self.id);

        let server_channel = match protocol.as_ref().map(server_channel_from_protocol) {
            Some(Some(sc)) => sc,
            _ => {
                info!("Search result received for non-RFCOMM protocol: {:?}", protocol);
                return;
            }
        };
        let params = bredr::ConnectParameters::Rfcomm(bredr::RfcommParameters {
            channel: Some(server_channel.into()),
            ..bredr::RfcommParameters::EMPTY
        });

        if self.connection_behavior.autoconnect {
            if let Err(e) = self.connect(params).await {
                info!("Error inititating connecting to peer {}: {:?}", self.id, e);
            }
        }
    }

    async fn notify_peer_connected(&mut self, manager_id: hfp::ManagerConnectionId) {
        let (proxy, handle) =
            fidl::endpoints::create_proxy().expect("Cannot create required fidl handle");
        self.hfp_sender
            .send(hfp::Event::PeerConnected { peer_id: self.id, manager_id, handle })
            .await
            .expect("Cannot communicate with main Hfp task");
        self.set_handler(Some(proxy));
    }

    async fn setup_handler(&mut self) -> Result<(), Error> {
        info!("Got request to handle peer {} headset using handler", self.id);
        if let Some(handler) = self.handler.clone() {
            // Getting the network information the first time should always return a complete table.
            // If the call returns an error, do not set up the handler.
            let info = match handler.watch_network_information().await {
                Ok(info) => info,
                Err(fidl::Error::ClientChannelClosed {
                    status: zx::Status::PEER_CLOSED, ..
                }) => {
                    return Ok(());
                }
                Err(e) => {
                    warn!("Error handling peer request: {}", e);
                    return Ok(());
                }
            };

            self.handle_network_update(info).await;

            let client_end = self.gain_control.get_client_end()?;
            if let Err(e) = handler.gain_control(client_end) {
                warn!("Error setting gain control for peer {}: {}", self.id, e);
                return Ok(());
            }

            self.calls = Calls::new(Some(handler.clone()));
            if let Err(e) = self.calls.iattach(self.inspect.node(), "calls") {
                warn!("Failed to attach Calls to PeerTaskInspect: {}", e)
            }

            self.create_network_updates_stream(handler);
        }
        Ok(())
    }

    /// A new call manager is connected. The PeerTask should
    async fn on_manager_connected(&mut self, id: hfp::ManagerConnectionId) -> Result<(), Error> {
        self.manager_id = Some(id);
        if !self.connection.connected() {
            // If there is no peer connection, there should not be a handler.
            self.set_handler(None);
            return Ok(());
        }
        self.notify_peer_connected(id).await;
        self.setup_handler().await?;
        Ok(())
    }

    /// When a new handler is received, the state is not known. It might be stale because the
    /// system is asynchronous and the PeerHandler connection has two sides. The handler will only
    /// be stored if all the associated fidl calls are successful. Otherwise it is dropped without
    /// being set.
    async fn on_peer_handler(&mut self, handler: PeerHandlerProxy) -> Result<(), Error> {
        info!("Got request to handle peer {} headset using handler", self.id);
        // Getting the network information the first time should always return a complete table.
        // If the call returns an error, do not set up the handler.
        let info = match handler.watch_network_information().await {
            Ok(info) => info,
            Err(fidl::Error::ClientChannelClosed { status: zx::Status::PEER_CLOSED, .. }) => {
                return Ok(());
            }
            Err(e) => {
                warn!("Error handling peer request: {}", e);
                return Ok(());
            }
        };

        self.handle_network_update(info).await;

        let client_end = self.gain_control.get_client_end()?;
        if let Err(e) = handler.gain_control(client_end) {
            warn!("Error setting gain control for peer {}: {}", self.id, e);
            return Ok(());
        }

        self.calls = Calls::new(Some(handler.clone()));
        if let Err(e) = self.calls.iattach(self.inspect.node(), "calls") {
            warn!("Failed to attach Calls to PeerTaskInspect: {}", e)
        }

        self.create_network_updates_stream(handler.clone());
        self.set_handler(Some(handler));

        Ok(())
    }

    /// Set a new HangingGetStream to watch for network information updates.
    fn create_network_updates_stream(&mut self, handler: PeerHandlerProxy) {
        self.network_updates =
            HangingGetStream::new_with_fn_ptr(handler, PeerHandlerProxy::watch_network_information)
                .left_stream();
    }

    async fn peer_request(&mut self, request: PeerRequest) -> Result<(), Error> {
        match request {
            PeerRequest::Profile(ProfileEvent::PeerConnected { protocol, channel, id: _ }) => {
                let protocol = protocol.iter().map(ProtocolDescriptor::from).collect();
                self.on_connection_request(protocol, channel).await?;
            }
            PeerRequest::Profile(ProfileEvent::SearchResult { protocol, attributes, id: _ }) => {
                let protocol = protocol.map(|p| p.iter().map(ProtocolDescriptor::from).collect());
                let attributes = attributes.iter().map(Attribute::from).collect();
                self.on_search_result(protocol, attributes).await;
            }
            PeerRequest::ManagerConnected { id } => self.on_manager_connected(id).await?,
            PeerRequest::Handle(handler) => self.on_peer_handler(handler).await?,
            PeerRequest::BatteryLevel(level) => {
                self.battery_level = level;
                let status = AgIndicator::BatteryLevel(self.battery_level);
                self.phone_status_update(status).await;
            }
            PeerRequest::Behavior(behavior) => {
                self.connection_behavior = behavior;
            }
        }
        Ok(())
    }

    /// Processes a `request` for information from an HFP procedure.
    async fn procedure_request(&mut self, request: SlcRequest) {
        info!("HF procedure request ({}): {:?}", self.id, request);
        let marker = (&request).into();
        match request {
            SlcRequest::GetAgFeatures { response } => {
                let features = (&self._local_config).into();
                // Update the procedure with the retrieved AG update.
                self.connection.receive_ag_request(marker, response(features)).await;
            }
            SlcRequest::GetSubscriberNumberInformation { response } => {
                let result = if let Some(handler) = &mut self.handler {
                    handler.subscriber_number_information().await.ok().unwrap_or_else(Vec::new)
                } else {
                    vec![]
                };
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::GetAgIndicatorStatus { response } => {
                let call_ind = self.calls.indicators();
                let status = AgIndicators {
                    service: self.network.service_available.unwrap_or(false),
                    call: call_ind.call,
                    callsetup: call_ind.callsetup,
                    callheld: call_ind.callheld,
                    signal: self.network.signal_strength.map(|ss| ss as u8).unwrap_or(0),
                    roam: self.network.roaming.unwrap_or(false),
                    battchg: self.battery_level,
                };
                // Update the procedure with the retrieved AG update.
                self.connection.receive_ag_request(marker, response(status)).await;
            }
            SlcRequest::GetNetworkOperatorName { response } => {
                let format = self.connection.network_operator_name_format();
                let name = match &self.handler {
                    Some(h) => {
                        let result = h.query_operator().await;
                        if let Err(err) = &result {
                            warn!(
                                "Got error attempting to retrieve operator name from AG: {:} for peer {:}",
                                err, self.id
                            );
                        };
                        result.ok().flatten()
                    }
                    None => None,
                };
                let name_option = match (name, format) {
                    (Some(n), Some(_)) => Some(n),
                    _ => None, // The format must be set before getting the network name.
                };
                // Update the procedure with the result of retrieving the AG network name.
                self.connection.receive_ag_request(marker, response(name_option)).await;
            }
            SlcRequest::SendDtmf { code, response } => {
                let result = self.calls.send_dtmf_code(code).await;
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::SendHfIndicator { indicator, response } => {
                self.hf_indicator_update(indicator);
                self.connection.receive_ag_request(marker, response()).await;
            }
            SlcRequest::SetNrec { enable, response } => {
                let result = if let Some(handler) = &mut self.handler {
                    if let Ok(Ok(())) = handler.set_nrec_mode(enable).await {
                        Ok(())
                    } else {
                        Err(())
                    }
                } else {
                    Err(())
                };
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::SpeakerVolumeSynchronization { level, response } => {
                self.gain_control.report_speaker_gain(level);
                self.connection.receive_ag_request(marker, response()).await;
            }
            SlcRequest::MicrophoneVolumeSynchronization { level, response } => {
                self.gain_control.report_microphone_gain(level);
                self.connection.receive_ag_request(marker, response()).await;
            }
            SlcRequest::QueryCurrentCalls { response } => {
                let result = self.calls.current_calls();
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::Answer { response } => {
                let result = self.calls.answer().map_err(|e| {
                    warn!("Unexpected Answer from Hands Free for peer {}: {}", self.id, e);
                });
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::HangUp { response } => {
                let result = self.calls.hang_up().map_err(|e| {
                    warn!("Unexpected Hang Up from Hands Free for peer {}: {}", self.id, e);
                });
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::Hold { command, response } => {
                let result = self.calls.hold(command).map_err(|e| {
                    warn!(
                        "Unexpected Action {:?} from Hands Free for peer {}: {}",
                        command, self.id, e
                    );
                });
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::InitiateCall { call_action, response } => {
                let result = self.handle_initiate_call(call_action).await;
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::SynchronousConnectionSetup { response } => {
                if self.sco_state.is_active() {
                    info!("Got SCO setup request when SCO state was active");
                    // Drop existing SCO connection.
                    self.sco_state.iset(ScoState::SettingUp);
                }
                // TODO(fxbug.dev/72681): Because we may need to send an OK response to the HF
                // just before  setting up the synchronous connection, we send it here by routing
                // through the procedure.
                self.connection.receive_ag_request(marker, AgUpdate::Ok).await;

                let codecs = self.get_codecs();
                let setup_result = self.sco_connector.connect(self.id.clone(), codecs).await;
                let finish_result = match setup_result {
                    Ok(conn) => self.finish_sco_connection(conn).await,
                    Err(err) => Err(err.into()),
                };
                let result =
                    finish_result.map_err(|e| warn!("Error setting up audio connection: {:?}", e));
                self.connection.receive_ag_request(marker, response(result)).await;
            }
            SlcRequest::RestartCodecConnectionSetup { response } => {
                self.connection.receive_ag_request(marker, response()).await;
                // Start CodecConnectionSetup running again.
                self.initiate_codec_negotiation().await;
            }
        };
    }

    pub async fn handle_initiate_call(&mut self, call_action: CallAction) -> Result<(), ()> {
        let three_way_calling = self.connection.three_way_calling();
        let call_active = self.calls.is_call_active();
        let handler = self.handler.as_ref().ok_or(())?;

        if call_active && !three_way_calling {
            warn!("Attempting to initiate unsupported outgoing call during active call.");
            return Err(());
        }

        if call_active && three_way_calling {
            // Hold calls, and return Err(_) if it fails.
            self.calls.hold_active().map_err(|e| {
                warn!(
                    "Failed to hold active call when making outgoing call for peer {}: {}",
                    self.id, e
                )
            })?;
        }

        match handler.request_outgoing_call(&mut call_action.into()).await {
            Ok(Ok(())) => Ok(()),
            err => {
                warn!("Error initiating outgoing call for peer {}: {:?}", self.id, err);
                Err(())
            }
        }
    }

    pub async fn run(mut self, mut task_channel: mpsc::Receiver<PeerRequest>) -> Self {
        loop {
            let mut active_sco_closed_fut = self.on_active_sco_closed().fuse();
            info!("Beginning select for peer {:?}, SCO state is {:?}", self.id, self.sco_state);
            let mut sco_state = self.sco_state.as_mut();
            select! {
                // Wait until the HF sets up a SCO connection.
                conn_res = sco_state.on_connected() => {
                    drop(sco_state);
                    info!("Handling SCO Connection accepted for peer {}.", self.id);
                    match conn_res {
                        Ok(sco) if !sco.is_closed() => {
                            let finish_sco_res = self.finish_sco_connection(sco).await;
                            if let Err(err) = finish_sco_res {
                                warn!("Failed to finish SCO connection with {:} for peer {}", err, self.id)
                            }
                            let call_transfer_res = self.calls.transfer_to_hf();
                            if let Err(err) = call_transfer_res {
                                warn!("Transfer to HF failed with {:} for peer {}", err, self.id)
                            }
                        },
                        // This can occur if the HF opens and closes a SCO connection immediately.
                        Ok(_) => warn!("Got already closed SCO connection for peer {}.", self.id),
                        Err(err) => {
                            warn!("Got error waiting for SCO connection {:} for peer {}", err, self.id);
                            break;
                        }
                    }
                }
                // New request coming from elsewhere in the component
                request = task_channel.next() => {
                    drop(sco_state);
                    info!("Handling peer request {:?} for peer {}", request, self.id);
                    if let Some(request) = request {
                        if let Err(e) = self.peer_request(request).await {
                            warn!("Error handling peer request {} for peer {}", e, self.id);
                            break;
                        }
                    } else {
                        info!("Peer task channel closed for peer {}", self.id);
                        break;
                    }
                }
                // New request on the gain control protocol
                request = self.gain_control.select_next_some() => {
                    info!("Handling gain control {:?} for peer {}", request, self.id);
                    self.connection.receive_ag_request(ProcedureMarker::VolumeControl, request.into()).await;
                },
                // A new call state has been received from the call service
                update = self.calls.select_next_some() => {
                    drop(sco_state);
                    info!("Handling call {:?} for peer {}", update, self.id);
                    // TODO(fxbug.dev/75538): for in-band ring  setup audio if should_ring is true
                    self.ringer.ring(self.calls.should_ring());
                    if update.callwaiting {
                        if let Some(call) = self.calls.waiting() {
                            self.call_waiting_update(call).await;
                        }
                    }
                    for status in update.to_vec() {
                        self.phone_status_update(status).await;
                    }
                   // Sync the SCO connection state to the call state.
                   // The error is already logged and we can't do anything.
                   let _ = self.update_sco_state().await;
                }
                // SCO connection has closed.
                _ = active_sco_closed_fut => {
                    drop(sco_state);
                    info!("Handling SCO Connection closed for peer {}, transferring call to AG.", self.id);
                    self.sco_state.iset(ScoState::TearingDown);
                    let call_transfer_res = self.calls.transfer_to_ag();
                    if let Err(err) = call_transfer_res {
                        warn!("Transfer to AG failed with {:} for peer {}", err, self.id)
                    }
                }
                request = self.connection.next() => {
                    drop(sco_state);
                    info!("Handling SLC request {:?} for peer {}.", request, self.id);
                    if let Some(request) = request {
                        match request {
                            Ok(r) => self.procedure_request(r).await,
                            Err(e) => {
                                warn!("SLC stream error {:?} for peer {}", e, self.id);
                                break;
                            }
                        }
                    } else {
                        info!("Peer task channel closed for peer {}", self.id);
                        break;
                    }
                }
                update = self.network_updates.next() => {
                    drop(sco_state);
                    info!("Handling network update {:?} for peer {}", update, self.id);
                    if let Some(update) = stream_item_map_or_log(update, "PeerHandler::WatchNetworkUpdate", &self.id) {
                        self.handle_network_update(update).await
                    } else {
                        break;
                    }
                }
                _ = self.ringer.select_next_some() => {
                    drop(sco_state);
                    info!("Handling ring for peer {}.", self.id);
                    if let Some(call) = self.calls.ringing() {
                        self.ring_update(call).await;
                    } else {
                        self.ringer.ring(false);
                    }
                }
                complete => break,
            }
        }

        info!("Stopping task for peer {}", self.id);

        self
    }

    /// Sends an HF Indicator update to the client.
    fn hf_indicator_update(&mut self, indicator: HfIndicator) {
        match indicator {
            ind @ HfIndicator::EnhancedSafety(_) => {
                info!("Received EnhancedSafety HF Indicator update {:?} for peer {}", ind, self.id);
            }
            HfIndicator::BatteryLevel(v) => {
                if let Some(handler) = &mut self.handler {
                    if let Err(e) = handler.report_headset_battery_level(v) {
                        warn!("Couldn't report headset battery level {:?} for peer {}", e, self.id);
                    }
                }
                self.inspect.set_hf_battery_level(v);
            }
        }
    }

    /// Start the Codec Connection procedure if no sco connection exists.
    /// This procedure will negotiate a codec and eventually call
    /// `sco_connector.connect`.
    async fn initiate_codec_negotiation(&mut self) {
        if self.sco_state.is_active() {
            return;
        }
        self.connection
            .receive_ag_request(ProcedureMarker::CodecConnectionSetup, AgUpdate::CodecSetup(None))
            .await;
    }

    /// Update the SCO connection state to reflect the call state, creating a
    /// a future to wait for incoming SCO connections if necessary.
    async fn update_sco_state(&mut self) -> Result<(), ()> {
        let call_active = self.calls.is_call_active();
        let call_transferred = self.calls.is_call_transferred_to_ag();
        if call_active && call_transferred {
            error!("Call both active and transferred for peer {}", self.id);
            return Err(());
        }

        let previous_sco_state = &*self.sco_state;

        info!(
            %self.id,
            ?previous_sco_state,
            "update_sco_state: active: {}, transferred: {}",
            call_active,
            call_transferred
        );

        if call_active {
            match previous_sco_state {
                // A call just started, so set up SCO.
                ScoState::Inactive
                // A call was just transferred to the HF, so set up SCO.
                | ScoState::AwaitingRemote(_)
                => {
                    self.initiate_codec_negotiation().await;
                    self.sco_state.iset(ScoState::SettingUp);
                },
                // We are negotiating codecs; wait for that to finish before starting the SCO
                // connection, so do nothing.
                ScoState::SettingUp
                // The SCO connection was closed by the peer and we requested the call manager set
                // that call as transferred to AG,  but we are waiting on the call manager to do
                // so, so do nothing.
                | ScoState::TearingDown
                // A call is active and we have a SCO connection, so do nothing.
                | ScoState::Active(_) => {},
            }
        } else if call_transferred {
            // If a call is transferred to the AG, we should be waiting for a connection from the
            // HF to transfer it back.
            if let ScoState::AwaitingRemote(_) = previous_sco_state {
                // Already waiting for a SCO connection, nothing to do.
                info!("update_sco_state: already awaiting");
                return Ok(());
            }
            let fut = self.sco_connector.accept(self.id.clone(), self.get_codecs());
            self.sco_state.iset(ScoState::AwaitingRemote(Box::pin(fut)));
        } else {
            /* No call in progress */
            self.sco_state.iset(ScoState::Inactive);
        };

        info!(%self.id, ?self.sco_state, "update_sco_state: finished");

        Ok(())
    }

    async fn finish_sco_connection(&mut self, sco_connection: ScoConnection) -> Result<(), Error> {
        let peer_id = self.id.clone();
        info!("Finishing SCO connection for peer {:}.", peer_id);
        let res = self.a2dp_control.pause(Some(self.id.clone())).await;
        let pause_token = match res {
            Err(e) => {
                warn!("Couldn't pause A2DP Audio: {:?} for peer {:}", e, peer_id);
                None
            }
            Ok(token) => {
                info!("Successfully paused audio for peer {:}.", peer_id);
                Some(token)
            }
        };
        {
            let mut audio = self.audio_control.lock();
            // Start the DAI with the given parameters
            if let Err(e) = audio.start(self.id.clone(), sco_connection.params.clone().into()) {
                // Cancel the SCO connection, we can't send audio.
                // TODO(fxbug.dev/79784): this probably means we should just cancel out of HFP and
                // this peer's connection entirely.
                warn!(
                    "Couldn't start Audio DAI ({:?}) for peer {:} - dropping audio connection",
                    self.id, e
                );
                return Err(Error::system(format!("Couldn't start audio DAI"), e));
            } else {
                info!("Successfully started Audio DAI for peer {:}.", peer_id);
            }
        }

        let vigil = Vigil::new(ScoActive { sco_connection, _pause_token: pause_token });
        Vigil::watch(&vigil, {
            let control = self.audio_control.clone();
            move |_| match control.lock().stop() {
                Err(e) => warn!("Couldn't stop audio for peer {:}: {:?}", peer_id, e),
                Ok(()) => info!("Stopped HFP Audio for peer {:}", peer_id),
            }
        });
        info!("In finish_sco_connection for peer {:}, SCO state is {:?}", peer_id, vigil);
        self.sco_state.iset(ScoState::Active(vigil));

        Ok(())
    }

    fn on_active_sco_closed(&self) -> impl Future<Output = ()> + 'static {
        match &*self.sco_state {
            ScoState::Active(connection) => connection.sco_connection.on_closed().left_future(),
            _ => future::pending().right_future(),
        }
    }

    /// Request to send the phone `status` by initiating the Phone Status Indicator
    /// procedure.
    async fn ring_update(&mut self, call: Call) {
        self.connection.receive_ag_request(ProcedureMarker::Ring, AgUpdate::Ring(call)).await;
    }

    /// Request to send a Call Waiting Notification.
    async fn call_waiting_update(&mut self, call: Call) {
        self.connection
            .receive_ag_request(
                ProcedureMarker::CallWaitingNotifications,
                AgUpdate::CallWaiting(call),
            )
            .await;
    }

    /// Request to send the phone `status` by initiating the Phone Status Indicator
    /// procedure.
    async fn phone_status_update(&mut self, status: AgIndicator) {
        self.connection.receive_ag_request(ProcedureMarker::PhoneStatus, status.into()).await;
    }

    /// Update the network information with the provided `update` value.
    async fn handle_network_update(&mut self, update: NetworkInformation) {
        if update_table_entry(&mut self.network.service_available, &update.service_available) {
            let status = AgIndicator::Service(self.network.service_available.unwrap() as u8);
            self.phone_status_update(status).await;
        }
        if update_table_entry(&mut self.network.signal_strength, &update.signal_strength) {
            let status = AgIndicator::Signal(self.network.signal_strength.unwrap() as u8);
            self.phone_status_update(status).await;
        }
        if update_table_entry(&mut self.network.roaming, &update.roaming) {
            let status = AgIndicator::Roam(self.network.roaming.unwrap() as u8);
            self.phone_status_update(status).await;
        }
        self.inspect.network.update(&self.network);
    }

    fn get_codecs(&self) -> Vec<CodecId> {
        self.connection.get_selected_codec().map_or(DEFAULT_CODECS.to_vec(), |c| vec![c])
    }
}

/// Table entries are all optional fields. Update `dst` if `src` is present and differs from `dst`.
///
/// Return true if the update occurred.
fn update_table_entry<T: PartialEq + Clone>(dst: &mut Option<T>, src: &Option<T>) -> bool {
    if src.is_some() && src != dst {
        *dst = src.clone();
        true
    } else {
        false
    }
}

/// Take an item from a stream the produces results and return an `Option<T>` when available.
/// Log a message including the `stream_name` when the stream produces an error or is exhausted.
///
/// This is useful when dealing with streams where the only meaningful difference between
/// a terminated stream and an error value is how it should be logged.
fn stream_item_map_or_log<T, E: fmt::Debug>(
    item: Option<Result<T, E>>,
    stream_name: &str,
    peer_id: &PeerId,
) -> Option<T> {
    match item {
        Some(Ok(value)) => Some(value),
        Some(Err(e)) => {
            warn!("Error on stream {} for peer {:}: {:?}", stream_name, peer_id, e);
            None
        }
        None => {
            info!("Stream {} closed for peer {:}", stream_name, peer_id);
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use {
        super::*,
        assert_matches::assert_matches,
        async_test_helpers::run_while,
        async_utils::PollExt,
        at_commands::{self as at, SerDe},
        bt_rfcomm::{profile::build_rfcomm_protocol, ServerChannel},
        core::task::Poll,
        fidl::AsHandleRef,
        fidl_fuchsia_bluetooth_bredr::{ProfileMarker, ProfileRequestStream, ScoErrorCode},
        fidl_fuchsia_bluetooth_hfp::{
            CallDirection, CallRequest, CallRequestStream, CallState, NextCall, PeerHandlerMarker,
            PeerHandlerRequest, PeerHandlerRequestStream, PeerHandlerWatchNextCallResponder,
            SignalStrength,
        },
        fuchsia_async as fasync,
        fuchsia_bluetooth::types::Channel,
        futures::{
            future::ready,
            pin_mut,
            stream::{FusedStream, Stream},
            SinkExt,
        },
        proptest::prelude::*,
        std::convert::TryFrom,
    };

    use crate::{
        audio::TestAudioControl,
        features::{AgFeatures, HfFeatures},
        peer::{
            calls::Number,
            indicators::{AgIndicatorsReporting, HfIndicators},
            service_level_connection::{
                tests::{
                    create_and_connect_slc, create_and_initialize_slc,
                    expect_data_received_by_peer, expect_peer_ready, serialize_at_response,
                },
                SlcState,
            },
        },
    };

    fn arb_signal() -> impl Strategy<Value = Option<SignalStrength>> {
        proptest::option::of(prop_oneof![
            Just(SignalStrength::None),
            Just(SignalStrength::VeryLow),
            Just(SignalStrength::Low),
            Just(SignalStrength::Medium),
            Just(SignalStrength::High),
            Just(SignalStrength::VeryHigh),
        ])
    }

    prop_compose! {
        fn arb_network()(
            service_available in any::<Option<bool>>(),
            signal_strength in arb_signal(),
            roaming in any::<Option<bool>>()
        ) -> NetworkInformation {
            NetworkInformation {
                service_available,
                roaming,
                signal_strength,
                ..NetworkInformation::EMPTY
            }
        }
    }

    fn setup_peer_task(
        connection: Option<ServiceLevelConnection>,
    ) -> (PeerTask, Sender<PeerRequest>, mpsc::Receiver<PeerRequest>, ProfileRequestStream) {
        let (sender, receiver) = mpsc::channel(1);
        let (proxy, stream) = fidl::endpoints::create_proxy_and_stream::<ProfileMarker>().unwrap();
        let audio: Arc<Mutex<Box<dyn AudioControl>>> =
            Arc::new(Mutex::new(Box::new(TestAudioControl::default())));
        let mut task = PeerTask::new(
            PeerId(1),
            proxy,
            audio,
            AudioGatewayFeatureSupport::default(),
            ConnectionBehavior::default(),
            mpsc::channel(1).0,
        )
        .expect("Could not create PeerTask");
        if let Some(conn) = connection {
            task.connection = conn;
        }
        (task, sender, receiver, stream)
    }

    proptest! {
        #![proptest_config(ProptestConfig{
            // Disable persistence to avoid the warning for not running in the
            // source code directory (since we're running on a Fuchsia target)
            failure_persistence: None,
            .. ProptestConfig::default()
        })]
        #[test]
        fn updates(a in arb_network(), b in arb_network()) {
            let mut exec = fasync::TestExecutor::new().unwrap();
            let mut task = setup_peer_task(None).0;

            task.network = a.clone();
            exec.run_singlethreaded(task.handle_network_update(b.clone()));

            let c = task.network.clone();

            // Check that the `service_available` field is correct.
            if b.service_available.is_some() {
                assert_eq!(c.service_available, b.service_available);
            } else if a.service_available.is_some() {
                assert_eq!(c.service_available, a.service_available);
            } else {
                assert_eq!(c.service_available, None);
            }

            // Check that the `signal_strength` field is correct.
            if b.signal_strength.is_some() {
                assert_eq!(c.signal_strength, b.signal_strength);
            } else if a.signal_strength.is_some() {
                assert_eq!(c.signal_strength, a.signal_strength);
            } else {
                assert_eq!(c.signal_strength, None);
            }

            // Check that the `roaming` field is correct.
            if b.roaming.is_some() {
                assert_eq!(c.roaming, b.roaming);
            } else if a.roaming.is_some() {
                assert_eq!(c.roaming, a.roaming);
            } else {
                assert_eq!(c.roaming, None);
            }
        }
    }

    #[fuchsia::test]
    fn handle_peer_request_stores_peer_handler_proxy() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        let mut peer = setup_peer_task(None).0;
        assert!(peer.handler.is_none());
        let (proxy, mut stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();

        {
            let request_fut = peer.peer_request(PeerRequest::Handle(proxy));
            pin_mut!(request_fut);

            let (result, request_fut) = run_while(&mut exec, request_fut, stream.next());
            match result {
                Some(Ok(PeerHandlerRequest::WatchNetworkInformation { responder })) => {
                    responder
                        .send(NetworkInformation::EMPTY)
                        .expect("Successfully send network information");
                }
                x => panic!("Expected watch network information request: {:?}", x),
            };

            // Request future should finish.
            let request_result = exec.run_singlethreaded(request_fut);
            assert!(request_result.is_ok());
        }

        assert!(peer.handler.is_some());
    }

    #[fuchsia::test(allow_stalls = false)]
    async fn handle_peer_request_decline_to_handle() {
        let mut peer = setup_peer_task(None).0;
        assert!(peer.handler.is_none());
        let (proxy, server_end) = fidl::endpoints::create_proxy::<PeerHandlerMarker>().unwrap();

        // close the PeerHandler channel by dropping the server endpoint.
        drop(server_end);

        peer.peer_request(PeerRequest::Handle(proxy)).await.expect("request to succeed");
        assert!(peer.handler.is_none());
    }

    #[fuchsia::test(allow_stalls = false)]
    async fn task_runs_until_all_event_sources_close() {
        let (peer, sender, receiver, _) = setup_peer_task(None);
        // Peer will stop running when all event sources are closed.
        // Close all sources
        drop(sender);

        // Test that `run()` completes.
        let _ = peer.run(receiver).await;
    }

    /// Expects a message to be received by the peer. This expectation does not validate the
    /// contents of the data received.
    #[track_caller]
    fn expect_message_received_by_peer(exec: &mut fasync::TestExecutor, remote: &mut Channel) {
        let mut vec = Vec::new();
        let mut remote_fut = Box::pin(remote.read_datagram(&mut vec));
        assert!(exec.run_until_stalled(&mut remote_fut).is_ready());
    }

    #[fuchsia::test]
    fn peer_task_drives_procedure() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        let (mut peer, _sender, receiver, _profile) = setup_peer_task(None);

        // Set up the RFCOMM connection.
        let (local, mut remote) = Channel::create();
        exec.run_singlethreaded(peer.on_connection_request(vec![], local))
            .expect("Connection request handling to succeed");

        let mut peer_task_fut = Box::pin(peer.run(receiver));
        assert!(exec.run_until_stalled(&mut peer_task_fut).is_pending());

        // Simulate remote peer (HF) sending AT command to start the SLC Init Procedure.
        let features = HfFeatures::empty();
        let command = format!("AT+BRSF={}\r", features.bits()).into_bytes();
        let _ = remote.as_ref().write(&command);
        let _ = exec.run_until_stalled(&mut peer_task_fut);
        // We then expect an outgoing message to the peer.
        expect_message_received_by_peer(&mut exec, &mut remote);
    }

    #[fuchsia::test]
    fn network_information_updates_are_relayed_to_peer() {
        // This test produces the following two network updates. Each update is expected to
        // be sent to the remote peer.
        let network_update_1 = NetworkInformation {
            signal_strength: Some(SignalStrength::Low),
            roaming: Some(false),
            ..NetworkInformation::EMPTY
        };
        // Expect to send the Signal and Roam indicators to the peer.
        let expected_data1 = vec![AgIndicator::Signal(3).into(), AgIndicator::Roam(0).into()];

        let network_update_2 = NetworkInformation {
            service_available: Some(true),
            roaming: Some(true),
            ..NetworkInformation::EMPTY
        };
        // Expect to send the Service and Roam indicators to the peer.
        let expected_data2 = vec![AgIndicator::Service(1).into(), AgIndicator::Roam(1).into()];

        // The value after the updates are applied is expected to be the following
        let expected_network = NetworkInformation {
            service_available: Some(true),
            roaming: Some(true),
            signal_strength: Some(SignalStrength::Low),
            ..NetworkInformation::EMPTY
        };

        // Set up the executor, peer, and background call manager task
        let mut exec = fasync::TestExecutor::new().unwrap();
        let state = SlcState {
            ag_indicator_events_reporting: AgIndicatorsReporting::new_enabled(),
            ..SlcState::default()
        };
        let (connection, mut remote) = create_and_initialize_slc(state);
        let (peer, mut sender, receiver, _profile) = setup_peer_task(Some(connection));

        let (proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();

        // A vec to hold all the stream items we don't care about for this test.
        let mut junk_drawer = vec![];

        // Filter out all items that are irrelevant to this particular test, placing them in
        // the junk_drawer.
        let mut stream = stream.filter_map(move |item| {
            let item = match item {
                Ok(PeerHandlerRequest::WatchNetworkInformation { responder }) => Some(responder),
                x => {
                    junk_drawer.push(x);
                    None
                }
            };
            ready(item)
        });

        // Pass in the client end connected to the call manager
        let result = exec.run_singlethreaded(sender.send(PeerRequest::Handle(proxy)));
        assert!(result.is_ok());

        // The stream should produce the network update while the peer runs in the background.
        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // Send the first network update - should be relayed to the peer.
        let (responder, run_fut) = run_while(&mut exec, run_fut, stream.next());
        responder.unwrap().send(network_update_1).expect("Successfully send network information");
        let ((), run_fut) = run_while(
            &mut exec,
            run_fut,
            expect_data_received_by_peer(&mut remote, expected_data1),
        );

        // Send the second network update - should be relayed to the peer.
        let (responder, run_fut) = run_while(&mut exec, run_fut, stream.next());
        responder.unwrap().send(network_update_2).expect("Successfully send network information");
        let ((), mut run_fut) = run_while(
            &mut exec,
            run_fut,
            expect_data_received_by_peer(&mut remote, expected_data2),
        );

        // Drop the peer task sender to force the PeerTask's run future to complete
        drop(sender);
        let task = exec.run_singlethreaded(&mut run_fut);

        // Check that the task's network information contains the expected values
        // based on the updates provided by the call manager task.
        assert_eq!(task.network, expected_network);
    }

    #[fuchsia::test]
    fn terminated_slc_ends_peer_task() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        let (connection, remote) = create_and_initialize_slc(SlcState::default());
        let (peer, _sender, receiver, _profile) = setup_peer_task(Some(connection));

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // The peer task is pending with no futher work to do at this time.
        let result = exec.run_until_stalled(&mut run_fut);
        assert!(result.is_pending());

        // Closing the SLC connection will result in the completion of the peer task.
        drop(remote);

        let result = exec.run_until_stalled(&mut run_fut);
        let peer = result.expect("run to complete");
        assert!(peer.connection.is_terminated());
    }

    #[fuchsia::test]
    fn error_in_slc_ends_peer_task() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        let (connection, remote) = create_and_initialize_slc(SlcState::default());
        let (peer, _sender, receiver, _profile) = setup_peer_task(Some(connection));

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // Produces an error when polling the ServiceLevelConnection stream by disabling write
        // on the remote socket and read on the local socket.
        let status = unsafe {
            zx::sys::zx_socket_set_disposition(
                remote.as_ref().raw_handle(),
                zx::sys::ZX_SOCKET_DISPOSITION_WRITE_DISABLED,
                0,
            )
        };
        zx::Status::ok(status).unwrap();

        // Error on the SLC connection will result in the completion of the peer task.
        let result = exec.run_until_stalled(&mut run_fut);
        assert!(result.is_ready());
    }

    /// Transform `stream` into a Stream of WatchNextCall responders.
    #[track_caller]
    async fn wait_for_call_stream(
        stream: PeerHandlerRequestStream,
    ) -> impl Stream<Item = PeerHandlerWatchNextCallResponder> {
        filtered_stream(stream, |item| match item {
            PeerHandlerRequest::WatchNextCall { responder } => Ok(responder),
            x => Err(x),
        })
        .await
    }

    /// Transform `stream` into a Stream of `T`.
    ///
    /// `f` is a function that takes a PeerHandlerRequest and either returns an Ok if the request
    /// is relevant to the test or Err if the request irrelevant.
    ///
    /// This test helper function can be used in the common case where the test interacts
    /// with a particular kind of PeerHandlerRequest.
    ///
    /// Initial setup of the handler is done, then a filtered stream is produced which
    /// outputs items based on the result of `f`. Ok return values from `f` are returned from the
    /// `filtered_stream`. Err return values from `f` are not returned from the `filtered_stream`.
    /// Instead they are stored within `filtered_stream` until `filtered_stream` is dropped
    /// so that they do not cause the underlying fidl channel to be closed.
    #[track_caller]
    async fn filtered_stream<T>(
        mut stream: PeerHandlerRequestStream,
        f: impl Fn(PeerHandlerRequest) -> Result<T, PeerHandlerRequest>,
    ) -> impl Stream<Item = T> {
        // Send the network information immediately so the peer can make progress.
        match stream.next().await {
            Some(Ok(PeerHandlerRequest::WatchNetworkInformation { responder })) => {
                responder
                    .send(NetworkInformation::EMPTY)
                    .expect("Successfully send network information");
            }
            x => panic!("Expected watch network information request: {:?}", x),
        };

        // A vec to hold all the stream items we don't care about for this test.
        let mut junk_drawer = vec![];

        // Filter out all items, placing them in the junk_drawer.
        stream.filter_map(move |item| {
            let item = match item {
                Ok(item) => match f(item) {
                    Ok(t) => Some(t),
                    Err(x) => {
                        junk_drawer.push(x);
                        None
                    }
                },
                _ => None,
            };
            ready(item)
        })
    }

    #[fuchsia::test]
    fn call_updates_update_ringer_state() {
        // Set up the executor, peer, and background call manager task
        let mut exec = fasync::TestExecutor::new().unwrap();

        // Setup the peer task with the specified SlcState to enable indicator events.
        let state = SlcState {
            ag_indicator_events_reporting: AgIndicatorsReporting::new_enabled(),
            ..SlcState::default()
        };
        let (connection, mut remote) = create_and_initialize_slc(state);
        let (peer, mut sender, receiver, _profile) = setup_peer_task(Some(connection));

        let (proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();

        // Pass in the client end connected to the call manager
        let result = exec.run_singlethreaded(sender.send(PeerRequest::Handle(proxy)));
        assert!(result.is_ok());

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);
        let (mut stream, run_fut) = run_while(&mut exec, run_fut, wait_for_call_stream(stream));

        // Send the incoming call
        let (responder, run_fut) = run_while(&mut exec, run_fut, stream.next());
        let (client_end, _call_stream) = fidl::endpoints::create_request_stream().unwrap();
        let next_call = NextCall {
            call: Some(client_end),
            remote: Some("1234567".to_string()),
            state: Some(CallState::IncomingRinging),
            direction: Some(CallDirection::MobileTerminated),
            ..NextCall::EMPTY
        };
        responder.unwrap().send(next_call).expect("Successfully send call information");

        let expected_data = vec![AgIndicator::CallSetup(1).into()];
        let ((), mut run_fut) =
            run_while(&mut exec, run_fut, expect_data_received_by_peer(&mut remote, expected_data));

        // Drop the peer task sender to force the PeerTask's run future to complete
        drop(sender);
        let task = exec.run_until_stalled(&mut run_fut).expect("run_fut to complete");

        // Check that the task's ringer has an active call with the expected call index.
        assert!(task.ringer.ringing());
    }

    #[fuchsia::test]
    fn transfers_change_sco_state() {
        // Set up the executor, peer, and background call manager task
        info!("transfers_change_sco_state: Creating executer.");
        let mut exec = fasync::TestExecutor::new().unwrap();

        // Setup the peer task.
        info!("transfers_change_sco_state: Creating SLC.");
        let (connection, _remote) = create_and_initialize_slc(SlcState::default());
        info!("transfers_change_sco_state: Creating peer task.");
        let (peer, mut sender, receiver, mut profile) = setup_peer_task(Some(connection));

        info!("transfers_change_sco_state: Creating peer handler proxy.");
        let (proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();

        // Pass in the client end connected to the call manager
        info!("transfers_change_sco_state: Sending peer handler proxy.");
        let result = exec.run_singlethreaded(sender.send(PeerRequest::Handle(proxy)));
        assert!(result.is_ok());

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);
        info!("transfers_change_sco_state: Receiving peer handler stream.");
        let (mut stream, run_fut) = run_while(&mut exec, run_fut, wait_for_call_stream(stream));

        // Send the incoming call
        info!("transfers_change_sco_state: Getting call stream responder.");
        let (responder, run_fut) = run_while(&mut exec, run_fut, stream.next());
        info!("transfers_change_sco_state: Creating call stream.");
        let (client_end, mut call_stream) = fidl::endpoints::create_request_stream().unwrap();
        let next_call = NextCall {
            call: Some(client_end),
            remote: Some("1234567".to_string()),
            state: Some(CallState::IncomingRinging),
            direction: Some(CallDirection::MobileTerminated),
            ..NextCall::EMPTY
        };
        info!("transfers_change_sco_state: Sending next call.");
        responder.unwrap().send(next_call).expect("Successfully send call information");

        // Answer call.
        info!("transfers_change_sco_state: Awaiting watch state to answer call.");
        let (request, run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        let state_responder = match request {
            Some(Ok(CallRequest::WatchState { responder })) => responder,
            req => panic!("Expected WatchState, got {:?}", req),
        };
        info!("transfers_change_sco_state: Sending OngoingActive to initially answer call.");
        state_responder.send(CallState::OngoingActive).expect("Sent OngoingActive.");
        info!("transfers_change_sco_state: Setting up SCO connection.");
        let (sco, mut run_fut) =
            run_while(&mut exec, run_fut, expect_sco_connection(&mut profile, true, Ok(())));
        info!("transfers_change_sco_state: Setting up SCO connection--pausing audio.");
        while let None = exec.run_one_step(&mut run_fut) {}
        let mut sco = sco.unwrap();

        // Call is transferred to AG by AG and SCO is torn down.
        info!("transfers_change_sco_state: Getting WatchState for transferring call to AG by AG part 1.");
        let (request, run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        let state_responder = match request {
            Some(Ok(CallRequest::WatchState { responder })) => responder,
            req => panic!("Expected WatchState, got {:?}", req),
        };
        info!("transfers_change_sco_state: Sending TransferedToAg.");
        state_responder.send(CallState::TransferredToAg).expect("Sent TransferredToAg.");
        info!("transfers_change_sco_state: Transferring call to AG by AG.");
        let (request, mut run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        info!("transfers_change_sco_state: Transferring call to AG by AG--unpausing audio.");
        while let None = exec.run_one_step(&mut run_fut) {}
        info!("transfers_change_sco_state: Checking if SCO is closed.");
        let (sco_result, run_fut) = run_while(&mut exec, run_fut, &mut sco.next());
        assert_matches!(sco_result, None);

        // Call is transferred to HF by AG and SCO is set up.
        let state_responder = match request {
            Some(Ok(CallRequest::WatchState { responder })) => responder,
            req => panic!("Expected WatchState, got {:?}", req),
        };
        info!("transfers_change_sco_state: Sending OngoingActive.");
        state_responder.send(CallState::OngoingActive).expect("Sent OngoingActive.");
        // Don't send a SCO connection until they are trying to connect.
        info!("transfers_change_sco_state: Transferring call to HF by AG.");
        let (sco, mut run_fut) =
            run_while(&mut exec, run_fut, expect_sco_connection(&mut profile, true, Ok(())));
        info!("transfers_change_sco_state: Transferring call to HF by AG--pausing audio.");
        let _ = exec.run_one_step(&mut run_fut);
        let sco = sco.expect("SCO Connection.");
        // Run until the connection is handled by the task.  This avoids a race where the
        // the incoming SCO connection is closed before it's received, in which case a call
        // is never set to active.
        info!("transfers_change_sco_state: Finishing transferring call to HF by AG.");
        let _ = exec.run_one_step(&mut run_fut);

        // SCO is torn down by HF and call is transferred to AG
        info!("transfers_change_sco_state: Dropping SCO.");
        drop(sco);
        info!("transfers_change_sco_state: Getting WatchState for transferring to AG.");
        let (watch_state_req, run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        info!("transfers_change_sco_state: Getting RequestTransferAudio.");
        let (req, run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        assert_matches!(req, Some(Ok(CallRequest::RequestTransferAudio { .. })));
        let state_responder = match watch_state_req {
            Some(Ok(CallRequest::WatchState { responder })) => responder,
            req => panic!("Expected WatchState, got {:?}", req),
        };
        info!("transfers_change_sco_state: Sending TransferredToAg.");
        state_responder.send(CallState::TransferredToAg).expect("Sent TransferredToAg");

        // SCO is set up by HF and call is transferred to HF
        info!("transfers_change_sco_state: Expecting SCO set up by HF.");
        let (_sco, mut run_fut) =
            run_while(&mut exec, run_fut, expect_sco_connection(&mut profile, false, Ok(())));
        info!("transfers_change_sco_state: Expected SCO set up by HF--pausing audio.");
        while let None = exec.run_one_step(&mut run_fut) {}
        info!("transfers_change_sco_state: Getting WatchState.");
        let (_watch_state_req, run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        info!("transfers_change_sco_state: Getting RequestActive.");
        let (req, mut run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        assert_matches!(req, Some(Ok(CallRequest::RequestActive { .. })));

        // Drop the peer task sender to force the PeerTask's run future to complete
        info!("transfers_change_sco_state: Dropping sender.");
        drop(sender);
        info!("transfers_change_sco_state: Finishing.");
        while let None = exec.run_one_step(&mut run_fut) {}
        info!("transfers_change_sco_state: Finished.");
    }

    #[fuchsia::test]
    fn incoming_hf_indicator_battery_level_is_propagated_to_peer_handler_stream() {
        // Set up the executor, peer, and background call manager task
        let mut exec = fasync::TestExecutor::new().unwrap();

        // Setup the peer task with the specified SlcState to enable the battery level HF indicator.
        let mut hf_indicators = HfIndicators::default();
        hf_indicators.enable_indicators(vec![at::BluetoothHFIndicator::BatteryLevel]);
        let state = SlcState { hf_indicators, ..SlcState::default() };
        let (connection, mut remote) = create_and_initialize_slc(state);
        let (peer, mut sender, receiver, _profile) = setup_peer_task(Some(connection));

        let (proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();
        // The battery level that will be reported by the peer.
        let expected_level = 4;

        // Pass in the client end connected to the call manager
        let result = exec.run_singlethreaded(sender.send(PeerRequest::Handle(proxy)));
        assert!(result.is_ok());

        // Run the PeerTask.
        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        let headset_level_stream_fut = filtered_stream(stream, |item| match item {
            PeerHandlerRequest::ReportHeadsetBatteryLevel { level, .. } => Ok(level),
            x => Err(x),
        });

        let (mut stream, run_fut) = run_while(&mut exec, run_fut, headset_level_stream_fut);

        // Peer sends us a battery level HF indicator update.
        let battery_level_cmd = at::Command::Biev {
            anum: at::BluetoothHFIndicator::BatteryLevel,
            value: expected_level as i64,
        };
        let mut buf = Vec::new();
        at::Command::serialize(&mut buf, &vec![battery_level_cmd]).expect("serialization is ok");
        let _ = remote.as_ref().write(&buf[..]).expect("channel write is ok");

        // Run the main future - the task should receive the HF indicator and report it.
        let (battery_level, _run_fut) = run_while(&mut exec, run_fut, stream.next());
        assert_eq!(battery_level, Some(expected_level));

        // Since we (the AG) received a valid HF indicator, we expect to send an OK back to the peer.
        expect_peer_ready(&mut exec, &mut remote, Some(serialize_at_response(at::Response::Ok)));
    }

    #[fuchsia::test]
    fn local_battery_level_change_initiates_phone_status_procedure() {
        let mut exec = fasync::TestExecutor::new().unwrap();

        // Setup the peer task with the specified SlcState to enable the battery level indicator on
        // both the HF and the AG.
        let mut hf_indicators = HfIndicators::default();
        hf_indicators.enable_indicators(vec![at::BluetoothHFIndicator::BatteryLevel]);
        let ag_indicator_events_reporting = AgIndicatorsReporting::new_enabled();
        let state =
            SlcState { hf_indicators, ag_indicator_events_reporting, ..SlcState::default() };
        let (connection, mut remote) = create_and_initialize_slc(state);
        let (peer, mut sender, receiver, _profile) = setup_peer_task(Some(connection));

        // Run the PeerTask.
        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // Receive a local update from the Fuchsia Battery Manager about a battery level change.
        let request = PeerRequest::BatteryLevel(3);
        let send_request_fut = sender.send(request);
        let (send_result, run_fut) = run_while(&mut exec, run_fut, send_request_fut);
        assert_matches!(send_result, Ok(()));

        // We expect the peer (HF) to receive the PhoneStatus update AT command.
        let expected_ciev = vec![AgIndicator::BatteryLevel(3).into()];
        let ((), _run_fut) =
            run_while(&mut exec, run_fut, expect_data_received_by_peer(&mut remote, expected_ciev));
    }

    #[fuchsia::test]
    fn call_updates_produce_call_waiting() {
        // Set up the executor, peer, and background call manager task
        let mut exec = fasync::TestExecutor::new().unwrap();

        let raw_number = "1234567";
        let number = Number::from(raw_number);
        let expected_ccwa =
            vec![at::success(at::Success::Ccwa { ty: number.type_(), number: number.into() })];
        let expected_ciev = vec![AgIndicator::CallSetup(1).into()];

        // Setup the peer task with the specified SlcState to enable indicator events.
        let state = SlcState {
            call_waiting_notifications: true,
            ag_indicator_events_reporting: AgIndicatorsReporting::new_enabled(),
            ..SlcState::default()
        };
        let (connection, mut remote) = create_and_initialize_slc(state);
        let (peer, mut sender, receiver, _profile) = setup_peer_task(Some(connection));

        let (proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // Pass in the client end connected to the call manager
        let result = exec.run_singlethreaded(sender.send(PeerRequest::Handle(proxy)));
        assert!(result.is_ok());

        let (mut stream, run_fut) = run_while(&mut exec, run_fut, wait_for_call_stream(stream));

        // Send the incoming waiting call.
        let (responder, run_fut) = run_while(&mut exec, run_fut, stream.next());
        let (client_end, _call_stream) = fidl::endpoints::create_request_stream().unwrap();
        let next_call = NextCall {
            call: Some(client_end),
            remote: Some(raw_number.to_string()),
            state: Some(CallState::IncomingWaiting),
            direction: Some(CallDirection::MobileTerminated),
            ..NextCall::EMPTY
        };
        responder.unwrap().send(next_call).expect("Successfully send call information");

        let ((), run_fut) =
            run_while(&mut exec, run_fut, expect_data_received_by_peer(&mut remote, expected_ccwa));
        let ((), mut run_fut) =
            run_while(&mut exec, run_fut, expect_data_received_by_peer(&mut remote, expected_ciev));

        // Drop the peer task sender to force the PeerTask's run future to complete
        drop(sender);
        let _ = exec.run_until_stalled(&mut run_fut).expect("run_fut to complete");
    }

    #[fuchsia::test]
    fn outgoing_call_holds_active() {
        // Set up the executor.
        let mut exec = fasync::TestExecutor::new().unwrap();

        // Setup the peer task with the specified SlcState to enable three way calling.
        let mut ag_features = AgFeatures::default();
        ag_features.set(AgFeatures::THREE_WAY_CALLING, true);
        let mut hf_features = HfFeatures::default();
        hf_features.set(HfFeatures::THREE_WAY_CALLING, true);
        let state = SlcState { ag_features, hf_features, ..SlcState::default() };
        let (connection, remote) = create_and_initialize_slc(state);
        let (peer, mut sender, receiver, mut profile) = setup_peer_task(Some(connection));

        let (proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // Pass in the client end connected to the call manager
        exec.run_singlethreaded(sender.send(PeerRequest::Handle(proxy))).expect("Connecting peer");

        let (mut stream, run_fut) = run_while(&mut exec, run_fut, wait_for_call_stream(stream));

        // Send the incoming waiting call.
        let (responder, run_fut) = run_while(&mut exec, run_fut, stream.next());
        let (client_end, mut call_stream) = fidl::endpoints::create_request_stream().unwrap();
        let next_call = NextCall {
            call: Some(client_end),
            remote: Some("1234567".to_string()),
            state: Some(CallState::IncomingWaiting),
            direction: Some(CallDirection::MobileTerminated),
            ..NextCall::EMPTY
        };
        responder.unwrap().send(next_call).expect("Successfully send call information");

        // Answer call.
        let (request, run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        let state_responder = match request {
            Some(Ok(CallRequest::WatchState { responder })) => responder,
            req => panic!("Expected WatchState, got {:?}", req),
        };
        state_responder.send(CallState::OngoingActive).expect("Sent OngoingActive.");
        let (sco, run_fut) =
            run_while(&mut exec, run_fut, expect_sco_connection(&mut profile, true, Ok(())));
        let _sco = sco.expect("SCO Connection.");

        // Make outgoing call from HF.
        let dial_cmd = at::Command::AtdNumber { number: String::from("7654321") };
        let mut buf = Vec::new();
        at::Command::serialize(&mut buf, &vec![dial_cmd]).expect("serialization is ok");
        let _ = remote.as_ref().write(&buf[..]).expect("channel write is ok");

        let (watch_state_req, run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        let _watch_state_resp = match watch_state_req {
            Some(Ok(CallRequest::WatchState { responder, .. })) => responder,
            req => panic!("Expected WatchState, got {:?}", req),
        };

        // Receive hold request from first call.
        let (hold_req, _run_fut) = run_while(&mut exec, run_fut, &mut call_stream.next());
        match hold_req {
            Some(Ok(CallRequest::RequestHold { .. })) => {}
            req => panic!("Expected RequestHold, got {:?}", req),
        };
    }

    #[fuchsia::test]
    fn connection_behavior_request_updates_state() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        let (peer, mut sender, receiver, mut profile) = setup_peer_task(None);

        let _peer_task = fasync::Task::local(peer.run(receiver));

        // First check that a connection is made when search results are received.
        // Send a valid search result.
        let random_channel_number = ServerChannel::try_from(4).expect("valid server channel");
        let protocol =
            Some(build_rfcomm_protocol(random_channel_number).iter().map(Into::into).collect());
        let event = ProfileEvent::SearchResult {
            id: PeerId(1),
            protocol: protocol.clone(),
            attributes: vec![],
        };
        exec.run_singlethreaded(sender.send(PeerRequest::Profile(event))).expect("Send to succeed");

        // Get a connection request on the `profile` stream.
        let result = exec.run_until_stalled(&mut profile.next());
        let _channel = match result {
            Poll::Ready(Some(Ok(bredr::ProfileRequest::Connect {
                connection:
                    bredr::ConnectParameters::Rfcomm(bredr::RfcommParameters {
                        channel: Some(sc), ..
                    }),
                responder,
                ..
            }))) => {
                assert_eq!(sc, u8::from(random_channel_number));
                let (local, remote) = Channel::create();
                let local = local.try_into().unwrap();
                responder.send(&mut Ok(local)).unwrap();

                remote
            }
            x => panic!("unexpected result: {:?}", x),
        };

        // Disable auto connect behavior
        exec.run_singlethreaded(
            sender.send(PeerRequest::Behavior(ConnectionBehavior { autoconnect: false })),
        )
        .expect("Send to succeed");

        // Connection is not made after autoconnect is disabled
        // Send search results.
        let event = ProfileEvent::SearchResult { id: PeerId(1), protocol, attributes: vec![] };
        exec.run_singlethreaded(sender.send(PeerRequest::Profile(event))).expect("Send to succeed");

        // No request is received on stream.
        let result = exec.run_until_stalled(&mut profile.next());
        assert!(result.is_pending());
    }

    #[fuchsia::test]
    fn non_rfcomm_search_result_is_ignored() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        let (peer, mut sender, receiver, mut profile) = setup_peer_task(None);

        let _peer_task = fasync::Task::local(peer.run(receiver));

        // No connection should be made for a non RFCOMM search result.
        // Send a valid search result with some random L2CAP protocol.
        let protocol = vec![bredr::ProtocolDescriptor {
            protocol: bredr::ProtocolIdentifier::L2Cap,
            params: vec![bredr::DataElement::Uint16(25)],
        }];
        let event = ProfileEvent::SearchResult {
            id: PeerId(1),
            protocol: Some(protocol),
            attributes: vec![],
        };
        exec.run_singlethreaded(sender.send(PeerRequest::Profile(event))).expect("Send to succeed");

        // No connection request on the `profile` stream.
        assert!(exec.run_until_stalled(&mut profile.next()).is_pending());
    }

    #[fuchsia::test]
    fn connect_request_triggers_connection() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        let connection = ServiceLevelConnection::new();
        let (local, mut remote) = Channel::create();
        let (peer, mut sender, receiver, _profile) = setup_peer_task(Some(connection));

        assert!(!peer.connection.connected());

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        let event_fut = sender.send(PeerRequest::Profile(ProfileEvent::PeerConnected {
            id: PeerId(0),
            protocol: vec![],
            channel: local,
        }));
        exec.run_singlethreaded(event_fut).unwrap();

        // The peer task is pending with no further work to do at this time.
        let result = exec.run_until_stalled(&mut run_fut);
        assert!(result.is_pending());

        // Closing the SLC connection will result in the completion of the peer task.
        drop(sender);

        let result = exec.run_until_stalled(&mut run_fut);
        let peer = result.expect("run to complete");
        assert!(peer.connection.connected());
        assert!(exec.run_until_stalled(&mut remote.next()).is_pending());
    }

    #[fuchsia::test]
    fn connect_request_replaces_connection() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        // SLC is connected at the start of the test.
        let (connection, mut old_remote) = create_and_connect_slc();
        let (peer, mut sender, receiver, _profile) = setup_peer_task(Some(connection));

        assert!(peer.connection.connected());

        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // create a new connection for the SLC
        let (local, mut new_remote) = Channel::create();
        let event_fut = sender.send(PeerRequest::Profile(ProfileEvent::PeerConnected {
            id: PeerId(0),
            protocol: vec![],
            channel: local,
        }));
        exec.run_singlethreaded(event_fut).unwrap();

        // The peer task is pending with no further work to do at this time.
        let result = exec.run_until_stalled(&mut run_fut);
        assert!(result.is_pending());

        // Closing the SLC connection will result in the completion of the peer task.
        drop(sender);

        let result = exec.run_until_stalled(&mut run_fut);
        let peer = result.expect("run to complete");
        assert!(peer.connection.connected());
        let result = exec.run_until_stalled(&mut old_remote.next());
        // old_remote is closed
        assert_matches!(result, Poll::Ready(None));
        // new_remote is open
        assert!(exec.run_until_stalled(&mut new_remote.next()).is_pending());
    }

    /// Run a PeerTask until it has no more work left to do, then return it.
    /// This function generates its own PeerTask channel for convenience. This means that
    /// the channel cannot be used to send messages to a running PeerTask.
    #[track_caller]
    fn run_peer_until_stalled(exec: &mut fasync::TestExecutor, peer: PeerTask) -> PeerTask {
        let (_sender, receiver) = mpsc::channel(0);
        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);
        exec.run_until_stalled(&mut run_fut)
            .expect_pending("shouldn't be done while _sender is live");
        drop(_sender);
        exec.run_until_stalled(&mut run_fut).unwrap()
    }

    async fn expect_sco_connection(
        profile_requests: &mut ProfileRequestStream,
        expected_initiator: bool,
        result: Result<(), ScoErrorCode>,
    ) -> Option<bredr::ScoConnectionRequestStream> {
        // Sometimes dropping the SCO connection accept future doesn't cancel the
        // existing request before we want to connect a new one.  In this case, we
        // may get *two* requests for a SCO connection, and we want the second one,
        // which has the expected direction.
        loop {
            let proxy = match profile_requests.next().await.expect("request").unwrap() {
                bredr::ProfileRequest::ConnectSco { receiver, params, initiator, .. } => {
                    if initiator != expected_initiator {
                        continue;
                    };
                    assert!(params.len() >= 1);
                    receiver.into_proxy().unwrap()
                }
                x => panic!("Unexpected request to profile stream: {:?}", x),
            };
            match result {
                Ok(()) => {
                    let (client, request_stream) =
                        fidl::endpoints::create_request_stream::<bredr::ScoConnectionMarker>()
                            .expect("request stream");
                    let default_params = bredr::ScoConnectionParameters {
                        parameter_set: Some(bredr::HfpParameterSet::CvsdD1),
                        air_coding_format: Some(bredr::CodingFormat::Cvsd),
                        air_frame_size: Some(60),
                        io_bandwidth: Some(16000),
                        io_coding_format: Some(bredr::CodingFormat::LinearPcm),
                        io_frame_size: Some(16),
                        io_pcm_data_format: Some(
                            fidl_fuchsia_hardware_audio::SampleFormat::PcmSigned,
                        ),
                        path: Some(bredr::DataPath::Offload),
                        ..bredr::ScoConnectionParameters::EMPTY
                    };
                    proxy.connected(client, default_params).unwrap();
                    return Some(request_stream);
                }
                Err(code) => {
                    proxy.error(code).unwrap();
                    return None;
                }
            }
        }
    }

    /// Setup a new audio connection between the PeerTask and an upstream ProfileRequestStream.
    /// This helper function asserts that the SCO connection is created and that the audio
    /// connection has started up.
    #[track_caller]
    fn setup_audio(
        exec: &mut fasync::TestExecutor,
        peer: &mut PeerTask,
        profile_requests: &mut ProfileRequestStream,
    ) -> bredr::ScoConnectionRequestStream {
        let codecs = peer
            .connection
            .get_selected_codec()
            .map_or(vec![CodecId::MSBC, CodecId::CVSD], |c| vec![c]);
        let sco_connector = peer.sco_connector.clone();
        let audio_connection_fut = sco_connector.connect(peer.id.clone(), codecs).fuse();
        pin_mut!(audio_connection_fut);

        exec.run_until_stalled(&mut audio_connection_fut).expect_pending("shouldn't be done yet");

        // Expect a sco connection, and have it succeed.
        let sco_complete_fut = expect_sco_connection(profile_requests, true, Ok(()));
        pin_mut!(sco_complete_fut);
        let result = exec.run_singlethreaded(&mut futures::future::select(
            audio_connection_fut,
            sco_complete_fut,
        ));

        let (remote_sco, mut audio_connection_fut) = match result {
            Either::Right(r) => r,
            Either::Left(_) => panic!("Audio connection future shouldn't have finished"),
        };

        let res = exec.run_until_stalled(&mut audio_connection_fut).expect("should be done");
        let local_sco = res.expect("should have started up okay");
        let audio_connection_fut2 = peer.finish_sco_connection(local_sco);
        pin_mut!(audio_connection_fut2);
        exec.run_singlethreaded(&mut audio_connection_fut2).expect("finished");

        remote_sco.unwrap()
    }

    #[fuchsia::test]
    fn setup_audio_connection_connects_and_starts_audio() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        // SLC is connected at the start of the test.
        let (connection, _old_remote) = create_and_connect_slc();
        let (mut peer, _sender, _receiver, mut profile_requests) =
            setup_peer_task(Some(connection));

        assert!(peer.connection.connected());

        let audio_control = peer.audio_control.clone();

        let _remote_sco = setup_audio(&mut exec, &mut peer, &mut profile_requests);

        // Should have started up the test audio control. Test by trying to start it again, it
        // should be an error.
        {
            let mut lock = audio_control.lock();
            let _ = lock
                .start(PeerId(0), bredr::ScoConnectionParameters::EMPTY)
                .expect_err("shouldn't be able to start, already started");
        }
    }

    #[fuchsia::test]
    fn audio_is_stopped_when_sco_connection_closes() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        // SLC is connected at the start of the test.
        let (connection, _old_remote) = create_and_connect_slc();
        let (mut peer, _sender, receiver, mut profile_requests) = setup_peer_task(Some(connection));

        assert!(peer.connection.connected());

        let audio_control = peer.audio_control.clone();

        let remote_sco = setup_audio(&mut exec, &mut peer, &mut profile_requests);

        // Should have started up the test audio control.
        {
            let mut lock = audio_control.lock();
            let _ = lock
                .start(PeerId(0), bredr::ScoConnectionParameters::EMPTY)
                .expect_err("shouldn't be able to start, already started");
        }

        // Set up the run task.
        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);
        let _ = exec.run_until_stalled(&mut run_fut);

        drop(remote_sco);

        // Spin the run task to notice that the SCO connection has failed, and drop
        // the audio / stop the audio.
        let _ = exec.run_until_stalled(&mut run_fut);

        // Should have stopped the audio - check by trying to stop it again, it should be an error
        let mut lock = audio_control.lock();
        let _ = lock.stop().expect_err("should already be stopped");
    }

    #[fuchsia::test]
    fn sco_connection_closed_when_call_ends() {
        let mut exec = fasync::TestExecutor::new().unwrap();
        // SLC is connected at the start of the test.
        let (connection, _old_remote) = create_and_connect_slc();
        let (mut peer, _sender, _receiver, mut profile_requests) =
            setup_peer_task(Some(connection));

        assert!(peer.connection.connected());

        let _remote_sco = setup_audio(&mut exec, &mut peer, &mut profile_requests);

        // Run the PeerTask to handle all audio setup tasks
        let peer = run_peer_until_stalled(&mut exec, peer);

        assert!(peer.sco_state.is_active());

        // Create the Call Manager side of a PeerHandler to send a call state update to the
        // `OngoingHeld` state in order to tear down the active sco connection.
        // Once call_manager does the work required by the test, it completes, returning items that
        // should be kept alive for the remainder of the test.
        async fn call_manager(
            stream: PeerHandlerRequestStream,
        ) -> (impl Stream<Item = PeerHandlerWatchNextCallResponder>, CallRequestStream) {
            let mut stream = wait_for_call_stream(stream).await;

            // Send the held call response
            let responder = stream.next().await.unwrap();
            let (client_end, call_stream) = fidl::endpoints::create_request_stream().unwrap();
            let next_call = NextCall {
                call: Some(client_end),
                remote: Some("1234567".to_string()),
                state: Some(CallState::OngoingHeld),
                direction: Some(CallDirection::MobileTerminated),
                ..NextCall::EMPTY
            };
            responder.send(next_call).expect("Successfully send call information");

            (stream, call_stream)
        }

        let (proxy, stream) =
            fidl::endpoints::create_proxy_and_stream::<PeerHandlerMarker>().unwrap();
        let call_manager_fut = call_manager(stream);
        pin_mut!(call_manager_fut);

        let (mut sender, receiver) = mpsc::channel(0);
        // Wire up the HFP side of PeerHandler by passing the proxy into the PeerTask.
        let handle_fut = sender.send(PeerRequest::Handle(proxy));

        // Join the futures that are responsible for sending events _into_ the PeerTask.
        let join_fut = futures::future::join(call_manager_fut, handle_fut);

        // Create the run future to drive the PeerTask forward.
        let run_fut = peer.run(receiver);
        pin_mut!(run_fut);

        // Run until `join_fut` completes.
        let (_stream, mut run_fut) = run_while(&mut exec, run_fut, join_fut);

        // Make sure all work that is pending in the PeerTask `run` future completes,
        // forcing the future to complete and return the `PeerTask` object.
        assert!(exec.run_until_stalled(&mut run_fut).is_pending());
        // Force `run_fut` to complete by dropping the `sender`.
        drop(sender);
        let peer = exec.run_singlethreaded(run_fut);

        // The SCO connection should be removed after the PeerTask has handled the Terminated call
        // update.
        assert!(!peer.sco_state.is_active());
    }
}