bacnet-server 0.15.0

use super::*;
use bacnet_types::enums::ObjectType;
use std::time::Duration;

fn ai1() -> ObjectIdentifier {
    ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 1).unwrap()
}

fn ai2() -> ObjectIdentifier {
    ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 2).unwrap()
}

fn make_sub(mac: &[u8], process_id: u32, oid: ObjectIdentifier) -> CovSubscription {
    CovSubscription {
        subscriber_mac: MacAddr::from_slice(mac),
        subscriber_network: None,
        subscriber_process_identifier: process_id,
        monitored_object_identifier: oid,
        issue_confirmed_notifications: false,
        expires_at: None,
        last_notified_observation: None,
        monitored_property: None,
        monitored_property_array_index: None,
        cov_increment: None,
        notification_kind: CovNotificationKind::Single,
        timestamped: false,
    }
}

#[test]
fn subscribe_and_lookup() {
    let mut table = CovSubscriptionTable::new();
    table.subscribe(make_sub(&[1, 2, 3], 1, ai1())).unwrap();
    assert_eq!(table.len(), 1);
    assert_eq!(table.subscriptions_for(&ai1()).len(), 1);
    assert_eq!(table.subscriptions_for(&ai2()).len(), 0);
}

#[test]
fn unsubscribe() {
    let mut table = CovSubscriptionTable::new();
    table.subscribe(make_sub(&[1, 2, 3], 1, ai1())).unwrap();
    assert!(table.unsubscribe(&make_sub(&[1, 2, 3], 1, ai1()).key().unwrap()));
    assert!(!table.unsubscribe(&make_sub(&[1, 2, 3], 1, ai1()).key().unwrap())); // already removed
    assert!(table.is_empty());
}

#[test]
fn expired_subscriptions_purged_on_lookup() {
    let mut table = CovSubscriptionTable::new();
    let mut sub = make_sub(&[1, 2, 3], 1, ai1());
    sub.expires_at = Some(Instant::now() - Duration::from_secs(1)); // already expired
    table.subscribe(sub).unwrap();
    assert_eq!(table.subscriptions_for(&ai1()).len(), 0);
    assert!(table.is_empty());
}

#[test]
fn multiple_subscribers_same_object() {
    let mut table = CovSubscriptionTable::new();
    table.subscribe(make_sub(&[1, 2, 3], 1, ai1())).unwrap();
    table.subscribe(make_sub(&[4, 5, 6], 2, ai1())).unwrap();
    assert_eq!(table.subscriptions_for(&ai1()).len(), 2);
}

#[test]
fn should_notify_no_increment_first_report_fires() {
    let sub = make_sub(&[1, 2, 3], 1, ai1());
    // Binary/multi-state objects have no COV_Increment
    assert!(CovSubscriptionTable::should_notify(
        &sub,
        Some(
            &crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(1.0))
                .unwrap()
        ),
        None
    ));
}

#[test]
fn should_notify_first_notification_always_fires() {
    let sub = make_sub(&[1, 2, 3], 1, ai1());
    // First notification (last_notified_observation = None)
    assert!(CovSubscriptionTable::should_notify(
        &sub,
        Some(
            &crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.5))
                .unwrap()
        ),
        Some(1.0)
    ));
}

#[test]
fn should_notify_change_exceeds_increment() {
    let mut sub = make_sub(&[1, 2, 3], 1, ai1());
    sub.last_notified_observation = Some(
        crate::cov::CovObservation::new(
            crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(70.0))
                .unwrap(),
            None,
        )
        .unwrap(),
    );
    // Change of 2.5 >= increment of 1.0
    assert!(CovSubscriptionTable::should_notify(
        &sub,
        Some(
            &crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.5))
                .unwrap()
        ),
        Some(1.0)
    ));
}

#[test]
fn should_notify_change_below_increment() {
    let mut sub = make_sub(&[1, 2, 3], 1, ai1());
    sub.last_notified_observation = Some(
        crate::cov::CovObservation::new(
            crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.0))
                .unwrap(),
            None,
        )
        .unwrap(),
    );
    // Change of 0.3 < increment of 1.0
    assert!(!CovSubscriptionTable::should_notify(
        &sub,
        Some(
            &crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.3))
                .unwrap()
        ),
        Some(1.0)
    ));
}

#[test]
fn should_notify_exact_increment() {
    let mut sub = make_sub(&[1, 2, 3], 1, ai1());
    sub.last_notified_observation = Some(
        crate::cov::CovObservation::new(
            crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(70.0))
                .unwrap(),
            None,
        )
        .unwrap(),
    );
    // Change of exactly 1.0 == increment of 1.0 → fires
    assert!(CovSubscriptionTable::should_notify(
        &sub,
        Some(
            &crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(71.0))
                .unwrap()
        ),
        Some(1.0)
    ));
}

#[test]
fn should_notify_zero_increment_fires_on_any_change() {
    let mut sub = make_sub(&[1, 2, 3], 1, ai1());
    sub.last_notified_observation = Some(
        crate::cov::CovObservation::new(
            crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.0))
                .unwrap(),
            None,
        )
        .unwrap(),
    );
    // COV_Increment = 0.0 means any change fires
    assert!(CovSubscriptionTable::should_notify(
        &sub,
        Some(
            &crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.001))
                .unwrap()
        ),
        Some(0.0)
    ));
}

#[test]
fn should_notify_ignores_an_unchanged_value() {
    use bacnet_types::primitives::PropertyValue;
    // Without an increment, or with a zero one, only a change reports, however
    // often the object is fanned out.
    for (value, increment) in [
        (PropertyValue::Real(72.0), Some(0.0)),
        (PropertyValue::Real(72.0), None),
        (PropertyValue::Enumerated(1), None),
    ] {
        let mut sub = make_sub(&[1, 2, 3], 1, ai1());
        let sample = crate::cov::CovSample::new(&value).unwrap();
        sub.last_notified_observation =
            Some(crate::cov::CovObservation::new(sample.clone(), None).unwrap());
        assert!(
            !CovSubscriptionTable::should_notify(&sub, Some(&sample), increment),
            "{value:?} with {increment:?}"
        );
    }
}

#[test]
fn complete_for_test_updates() {
    let mut table = CovSubscriptionTable::new();
    let snapshot = table.subscribe(make_sub(&[1, 2, 3], 1, ai1())).unwrap();
    table.complete_for_test(
        &snapshot,
        crate::cov::CovObservation::new(
            crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.5))
                .unwrap(),
            None,
        )
        .unwrap(),
    );

    let subs = table.subscriptions_for(&ai1());
    assert_eq!(
        subs[0].last_notified_observation,
        Some(
            crate::cov::CovObservation::new(
                crate::cov::CovSample::new(&bacnet_types::primitives::PropertyValue::Real(72.5))
                    .unwrap(),
                None
            )
            .unwrap()
        )
    );
}

#[test]
fn upsert_replaces_existing() {
    let mut table = CovSubscriptionTable::new();
    let mut sub = make_sub(&[1, 2, 3], 1, ai1());
    sub.issue_confirmed_notifications = false;
    table.subscribe(sub).unwrap();
    // Same (mac, process_id, object) key — replaces the existing entry
    let mut sub2 = make_sub(&[1, 2, 3], 1, ai1());
    sub2.issue_confirmed_notifications = true;
    table.subscribe(sub2).unwrap();
    assert_eq!(table.len(), 1);
    let subs = table.subscriptions_for(&ai1());
    assert!(subs[0].issue_confirmed_notifications);
}

#[test]
fn same_subscriber_different_objects_both_exist() {
    let mut table = CovSubscriptionTable::new();
    // Same (mac, process_id) but different monitored objects
    table.subscribe(make_sub(&[1, 2, 3], 1, ai1())).unwrap();
    table.subscribe(make_sub(&[1, 2, 3], 1, ai2())).unwrap();
    assert_eq!(table.len(), 2);
    assert_eq!(table.subscriptions_for(&ai1()).len(), 1);
    assert_eq!(table.subscriptions_for(&ai2()).len(), 1);
}

#[test]
fn purge_expired_removes_stale_subscriptions() {
    let mut table = CovSubscriptionTable::new();
    let mut sub1 = make_sub(&[1, 2, 3], 1, ai1());
    sub1.expires_at = Some(Instant::now() - Duration::from_secs(10));
    let mut sub2 = make_sub(&[4, 5, 6], 2, ai1());
    sub2.expires_at = None; // infinite lifetime
    table.subscribe(sub2).unwrap();
    table.subscribe(sub1).unwrap();

    let purged = table.purge_expired();
    assert_eq!(purged, 1);
    assert_eq!(table.len(), 1);
}

#[test]
fn cov_multiple_context_lifetime_refreshes_and_expires() {
    let mut table = CovSubscriptionTable::new();
    let original_expiry = Instant::now() + Duration::from_secs(30);
    let refreshed_expiry = Instant::now() + Duration::from_secs(60);

    let mut present_value = make_sub(&[1, 2, 3], 1, ai1());
    present_value.notification_kind = CovNotificationKind::Multiple;
    present_value.monitored_property = Some(PropertyIdentifier::PRESENT_VALUE);
    present_value.expires_at = Some(original_expiry);
    table.admit_for_test(present_value, 5).unwrap();

    let mut status_flags = make_sub(&[1, 2, 3], 1, ai1());
    status_flags.notification_kind = CovNotificationKind::Multiple;
    status_flags.monitored_property = Some(PropertyIdentifier::STATUS_FLAGS);
    status_flags.expires_at = Some(original_expiry);
    table.admit_for_test(status_flags, 5).unwrap();

    let mut single = make_sub(&[1, 2, 3], 1, ai1());
    single.expires_at = Some(original_expiry);
    table.subscribe(single).unwrap();

    let context = MultipleContextKey {
        recipient: CovRecipient::from_endpoint(&[1, 2, 3], None),
        process_id: 1,
        confirmed: false,
    };
    table
        .subscribe_multiple(
            &context,
            &SubscriberEndpoint::new(&[1, 2, 3], None),
            refreshed_expiry,
            9,
            None,
            vec![],
        )
        .unwrap();

    // The refresh point updates the whole context's expiry and reported
    // delay together; the Single entry has neither refreshed.
    let multiple_terms: Vec<_> = table
        .subs
        .values()
        .filter(|sub| sub.notification_kind == CovNotificationKind::Multiple)
        .map(|sub| (sub.expires_at, sub.max_notification_delay()))
        .collect();
    assert_eq!(multiple_terms, vec![(Some(refreshed_expiry), Some(9)); 2]);
    assert!(
        table
            .subs
            .values()
            .any(|sub| sub.notification_kind == CovNotificationKind::Single
                && sub.expires_at == Some(original_expiry)
                && sub.max_notification_delay().is_none())
    );

    table
        .subscribe_multiple(
            &context,
            &SubscriberEndpoint::new(&[1, 2, 3], None),
            Instant::now() - Duration::from_secs(1),
            9,
            None,
            vec![],
        )
        .unwrap();

    assert_eq!(table.purge_expired(), 2);
    assert_eq!(table.len(), 1);
    assert!(table.subs.values().all(|sub| {
        sub.notification_kind == CovNotificationKind::Single
            && sub.expires_at == Some(original_expiry)
    }));
}

#[test]
fn purge_expired_returns_zero_when_none_expired() {
    let mut table = CovSubscriptionTable::new();
    table.subscribe(make_sub(&[1, 2, 3], 1, ai1())).unwrap();
    let purged = table.purge_expired();
    assert_eq!(purged, 0);
    assert_eq!(table.len(), 1);
}

#[test]
fn default_policy_allows_1024th_subscription() {
    let mut table = CovSubscriptionTable::new();
    let oid = ai1();
    // 16 peers * 64 subscriptions each = 1024 subscriptions
    for peer_idx in 0..16u8 {
        let mac = [192, 168, 1, peer_idx];
        let peer_key = CovRecipient::Direct(MacAddr::from_slice(&mac));
        for proc_id in 0..64u32 {
            table
                .check_admission(&peer_key, false, None)
                .expect("subscription admitted");
            let mut sub = make_sub(&mac, proc_id, oid);
            sub.expires_at = Some(Instant::now() + Duration::from_secs(300));
            table.subscribe(sub).unwrap();
        }
    }
    assert_eq!(table.len(), 1024);

    // 1025th subscription from a 17th peer fails due to global capacity
    let mac17 = [192, 168, 1, 17];
    let peer17 = CovRecipient::Direct(MacAddr::from_slice(&mac17));
    assert!(table.check_admission(&peer17, false, None).is_err());
}

#[test]
fn effective_unreserved_capacity_respects_reserved_peers() {
    let mut policy = CovPolicy {
        max_subscriptions_global: 100,
        reserved_capacity: 20,
        reserved_peers: Vec::new(),
        ..Default::default()
    };
    // No reserved peers -> full global capacity available
    assert_eq!(policy.effective_unreserved_capacity(), 100);

    // With reserved peers -> reserved capacity is deducted
    policy.reserved_peers.push(MacAddr::from_slice(&[1, 2, 3]));
    assert_eq!(policy.effective_unreserved_capacity(), 80);

    // If reserved_capacity is 0 -> full global capacity
    policy.reserved_capacity = 0;
    assert_eq!(policy.effective_unreserved_capacity(), 100);
}

#[test]
fn in_flight_tracker_does_not_leak_zero_count_entries_on_failure() {
    let tracker = Arc::new(CovInFlightTracker::default());
    let semaphore = Arc::new(tokio::sync::Semaphore::new(0));
    let peer = CovRecipient::Direct(MacAddr::from_slice(&[1, 2, 3, 4]));

    // Acquisition fails due to global pool exhausted
    let err = tracker
        .try_acquire(peer.clone(), 10, &semaphore)
        .unwrap_err();
    assert_eq!(err, InFlightAcquireError::GlobalPoolExhausted);
    assert_eq!(tracker.active_peer_count(), 0);

    // Acquisition fails due to peer limit exceeded (max_per_peer = 0)
    let semaphore2 = Arc::new(tokio::sync::Semaphore::new(10));
    let peer2 = CovRecipient::Direct(MacAddr::from_slice(&[5, 6, 7, 8]));
    let err2 = tracker.try_acquire(peer2, 0, &semaphore2).unwrap_err();
    assert_eq!(err2, InFlightAcquireError::PeerLimitExceeded);
    assert_eq!(tracker.active_peer_count(), 0);
}

#[test]
fn expired_subscriptions_immediately_release_quota_on_admission() {
    let policy = CovPolicy {
        max_subscriptions_per_peer: 1,
        ..Default::default()
    };
    let mut table =
        CovSubscriptionTable::with_policy(policy, Arc::new(AtomicCovCounters::default()));
    let peer = CovRecipient::Direct(MacAddr::from_slice(&[1, 2, 3]));

    // Create a subscription with an expiry in the past
    let mut sub = make_sub(&[1, 2, 3], 1, ai1());
    sub.expires_at = Some(Instant::now() - Duration::from_secs(5));
    table.subscribe(sub).unwrap();
    assert_eq!(table.len(), 1);

    // Admitting a new subscription from the same peer immediately purges the expired subscription
    // and succeeds, rather than being rejected by per-peer quota!
    assert!(table.check_admission(&peer, false, None).is_ok());
    assert_eq!(table.len(), 0);
}

#[test]
fn is_peer_reserved_checks_canonical_peer_identity() {
    let direct_mac = MacAddr::from_slice(&[1, 2, 3]);
    let policy = CovPolicy {
        reserved_peers: vec![direct_mac.clone()],
        reserved_recipients: vec![CovRecipient::Routed(NpduAddress {
            network: 10,
            mac_address: direct_mac.clone(),
        })],
        ..Default::default()
    };

    // Direct peer matching reserved_peers is reserved
    assert!(policy.is_peer_reserved(&CovRecipient::Direct(direct_mac.clone())));

    // Routed peer on network 10 matching reserved_recipients is reserved
    assert!(policy.is_peer_reserved(&CovRecipient::Routed(NpduAddress {
        network: 10,
        mac_address: direct_mac.clone()
    })));

    // Routed peer on different network (20) with same MAC is NOT reserved
    assert!(!policy.is_peer_reserved(&CovRecipient::Routed(NpduAddress {
        network: 20,
        mac_address: direct_mac.clone()
    })));

    // Unconfigured direct peer is NOT reserved
    let other_mac = MacAddr::from_slice(&[4, 5, 6]);
    assert!(!policy.is_peer_reserved(&CovRecipient::Direct(other_mac.clone())));
    assert!(!policy.is_peer_reserved(&CovRecipient::Routed(NpduAddress {
        network: 10,
        mac_address: other_mac
    })));
}

#[test]
fn finest_increments_take_each_objects_smallest_live_increment_on_the_property() {
    let tv = PropertyIdentifier::TRACKING_VALUE;
    let on = |mac: u8, oid, property, cov_increment| CovSubscription {
        monitored_property: Some(property),
        cov_increment,
        ..make_sub(&[mac], 1, oid)
    };
    let mut table = CovSubscriptionTable::new();
    for sub in [
        on(1, ai1(), tv, Some(5.0)),
        on(2, ai1(), tv, Some(0.5)),
        // Another property's increment doesn't count.
        on(3, ai1(), PropertyIdentifier::PRESENT_VALUE, Some(0.1)),
        // Nor does one that reports nothing finer, nor none at all.
        on(4, ai2(), tv, Some(f32::NAN)),
        on(5, ai2(), tv, Some(f32::INFINITY)),
        on(6, ai2(), tv, None),
    ] {
        table.subscribe(sub).unwrap();
    }
    assert_eq!(table.finest_increments(tv), HashMap::from([(ai1(), 0.5)]));
    // A negative increment reports any change, as zero does; an expired
    // subscription counts for nothing.
    table.subscribe(on(7, ai2(), tv, Some(-1.0))).unwrap();
    let mut expired = on(8, ai1(), tv, Some(0.01));
    expired.expires_at = Some(Instant::now() - Duration::from_secs(1));
    table.subscribe(expired).unwrap();
    assert_eq!(
        table.finest_increments(tv),
        HashMap::from([(ai1(), 0.5), (ai2(), 0.0)])
    );
}