sat-rs/satrs/tests/pus_verification.rs

#[cfg(feature = "crossbeam")]
pub mod crossbeam_test {
    use hashbrown::HashMap;
    use satrs::pool::{PoolProvider, PoolProviderWithGuards, StaticMemoryPool, StaticPoolConfig};
    use satrs::pus::verification::{
        FailParams, RequestId, VerificationReporter, VerificationReporterCfg,
        VerificationReportingProvider,
    };
    use satrs::pus::TmInSharedPoolSenderWithCrossbeam;
    use satrs::request::UniqueApidTargetId;
    use satrs::tmtc::tm_helper::SharedTmPool;
    use spacepackets::ecss::tc::{PusTcCreator, PusTcReader, PusTcSecondaryHeader};
    use spacepackets::ecss::tm::PusTmReader;
    use spacepackets::ecss::{EcssEnumU16, EcssEnumU8, PusPacket, WritablePusPacket};
    use spacepackets::SpHeader;
    use std::sync::{Arc, RwLock};
    use std::thread;
    use std::time::Duration;

    const TEST_APID: u16 = 0x03;
    const TEST_ID: TargetAndApidId = TargetAndApidId::new(TEST_APID, 0x05);

    const FIXED_STAMP: [u8; 7] = [0; 7];
    const PACKETS_SENT: u8 = 8;

    /// This test also shows how the verification report could be used in a multi-threaded context,
    /// wrapping it into an [Arc] and [Mutex] and then passing it to two threads.
    ///
    ///  - The first thread generates a acceptance, a start, two steps and one completion report
    ///  - The second generates an acceptance and start success report and a completion failure
    ///  - The third thread is the verification receiver. In the test case, it verifies the other two
    ///    threads have sent the correct expected verification reports
    #[test]
    fn test_shared_reporter() {
        // We use a synced sequence count provider here because both verification reporters have the
        // the same APID. If they had distinct APIDs, the more correct approach would be to have
        // each reporter have an own sequence count provider.
        let cfg = VerificationReporterCfg::new(TEST_APID, 1, 2, 8).unwrap();
        // Shared pool object to store the verification PUS telemetry
        let pool_cfg =
            StaticPoolConfig::new(vec![(10, 32), (10, 64), (10, 128), (10, 1024)], false);
        let shared_tm_pool = SharedTmPool::new(StaticMemoryPool::new(pool_cfg.clone()));
        let shared_tc_pool_0 = Arc::new(RwLock::new(StaticMemoryPool::new(pool_cfg)));
        let shared_tc_pool_1 = shared_tc_pool_0.clone();
        let (tx, rx) = crossbeam_channel::bounded(10);
        let sender_0 = TmInSharedPoolSenderWithCrossbeam::new(shared_tm_pool.clone(), tx.clone());
        let sender_1 = sender_0.clone();
        let mut reporter_with_sender_0 = VerificationReporter::new(&cfg);
        let mut reporter_with_sender_1 = reporter_with_sender_0.clone();
        // For test purposes, we retrieve the request ID from the TCs and pass them to the receiver
        // tread.
        let req_id_0;
        let req_id_1;

        let (tx_tc_0, rx_tc_0) = crossbeam_channel::bounded(3);
        let (tx_tc_1, rx_tc_1) = crossbeam_channel::bounded(3);
        {
            let mut tc_guard = shared_tc_pool_0.write().unwrap();
            let mut sph = SpHeader::tc_unseg(TEST_APID, 0, 0).unwrap();
            let tc_header = PusTcSecondaryHeader::new_simple(17, 1);
            let pus_tc_0 = PusTcCreator::new_no_app_data(&mut sph, tc_header, true);
            req_id_0 = RequestId::new(&pus_tc_0);
            let addr = tc_guard
                .free_element(pus_tc_0.len_written(), |buf| {
                    pus_tc_0.write_to_bytes(buf).unwrap();
                })
                .unwrap();
            tx_tc_0.send(addr).unwrap();
            let mut sph = SpHeader::tc_unseg(TEST_APID, 1, 0).unwrap();
            let tc_header = PusTcSecondaryHeader::new_simple(5, 1);
            let pus_tc_1 = PusTcCreator::new_no_app_data(&mut sph, tc_header, true);
            req_id_1 = RequestId::new(&pus_tc_1);
            let addr = tc_guard
                .free_element(pus_tc_0.len_written(), |buf| {
                    pus_tc_1.write_to_bytes(buf).unwrap();
                })
                .unwrap();
            tx_tc_1.send(addr).unwrap();
        }
        let verif_sender_0 = thread::spawn(move || {
            let mut tc_buf: [u8; 1024] = [0; 1024];
            let tc_addr = rx_tc_0
                .recv_timeout(Duration::from_millis(20))
                .expect("Receive timeout");
            let tc_len;
            {
                let mut tc_guard = shared_tc_pool_0.write().unwrap();
                let pg = tc_guard.read_with_guard(tc_addr);
                tc_len = pg.read(&mut tc_buf).unwrap();
            }
            let (_tc, _) = PusTcReader::new(&tc_buf[0..tc_len]).unwrap();

            let token = reporter_with_sender_0.add_tc_with_req_id(req_id_0);
            let accepted_token = reporter_with_sender_0
                .acceptance_success(TEST_ID.raw(), &sender_0, token, &FIXED_STAMP)
                .expect("Acceptance success failed");

            // Do some start handling here
            let started_token = reporter_with_sender_0
                .start_success(TEST_ID.raw(), &sender_0, accepted_token, &FIXED_STAMP)
                .expect("Start success failed");
            // Do some step handling here
            reporter_with_sender_0
                .step_success(
                    TEST_ID.raw(),
                    &sender_0,
                    &started_token,
                    &FIXED_STAMP,
                    EcssEnumU8::new(0),
                )
                .expect("Start success failed");

            // Finish up
            reporter_with_sender_0
                .step_success(
                    TEST_ID.raw(),
                    &sender_0,
                    &started_token,
                    &FIXED_STAMP,
                    EcssEnumU8::new(1),
                )
                .expect("Start success failed");
            reporter_with_sender_0
                .completion_success(TEST_ID.raw(), &sender_0, started_token, &FIXED_STAMP)
                .expect("Completion success failed");
        });

        let verif_sender_1 = thread::spawn(move || {
            let mut tc_buf: [u8; 1024] = [0; 1024];
            let tc_addr = rx_tc_1
                .recv_timeout(Duration::from_millis(20))
                .expect("Receive timeout");
            let tc_len;
            {
                let mut tc_guard = shared_tc_pool_1.write().unwrap();
                let pg = tc_guard.read_with_guard(tc_addr);
                tc_len = pg.read(&mut tc_buf).unwrap();
            }
            let (tc, _) = PusTcReader::new(&tc_buf[0..tc_len]).unwrap();
            let token = reporter_with_sender_1.add_tc(&tc);
            let accepted_token = reporter_with_sender_1
                .acceptance_success(TEST_ID.raw(), &sender_1, token, &FIXED_STAMP)
                .expect("Acceptance success failed");
            let started_token = reporter_with_sender_1
                .start_success(TEST_ID.raw(), &sender_1, accepted_token, &FIXED_STAMP)
                .expect("Start success failed");
            let fail_code = EcssEnumU16::new(2);
            let params = FailParams::new_no_fail_data(&FIXED_STAMP, &fail_code);
            reporter_with_sender_1
                .completion_failure(TEST_ID.raw(), &sender_1, started_token, params)
                .expect("Completion success failed");
        });

        let verif_receiver = thread::spawn(move || {
            let mut packet_counter = 0;
            let mut tm_buf: [u8; 1024] = [0; 1024];
            let mut verif_map = HashMap::new();
            while packet_counter < PACKETS_SENT {
                let tm_in_pool = rx
                    .recv_timeout(Duration::from_millis(50))
                    .expect("Packet reception timeout");
                let tm_len;
                let shared_tm_store = shared_tm_pool.clone_backing_pool();
                {
                    let mut rg = shared_tm_store.write().expect("Error locking shared pool");
                    let store_guard = rg.read_with_guard(tm_in_pool.store_addr);
                    tm_len = store_guard
                        .read(&mut tm_buf)
                        .expect("Error reading TM slice");
                }
                let (pus_tm, _) =
                    PusTmReader::new(&tm_buf[0..tm_len], 7).expect("Error reading verification TM");
                let req_id =
                    RequestId::from_bytes(&pus_tm.source_data()[0..RequestId::SIZE_AS_BYTES])
                        .expect("reading request ID from PUS TM source data failed");
                if !verif_map.contains_key(&req_id) {
                    let content = vec![pus_tm.subservice()];
                    verif_map.insert(req_id, content);
                } else {
                    let content = verif_map.get_mut(&req_id).unwrap();
                    content.push(pus_tm.subservice())
                }
                packet_counter += 1;
            }
            for (req_id, content) in verif_map {
                if req_id == req_id_1 {
                    assert_eq!(content[0], 1);
                    assert_eq!(content[1], 3);
                    assert_eq!(content[2], 8);
                } else if req_id == req_id_0 {
                    assert_eq!(content[0], 1);
                    assert_eq!(content[1], 3);
                    assert_eq!(content[2], 5);
                    assert_eq!(content[3], 5);
                    assert_eq!(content[4], 7);
                } else {
                    panic!("Unexpected request ID {:?}", req_id);
                }
            }
        });
        verif_sender_0.join().expect("Joining thread 0 failed");
        verif_sender_1.join().expect("Joining thread 1 failed");
        verif_receiver.join().expect("Joining thread 2 failed");
    }
}