bacnet-server 0.15.0

use super::*;

use std::sync::Mutex as StdMutex;
use std::sync::atomic::AtomicUsize;

use bacnet_encoding::apdu::decode_apdu;
use bacnet_encoding::npdu::decode_npdu;
use bacnet_objects::analog::AnalogInputObject;
use bacnet_objects::life_safety::{LifeSafetyPointObject, LifeSafetyPointResetCommit};
use bacnet_services::life_safety::LifeSafetyOperationRequest;
use bacnet_types::enums::{EngineeringUnits, LifeSafetyOperation, LifeSafetyState, SilencedState};

fn point_oid(instance: u32) -> ObjectIdentifier {
    ObjectIdentifier::new(ObjectType::LIFE_SAFETY_POINT, instance).unwrap()
}

fn request(
    operation: LifeSafetyOperation,
    oid: Option<ObjectIdentifier>,
) -> LifeSafetyOperationRequest {
    LifeSafetyOperationRequest {
        requesting_process_identifier: 41,
        requesting_source: "operator label".into(),
        request: operation,
        object_identifier: oid,
    }
}

async fn dispatch_life_safety_operation(
    db: Arc<RwLock<ObjectDatabase>>,
    config: ServerConfig,
    source_mac: MacAddr,
    source_network: Option<NpduAddress>,
    request: LifeSafetyOperationRequest,
) -> Apdu {
    dispatch_life_safety_operation_with_tracker(
        db,
        config,
        &Arc::new(ConfirmedRequestTracker::default()),
        source_mac,
        source_network,
        0x51,
        request,
    )
    .await
    .expect("first request should receive a response")
}

async fn dispatch_life_safety_operation_with_tracker(
    db: Arc<RwLock<ObjectDatabase>>,
    config: ServerConfig,
    confirmed_request_tracker: &Arc<ConfirmedRequestTracker>,
    source_mac: MacAddr,
    source_network: Option<NpduAddress>,
    invoke_id: u8,
    request: LifeSafetyOperationRequest,
) -> Result<Apdu, tokio::sync::oneshot::error::RecvError> {
    let network = Arc::new(NetworkLayer::new(BipTransport::new(
        Ipv4Addr::LOCALHOST,
        0,
        Ipv4Addr::BROADCAST,
    )));
    let mut service_request = BytesMut::new();
    request.encode(&mut service_request).unwrap();
    let confirmed = ConfirmedRequestPdu {
        segmented: false,
        more_follows: false,
        segmented_response_accepted: false,
        max_segments: None,
        max_apdu_length: 480,
        invoke_id,
        sequence_number: None,
        proposed_window_size: None,
        service_choice: ConfirmedServiceChoice::LIFE_SAFETY_OPERATION,
        service_request: service_request.freeze(),
    };
    let (tx, rx) = oneshot::channel();

    BACnetServer::<BipTransport>::handle_confirmed_request(
        &RequestServices {
            db: Arc::clone(&db),
            ..RequestServices::for_test(Arc::clone(&network), config.clone())
        },
        confirmed_request_tracker,
        &Arc::new(crate::server::request_tasks::RequestTasks::default()).spawner(),
        &source_mac,
        source_network,
        confirmed,
        Some(tx),
    )
    .await;

    rx.await.map(|bytes| {
        let npdu = decode_npdu(bytes).unwrap();
        decode_apdu(npdu.payload).unwrap()
    })
}

async fn dispatch_raw_with_tracker(
    db: Arc<RwLock<ObjectDatabase>>,
    config: ServerConfig,
    confirmed_request_tracker: &Arc<ConfirmedRequestTracker>,
    source_mac: MacAddr,
    source_network: Option<NpduAddress>,
    invoke_id: u8,
    request: LifeSafetyOperationRequest,
) -> Result<Bytes, tokio::sync::oneshot::error::RecvError> {
    let mut service_request = BytesMut::new();
    request.encode(&mut service_request).unwrap();
    dispatch_confirmed_raw_with_tracker(
        db,
        config,
        confirmed_request_tracker,
        source_mac,
        source_network,
        ConfirmedRequestPdu {
            segmented: false,
            more_follows: false,
            segmented_response_accepted: false,
            max_segments: None,
            max_apdu_length: 480,
            invoke_id,
            sequence_number: None,
            proposed_window_size: None,
            service_choice: ConfirmedServiceChoice::LIFE_SAFETY_OPERATION,
            service_request: service_request.freeze(),
        },
    )
    .await
}

async fn dispatch_confirmed_raw_with_tracker(
    db: Arc<RwLock<ObjectDatabase>>,
    config: ServerConfig,
    confirmed_request_tracker: &Arc<ConfirmedRequestTracker>,
    source_mac: MacAddr,
    source_network: Option<NpduAddress>,
    confirmed: ConfirmedRequestPdu,
    // Note: takes the full PDU so oversize/changed-byte cases can craft raw
    // service requests without going through the typed encoder.
) -> Result<Bytes, tokio::sync::oneshot::error::RecvError> {
    let network = Arc::new(NetworkLayer::new(BipTransport::new(
        Ipv4Addr::LOCALHOST,
        0,
        Ipv4Addr::BROADCAST,
    )));
    let (tx, rx) = oneshot::channel();

    BACnetServer::<BipTransport>::handle_confirmed_request(
        &RequestServices {
            db: Arc::clone(&db),
            ..RequestServices::for_test(Arc::clone(&network), config.clone())
        },
        confirmed_request_tracker,
        &Arc::new(crate::server::request_tasks::RequestTasks::default()).spawner(),
        &source_mac,
        source_network,
        confirmed,
        Some(tx),
    )
    .await;

    rx.await
}

fn decode_raw(bytes: &Bytes) -> Apdu {
    let npdu = decode_npdu(bytes.clone()).unwrap();
    decode_apdu(npdu.payload).unwrap()
}

fn assert_error(apdu: Apdu, class: ErrorClass, code: ErrorCode) {
    match apdu {
        Apdu::Error(error) => {
            assert_eq!(error.invoke_id, 0x51);
            assert_eq!(
                error.service_choice,
                ConfirmedServiceChoice::LIFE_SAFETY_OPERATION
            );
            assert_eq!(error.error_class, class);
            assert_eq!(error.error_code, code);
        }
        other => panic!("expected Error PDU, got {other:?}"),
    }
}

fn assert_simple_ack(apdu: Apdu) {
    match apdu {
        Apdu::SimpleAck(ack) => {
            assert_eq!(ack.invoke_id, 0x51);
            assert_eq!(
                ack.service_choice,
                ConfirmedServiceChoice::LIFE_SAFETY_OPERATION
            );
        }
        other => panic!("expected SimpleACK, got {other:?}"),
    }
}

#[tokio::test]
async fn local_rearm_api_supports_two_operation_cycles() {
    let oid = point_oid(1);
    let executions = Arc::new(AtomicUsize::new(0));
    let observed = Arc::clone(&executions);
    let point = LifeSafetyPointObject::new(1, "point")
        .unwrap()
        .with_reset_executor(Arc::new(move |_| {
            observed.fetch_add(1, Ordering::AcqRel);
            Ok(LifeSafetyPointResetCommit::default())
        }));
    let mut objects = ObjectDatabase::new();
    objects.add(Box::new(point)).unwrap();
    let mut server = BACnetServer::<BipTransport>::bip_builder()
        .interface(Ipv4Addr::LOCALHOST)
        .port(0)
        .database(objects)
        .enable_event_enrollment(false)
        .build()
        .await
        .unwrap();

    for operation in [LifeSafetyOperation::RESET, LifeSafetyOperation::RESET_FAULT] {
        server
            .set_life_safety_operation_expected_local(&oid, operation)
            .await
            .unwrap();
        let mut db = server.db.write().await;
        let changed =
            handlers::handle_life_safety_operation(&mut db, &request(operation, Some(oid)))
                .unwrap();
        assert_eq!(
            changed,
            vec![crate::life_safety_cov::LifeSafetyCovChange {
                object_identifier: oid,
                changed_properties: vec![PropertyIdentifier::OPERATION_EXPECTED],
            }]
        );
    }

    assert_eq!(executions.load(Ordering::Acquire), 2);
    let db = server.db.read().await;
    assert_eq!(
        db.get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::OPERATION_EXPECTED, None)
            .unwrap(),
        PropertyValue::Enumerated(LifeSafetyOperation::NONE.to_raw())
    );
    drop(db);
    server.stop().await.unwrap();
}

#[tokio::test]
async fn life_safety_operation_default_policy_denies_without_mutation() {
    let oid = point_oid(1);
    let executions = Arc::new(AtomicUsize::new(0));
    let observed = Arc::clone(&executions);
    let mut point = LifeSafetyPointObject::new(1, "point").unwrap();
    point.set_present_value(LifeSafetyState::ALARM);
    point.set_operation_expected(LifeSafetyOperation::RESET);
    point.set_reset_executor(Arc::new(move |_| {
        observed.fetch_add(1, Ordering::AcqRel);
        Ok(LifeSafetyPointResetCommit {
            present_value: Some(LifeSafetyState::QUIET),
            ..Default::default()
        })
    }));
    let mut objects = ObjectDatabase::new();
    objects.add(Box::new(point)).unwrap();
    let db = Arc::new(RwLock::new(objects));

    let apdu = dispatch_life_safety_operation(
        Arc::clone(&db),
        ServerConfig::default(),
        MacAddr::from_slice(&[1, 2, 3]),
        None,
        request(LifeSafetyOperation::RESET, Some(oid)),
    )
    .await;

    assert_error(
        apdu,
        ErrorClass::SERVICES,
        ErrorCode::SERVICE_REQUEST_DENIED,
    );
    assert_eq!(executions.load(Ordering::Acquire), 0);
    let guard = db.read().await;
    assert_eq!(
        guard
            .get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::PRESENT_VALUE, None)
            .unwrap(),
        PropertyValue::Enumerated(LifeSafetyState::ALARM.to_raw())
    );
    assert_eq!(
        guard
            .get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::OPERATION_EXPECTED, None)
            .unwrap(),
        PropertyValue::Enumerated(LifeSafetyOperation::RESET.to_raw())
    );
}

#[tokio::test]
async fn unknown_target_is_rejected_before_authorization() {
    let authorizations = Arc::new(AtomicUsize::new(0));
    let observed = Arc::clone(&authorizations);
    let config = ServerConfig {
        policy: Some(Arc::new(crate::FnPolicy::new().life_safety_operation(
            move |_| {
                observed.fetch_add(1, Ordering::AcqRel);
                true
            },
        ))),
        ..ServerConfig::default()
    };

    let apdu = dispatch_life_safety_operation(
        Arc::new(RwLock::new(ObjectDatabase::new())),
        config,
        MacAddr::from_slice(&[1]),
        None,
        request(LifeSafetyOperation::RESET, Some(point_oid(99))),
    )
    .await;

    assert_error(apdu, ErrorClass::OBJECT, ErrorCode::UNKNOWN_OBJECT);
    assert_eq!(authorizations.load(Ordering::Acquire), 0);
}

#[tokio::test]
async fn life_safety_operation_policy_receives_routed_identity_before_success() {
    let oid = point_oid(1);
    let mut point = LifeSafetyPointObject::new(1, "point").unwrap();
    point.set_operation_expected(LifeSafetyOperation::SILENCE);
    let mut objects = ObjectDatabase::new();
    objects.add(Box::new(point)).unwrap();
    let db = Arc::new(RwLock::new(objects));
    let seen = Arc::new(StdMutex::new(None));
    let seen_by_policy = Arc::clone(&seen);
    let db_by_policy = Arc::clone(&db);
    let source_mac = MacAddr::from_slice(&[10, 11]);
    let routed_source = NpduAddress {
        network: 222,
        mac_address: MacAddr::from_slice(&[12, 13]),
    };
    let config = ServerConfig {
        policy: Some(Arc::new(crate::FnPolicy::new().life_safety_operation(
            move |context| {
                assert!(
                    db_by_policy.try_write().is_ok(),
                    "authorization must run outside the database lock"
                );
                *seen_by_policy.lock().unwrap() = Some(context.clone());
                true
            },
        ))),
        ..ServerConfig::default()
    };

    let apdu = dispatch_life_safety_operation(
        Arc::clone(&db),
        config,
        source_mac.clone(),
        Some(routed_source.clone()),
        request(LifeSafetyOperation::SILENCE, Some(oid)),
    )
    .await;

    assert_simple_ack(apdu);
    let context = seen.lock().unwrap().clone().expect("policy was invoked");
    assert_eq!(context.source_mac, source_mac);
    assert_eq!(context.source_network, Some(routed_source));
    assert_eq!(context.invoke_id, 0x51);
    assert_eq!(context.request.object_identifier, Some(oid));
    let guard = db.read().await;
    assert_eq!(
        guard
            .get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::SILENCED, None)
            .unwrap(),
        PropertyValue::Enumerated(SilencedState::ALL_SILENCED.to_raw())
    );
}

#[tokio::test]
async fn life_safety_operation_dispatch_preserves_all_targeted_error_rows() {
    let allow = || ServerConfig {
        policy: Some(Arc::new(
            crate::FnPolicy::new().life_safety_operation(|_| true),
        )),
        ..ServerConfig::default()
    };
    let source = MacAddr::from_slice(&[1]);

    let missing_db = Arc::new(RwLock::new(ObjectDatabase::new()));
    let missing = point_oid(99);
    let apdu = dispatch_life_safety_operation(
        missing_db,
        allow(),
        source.clone(),
        None,
        request(LifeSafetyOperation::SILENCE, Some(missing)),
    )
    .await;
    assert_error(apdu, ErrorClass::OBJECT, ErrorCode::UNKNOWN_OBJECT);

    let apdu = dispatch_life_safety_operation(
        Arc::new(RwLock::new(ObjectDatabase::new())),
        allow(),
        source.clone(),
        None,
        request(LifeSafetyOperation::RESET, Some(missing)),
    )
    .await;
    assert_error(apdu, ErrorClass::OBJECT, ErrorCode::UNKNOWN_OBJECT);

    let analog_oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 1).unwrap();
    let mut objects = ObjectDatabase::new();
    objects
        .add(Box::new(
            AnalogInputObject::new(1, "analog", EngineeringUnits::DEGREES_CELSIUS).unwrap(),
        ))
        .unwrap();
    let apdu = dispatch_life_safety_operation(
        Arc::new(RwLock::new(objects)),
        allow(),
        source.clone(),
        None,
        request(LifeSafetyOperation::RESET, Some(analog_oid)),
    )
    .await;
    assert_error(
        apdu,
        ErrorClass::OBJECT,
        ErrorCode::OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED,
    );

    let oid = point_oid(1);
    let mut point = LifeSafetyPointObject::new(1, "point").unwrap();
    point.set_operation_expected(LifeSafetyOperation::RESET);
    let mut objects = ObjectDatabase::new();
    objects.add(Box::new(point)).unwrap();
    let db = Arc::new(RwLock::new(objects));
    let apdu = dispatch_life_safety_operation(
        Arc::clone(&db),
        allow(),
        source.clone(),
        None,
        request(LifeSafetyOperation::RESET, Some(oid)),
    )
    .await;
    assert_error(
        apdu,
        ErrorClass::OBJECT,
        ErrorCode::OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED,
    );

    let apdu = dispatch_life_safety_operation(
        db,
        allow(),
        source,
        None,
        request(LifeSafetyOperation::SILENCE, Some(oid)),
    )
    .await;
    assert_error(
        apdu,
        ErrorClass::OBJECT,
        ErrorCode::INVALID_OPERATION_IN_THIS_STATE,
    );
}

#[tokio::test]
async fn life_safety_operation_targetless_missing_executor_returns_simple_ack_without_mutation() {
    let oid = point_oid(1);
    let mut point = LifeSafetyPointObject::new(1, "point").unwrap();
    point.set_operation_expected(LifeSafetyOperation::RESET);
    let mut objects = ObjectDatabase::new();
    objects.add(Box::new(point)).unwrap();
    let db = Arc::new(RwLock::new(objects));
    let config = ServerConfig {
        policy: Some(Arc::new(
            crate::FnPolicy::new().life_safety_operation(|_| true),
        )),
        ..ServerConfig::default()
    };

    let apdu = dispatch_life_safety_operation(
        Arc::clone(&db),
        config,
        MacAddr::from_slice(&[1]),
        None,
        request(LifeSafetyOperation::RESET, None),
    )
    .await;

    assert_simple_ack(apdu);
    let guard = db.read().await;
    assert_eq!(
        guard
            .get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::OPERATION_EXPECTED, None)
            .unwrap(),
        PropertyValue::Enumerated(LifeSafetyOperation::RESET.to_raw())
    );
}

#[tokio::test]
async fn life_safety_operation_panicking_policy_fails_closed() {
    let oid = point_oid(1);
    let mut objects = ObjectDatabase::new();
    objects
        .add(Box::new(LifeSafetyPointObject::new(1, "point").unwrap()))
        .unwrap();
    let config = ServerConfig {
        policy: Some(Arc::new(
            crate::FnPolicy::new().life_safety_operation(|_| panic!("policy failure")),
        )),
        ..ServerConfig::default()
    };

    let apdu = dispatch_life_safety_operation(
        Arc::new(RwLock::new(objects)),
        config,
        MacAddr::from_slice(&[1]),
        None,
        request(LifeSafetyOperation::SILENCE, Some(oid)),
    )
    .await;

    assert_error(
        apdu,
        ErrorClass::SERVICES,
        ErrorCode::SERVICE_REQUEST_DENIED,
    );
}

// The replay tests run paused: the replay window is measured on tokio's
// clock (#1556), so a runner stall between the sends can't close it.
#[tokio::test(start_paused = true)]
async fn exact_success_duplicate_replays_identical_simple_ack_single_execution() {
    let oid = point_oid(1);
    let executions = Arc::new(AtomicUsize::new(0));
    let observed_executions = Arc::clone(&executions);
    let mut point = LifeSafetyPointObject::new(1, "point").unwrap();
    point.set_present_value(LifeSafetyState::ALARM);
    point.set_operation_expected(LifeSafetyOperation::RESET);
    point.set_reset_executor(Arc::new(move |context| {
        observed_executions.fetch_add(1, Ordering::AcqRel);
        let present_value = if context.operation == LifeSafetyOperation::RESET {
            LifeSafetyState::QUIET
        } else {
            LifeSafetyState::ACTIVE
        };
        Ok(LifeSafetyPointResetCommit {
            present_value: Some(present_value),
            ..Default::default()
        })
    }));
    let mut objects = ObjectDatabase::new();
    objects.add(Box::new(point)).unwrap();
    let db = Arc::new(RwLock::new(objects));
    let authorizations = Arc::new(AtomicUsize::new(0));
    let observed = Arc::clone(&authorizations);
    let config = ServerConfig {
        policy: Some(Arc::new(crate::FnPolicy::new().life_safety_operation(
            move |_| {
                observed.fetch_add(1, Ordering::AcqRel);
                true
            },
        ))),
        ..ServerConfig::default()
    };
    let tracker = Arc::new(ConfirmedRequestTracker::default());
    let source = MacAddr::from_slice(&[1, 2, 3]);
    let reset = request(LifeSafetyOperation::RESET, Some(oid));

    // Byte-identical replay: compare the raw NPDU bytes, not just the decoded
    // APDU, so any future encoding divergence fails the test.
    let first_raw = dispatch_raw_with_tracker(
        Arc::clone(&db),
        config.clone(),
        &tracker,
        source.clone(),
        None,
        0x51,
        reset.clone(),
    )
    .await
    .unwrap();
    assert_simple_ack(decode_raw(&first_raw));
    let second_raw = dispatch_raw_with_tracker(
        Arc::clone(&db),
        config.clone(),
        &tracker,
        source.clone(),
        None,
        0x51,
        reset,
    )
    .await
    .unwrap();
    assert_simple_ack(decode_raw(&second_raw));
    assert_eq!(
        first_raw, second_raw,
        "retransmitted executed LSO must replay byte-identical bytes"
    );
    // Single execution: no second authorizer invocation, no second mutation,
    // and therefore no second COV/event/reset re-fire (replay is byte resend
    // only).
    assert_eq!(authorizations.load(Ordering::Acquire), 1);
    assert_eq!(executions.load(Ordering::Acquire), 1);
    assert_eq!(
        db.read()
            .await
            .get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::PRESENT_VALUE, None)
            .unwrap(),
        PropertyValue::Enumerated(LifeSafetyState::QUIET.to_raw())
    );

    // Changed bytes (different operation) execute normally with a new response.
    db.write()
        .await
        .get_mut(&oid)
        .unwrap()
        .set_life_safety_operation_expected_internal(LifeSafetyOperation::RESET_ALARM)
        .unwrap();
    let changed = dispatch_life_safety_operation_with_tracker(
        Arc::clone(&db),
        config,
        &tracker,
        source,
        None,
        0x51,
        request(LifeSafetyOperation::RESET_ALARM, Some(oid)),
    )
    .await
    .unwrap();
    assert_simple_ack(changed);
    assert_eq!(authorizations.load(Ordering::Acquire), 2);
    assert_eq!(executions.load(Ordering::Acquire), 2);
    assert_eq!(
        db.read()
            .await
            .get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::PRESENT_VALUE, None)
            .unwrap(),
        PropertyValue::Enumerated(LifeSafetyState::ACTIVE.to_raw())
    );
}

#[tokio::test(start_paused = true)]
async fn exact_denied_duplicate_replays_identical_error_single_authorization() {
    let oid = point_oid(1);
    let mut point = LifeSafetyPointObject::new(1, "point").unwrap();
    point.set_operation_expected(LifeSafetyOperation::SILENCE);
    let mut objects = ObjectDatabase::new();
    objects.add(Box::new(point)).unwrap();
    let db = Arc::new(RwLock::new(objects));
    let authorizations = Arc::new(AtomicUsize::new(0));
    let observed = Arc::clone(&authorizations);
    let config = ServerConfig {
        policy: Some(Arc::new(crate::FnPolicy::new().life_safety_operation(
            move |_| {
                observed.fetch_add(1, Ordering::AcqRel);
                false
            },
        ))),
        ..ServerConfig::default()
    };
    let tracker = Arc::new(ConfirmedRequestTracker::default());
    let source = MacAddr::from_slice(&[4, 5, 6]);
    let denied = request(LifeSafetyOperation::SILENCE, Some(oid));

    let first_raw = dispatch_raw_with_tracker(
        Arc::clone(&db),
        config.clone(),
        &tracker,
        source.clone(),
        None,
        0x51,
        denied.clone(),
    )
    .await
    .unwrap();
    assert_error(
        decode_raw(&first_raw),
        ErrorClass::SERVICES,
        ErrorCode::SERVICE_REQUEST_DENIED,
    );
    let second_raw = dispatch_raw_with_tracker(
        Arc::clone(&db),
        config,
        &tracker,
        source,
        None,
        0x51,
        denied,
    )
    .await
    .unwrap();
    assert_error(
        decode_raw(&second_raw),
        ErrorClass::SERVICES,
        ErrorCode::SERVICE_REQUEST_DENIED,
    );
    assert_eq!(
        first_raw, second_raw,
        "denied LSO must replay byte-identical Error bytes"
    );
    assert_eq!(authorizations.load(Ordering::Acquire), 1);
    assert_eq!(
        db.read()
            .await
            .get(&oid)
            .unwrap()
            .read_property(PropertyIdentifier::SILENCED, None)
            .unwrap(),
        PropertyValue::Enumerated(SilencedState::UNSILENCED.to_raw())
    );
}