use std::sync::Arc;
use bacnet_transport::loopback::LoopbackTransport;
use bacnet_transport::port::{ReceivedNpdu, TransportPort};
use bacnet_types::enums::NetworkMessageType;
use tokio::time::{timeout, Duration};
use super::*;
use crate::layer::QueueAdmissionSnapshot;
use crate::loopback_fixture::recv;
use crate::router::RouterPort;
const I_AM_3000: [u8; 5] = [0x01, 0x80, 0x01, 0x0B, 0xB8];
const REJECT_3000: [u8; 6] = [0x01, 0x80, 0x03, 0x01, 0x0B, 0xB8];
const TO_6000: [u8; 8] = [0x01, 0x20, 0x17, 0x70, 0x00, 0xFF, 0x10, 0x08];
const REJECT_6000: [u8; 6] = [0x01, 0x80, 0x03, 0x01, 0x17, 0x70];
async fn ports() -> (
Vec<RouterPort<LoopbackTransport>>,
[(LoopbackTransport, mpsc::Receiver<ReceivedNpdu>); 2],
) {
let (port_a, mut peer_a) = LoopbackTransport::pair(vec![0x01], vec![0x0A]);
let (port_b, mut peer_b) = LoopbackTransport::pair(vec![0x02], vec![0x0B]);
let from_router_a = peer_a.start().await.unwrap();
let from_router_b = peer_b.start().await.unwrap();
let ports = vec![
RouterPort {
transport: port_a,
network_number: 1000,
},
RouterPort {
transport: port_b,
network_number: 2000,
},
];
(ports, [(peer_a, from_router_a), (peer_b, from_router_b)])
}
#[tokio::test]
async fn one_start_combines_admission_tracking_a_control_policy_and_the_control_receiver() {
let (ports, [(mut peer_a, _), (mut peer_b, _)]) = ports().await;
let reject = NetworkMessageType::REJECT_MESSAGE_TO_NETWORK.to_raw();
let options = RouterOptions::new()
.track_admission()
.control_policy(ControlPolicy::Hardened)
.control_authorizer(Arc::new(move |ctx| ctx.message_type == reject))
.network_control_receiver();
let StartedRouter {
mut router,
mut apdus,
network_control,
} = BACnetRouter::start(ports, options).await.unwrap();
let mut controls = network_control.expect("asked for");
let counters = apdus.counters();
peer_b.send_broadcast(&I_AM_3000).await.unwrap();
let reject_3000 = [0x01, 0x80, 0x03, 0x01, 0x0B, 0xB8];
peer_b.send_unicast(&reject_3000, &[0x02]).await.unwrap();
let control = recv(&mut controls).await;
assert_eq!(control.npdu.payload[..], reject_3000[3..]);
assert_eq!(control.source_mac.as_slice(), [0x0B]);
let decisions = router.control_snapshot();
assert_eq!(
(decisions.i_am.allow_total, decisions.i_am.deny_total),
(0, 1)
);
assert_eq!(decisions.reject.allow_total, 1);
assert!(router.table().lock().await.lookup(3000).is_none());
peer_a
.send_unicast(&[0x01, 0x00, 0x10, 0x08], &[0x01])
.await
.unwrap();
let apdu = timeout(Duration::from_secs(2), apdus.recv())
.await
.expect("local APDU")
.expect("queue open");
assert_eq!(apdu.apdu[..], [0x10, 0x08]);
assert_eq!(apdu.ingress_network, Some(1000));
let snapshot = counters.snapshot();
assert_eq!((snapshot.current_depth, snapshot.high_water), (0, 1));
router.stop().await;
peer_a.stop().await.unwrap();
peer_b.stop().await.unwrap();
}
#[tokio::test]
async fn plain_options_start_a_permissive_router_without_a_control_receiver() {
let (ports, [(mut peer_a, mut from_router_a), (mut peer_b, _)]) = ports().await;
let StartedRouter {
mut router,
apdus: _apdus,
network_control,
} = BACnetRouter::start(ports, RouterOptions::new())
.await
.unwrap();
assert!(network_control.is_none());
peer_b.send_broadcast(&I_AM_3000).await.unwrap();
loop {
let frame = recv(&mut from_router_a).await;
if frame.npdu[..] == I_AM_3000 {
break;
}
}
assert_eq!(router.control_snapshot().i_am.allow_total, 1);
assert_eq!(
router.table().lock().await.lookup(3000).unwrap().port_index,
1
);
router.stop().await;
peer_a.stop().await.unwrap();
peer_b.stop().await.unwrap();
}
async fn rejects_to_the_router(
peer_b: &LoopbackTransport,
from_router_b: &mut mpsc::Receiver<ReceivedNpdu>,
count: usize,
) {
for _ in 0..count {
peer_b.send_unicast(&REJECT_3000, &[0x02]).await.unwrap();
}
peer_b.send_unicast(&TO_6000, &[0x02]).await.unwrap();
while recv(from_router_b).await.npdu[..] != REJECT_6000 {}
}
#[tokio::test]
async fn tracked_network_control_receiver_counts_its_high_water_mark_and_drops() {
let (ports, [(mut peer_a, _from_router_a), (mut peer_b, mut from_router_b)]) = ports().await;
let options = RouterOptions::new().network_control_receiver_with_admission();
let StartedRouter {
mut router,
apdus: _apdus,
network_control,
} = BACnetRouter::start(ports, options).await.unwrap();
let mut controls = network_control.expect("asked for");
let counters = controls.counters();
rejects_to_the_router(&peer_b, &mut from_router_b, 260).await;
assert_eq!(
counters.snapshot(),
QueueAdmissionSnapshot {
current_depth: 256,
high_water: 256,
full_drops: 4,
fairness_drops: 0,
closed_drops: 0,
}
);
assert_eq!(router.network_control_ingress_sequence(), 260);
for sequence in 1..=256 {
let control = controls.try_recv().unwrap();
assert_eq!(control.ingress_sequence, sequence);
assert_eq!(control.npdu.payload[..], REJECT_3000[3..]);
}
assert_eq!(
counters.snapshot(),
QueueAdmissionSnapshot {
high_water: 256,
full_drops: 4,
..Default::default()
}
);
controls.close();
rejects_to_the_router(&peer_b, &mut from_router_b, 2).await;
assert_eq!(
counters.snapshot(),
QueueAdmissionSnapshot {
high_water: 256,
full_drops: 4,
closed_drops: 2,
..Default::default()
}
);
router.stop().await;
peer_a.stop().await.unwrap();
peer_b.stop().await.unwrap();
}