//! Pure exit-gate coverage for the phase-3 observer core.
//!
//! Five of the six exit-gate rows live here (the sixth — a live create/destroy
//! topology diff — needs the daemon and belongs to the adapter). Each test
//! builds the [`RegEvent`] stream by hand; nothing links PipeWire.
//!
//! Carrying the phase-0a lesson: the id/pid/serial tests use **interior**
//! values, not just 1 and a huge number, so a middle-of-range mistake cannot
//! hide.

use super::classify::{Classification, DeviceClaim, classify};
use super::pulse_pid;
use super::*;
use crate::host::taint::snapshot::{
    ClientSnapshot, GlobalId, IdLookup, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial,
};

// ---- builders -------------------------------------------------------------

fn ser(n: u64) -> Serial {
    Serial(n)
}
fn gid(n: u32) -> GlobalId {
    GlobalId(n)
}
fn model() -> RegistryModel {
    // now=0, a 5 s readiness budget.
    RegistryModel::new(0, 5000)
}

fn no_device() -> DeviceClaim {
    DeviceClaim::default()
}

fn hw_claim(device_id: u32, api: &str, factory: &str) -> DeviceClaim {
    DeviceClaim {
        device_id: Some(gid(device_id)),
        device_api: Some(api.to_string()),
        factory_name: Some(factory.to_string()),
    }
}

/// A `Stream/Output/Audio` node with no backing Device — admitted at once.
fn stream_out(serial: u64, id: u32) -> RegEvent {
    RegEvent::NodeAdded(NodeObservation {
        serial: ser(serial),
        id: gid(id),
        name: Some(format!("stream-{id}")),
        role: MediaRole::StreamOutput,
        props: NodeProps::default(),
        device_claim: no_device(),
    })
}

/// A node backed by a Device (withheld until that Device resolves).
fn device_node(serial: u64, id: u32, role: MediaRole, claim: DeviceClaim) -> RegEvent {
    RegEvent::NodeAdded(NodeObservation {
        serial: ser(serial),
        id: gid(id),
        name: Some(format!("dev-node-{id}")),
        role,
        props: NodeProps::default(),
        device_claim: claim,
    })
}

fn client(serial: u64, id: u32, sec_pid: Option<u32>) -> RegEvent {
    RegEvent::ClientAdded(ClientSnapshot {
        serial: ser(serial),
        id: gid(id),
        sec_pid,
    })
}

fn port(serial: u64, id: u32, node_id: u32, dir: PortDirection) -> RegEvent {
    RegEvent::PortAdded(PortSnapshot {
        serial: ser(serial),
        id: gid(id),
        node: gid(node_id),
        direction: dir,
        exclusive: false,
        monitor: false,
    })
}

fn endpoints(out_node: u32, in_node: u32) -> LinkEndpoints {
    LinkEndpoints {
        output_node: gid(out_node),
        input_node: gid(in_node),
        output_port: None,
        input_port: None,
    }
}

// ==========================================================================
// classify() — session_device
// ==========================================================================

#[test]
fn classify_no_device_is_not_a_device() {
    assert_eq!(classify(&no_device(), false), Classification::NotADevice);
    // `device_resolved` is irrelevant with no device_id.
    assert_eq!(classify(&no_device(), true), Classification::NotADevice);
}

#[test]
fn classify_unresolved_device_withholds() {
    let claim = hw_claim(42, "alsa", "api.alsa.pcm.sink");
    assert_eq!(
        classify(&claim, false),
        Classification::Withhold { device_id: gid(42) }
    );
}

#[test]
fn classify_resolved_hardware_pcm_is_session_device() {
    for factory in [
        "api.alsa.pcm.sink",
        "api.alsa.pcm.source",
        "api.bluez5.pcm.sink",
        "api.bluez5.pcm.source",
    ] {
        let api = if factory.contains("bluez5") {
            "bluez5"
        } else {
            "alsa"
        };
        assert_eq!(
            classify(&hw_claim(7, api, factory), true),
            Classification::SessionDevice,
            "factory {factory} should be a session device"
        );
    }
}

#[test]
fn classify_resolved_but_not_hardware_pcm_fails_closed() {
    // A null sink, a loopback, and an unknown factory are all forwarders, not
    // terminals: resolved, but session_device stays false.
    for factory in ["support.null-audio-sink", "api.alsa.pcm.loopback", "wat"] {
        assert_eq!(
            classify(&hw_claim(7, "alsa", factory), true),
            Classification::NotSessionDevice,
            "factory {factory} must not be a session device"
        );
    }
}

#[test]
fn classify_missing_device_api_fails_closed() {
    // Even with an allowlisted factory, no device.api ⇒ not positively a
    // real-backend terminal.
    let claim = DeviceClaim {
        device_id: Some(gid(7)),
        device_api: None,
        factory_name: Some("api.alsa.pcm.sink".to_string()),
    };
    assert_eq!(classify(&claim, true), Classification::NotSessionDevice);
}

#[test]
fn classify_allowlist_is_exact_not_substring() {
    // A factory that merely *contains* an allowlisted name must not pass.
    let claim = hw_claim(7, "alsa", "api.alsa.pcm.sink.evil");
    assert_eq!(classify(&claim, true), Classification::NotSessionDevice);
}

// ==========================================================================
// pulse_pid — the six-case derivation matrix
// ==========================================================================

fn clients_with(pids: &[Option<u32>]) -> Vec<ClientSnapshot> {
    pids.iter()
        .enumerate()
        .map(|(i, &sec_pid)| ClientSnapshot {
            serial: ser(1000 + i as u64),
            id: gid(200 + i as u32),
            sec_pid,
        })
        .collect()
}

#[test]
fn pid_candidate_consistent_repeated_value() {
    // interior pid values, not 1 / u32::MAX.
    let cs = clients_with(&[Some(4137), Some(4137), Some(9001), Some(12034)]);
    assert_eq!(pulse_pid::candidate(&cs), Some(4137));
}

#[test]
fn pid_candidate_inconsistent_two_repeats_is_none() {
    let cs = clients_with(&[Some(4137), Some(4137), Some(9001), Some(9001)]);
    assert_eq!(pulse_pid::candidate(&cs), None);
}

#[test]
fn pid_candidate_missing_property_is_none() {
    let cs = clients_with(&[None, None, None]);
    assert_eq!(pulse_pid::candidate(&cs), None);
}

#[test]
fn pid_candidate_single_occurrence_is_none() {
    // A lone native client carrying its own pid is indistinguishable from a
    // one-client pulse; the >=2 threshold rejects it.
    let cs = clients_with(&[Some(4137), Some(9001), Some(12034)]);
    assert_eq!(pulse_pid::candidate(&cs), None);
}

#[test]
fn pid_validate_matches_pulse_comm() {
    assert_eq!(
        pulse_pid::validate(4137, Some("pipewire-pulse")),
        Some(4137)
    );
}

#[test]
fn pid_validate_proc_missing_is_none() {
    // case 4: /proc entry gone.
    assert_eq!(pulse_pid::validate(4137, None), None);
}

#[test]
fn pid_validate_comm_mismatch_is_none() {
    // case 5: a different process holds the number.
    assert_eq!(pulse_pid::validate(4137, Some("firefox")), None);
}

#[test]
fn pid_validate_reuse_named_other_process_is_none() {
    // case 6: PID reuse — the number is live but /proc names someone else.
    assert_eq!(pulse_pid::validate(4137, Some("Xwayland")), None);
    // and a truncation-adjacent near-miss must not pass an exact match.
    assert_eq!(pulse_pid::validate(4137, Some("pipewire-pulseX")), None);
}

#[test]
fn pid_derive_end_to_end_valid() {
    let cs = clients_with(&[Some(4137), Some(4137), Some(9001)]);
    let got = pulse_pid::derive(&cs, |pid| {
        (pid == 4137).then(|| "pipewire-pulse".to_string())
    });
    assert_eq!(got, Some(4137));
}

// ==========================================================================
// model — pulse pid through project()
// ==========================================================================

/// Drive the model to Complete so `project` reflects a trusted graph, without
/// caring about the specific objects.
fn drive_ready(m: &mut RegistryModel) {
    m.apply(RegEvent::ServerSynced);
}

#[test]
fn model_pulse_pid_valid_through_projection() {
    let mut m = model();
    m.apply(client(1, 200, Some(4137)));
    m.apply(client(2, 201, Some(4137)));
    m.apply(client(3, 202, Some(9001)));
    assert_eq!(m.pulse_pid_candidate(), Some(4137));
    m.apply(RegEvent::ProcCommProbed {
        pid: 4137,
        comm: Some("pipewire-pulse".to_string()),
    });
    assert_eq!(m.project().pipewire_pulse_pid, Some(4137));
}

#[test]
fn model_pulse_pid_none_until_probed() {
    let mut m = model();
    m.apply(client(1, 200, Some(4137)));
    m.apply(client(2, 201, Some(4137)));
    // candidate exists, but no /proc confirmation yet ⇒ fail closed.
    assert_eq!(m.project().pipewire_pulse_pid, None);
}

#[test]
fn model_pulse_pid_none_on_comm_mismatch() {
    let mut m = model();
    m.apply(client(1, 200, Some(4137)));
    m.apply(client(2, 201, Some(4137)));
    m.apply(RegEvent::ProcCommProbed {
        pid: 4137,
        comm: Some("firefox".to_string()),
    });
    assert_eq!(m.project().pipewire_pulse_pid, None);
}

// ==========================================================================
// model — add / remove of all four object types
// ==========================================================================

#[test]
fn model_adds_all_four_object_types() {
    let mut m = model();
    m.apply(stream_out(100, 50));
    m.apply(port(101, 60, 50, PortDirection::Out));
    m.apply(client(102, 70, Some(4137)));
    m.apply(RegEvent::LinkAdded {
        serial: ser(103),
        id: gid(80),
        endpoints: Some(endpoints(50, 55)),
    });
    let snap = m.project().snapshot;
    assert_eq!(snap.nodes().count(), 1);
    assert_eq!(snap.ports().count(), 1);
    assert_eq!(snap.clients().count(), 1);
    assert_eq!(snap.links().count(), 1);
}

#[test]
fn model_removes_all_four_object_types() {
    let mut m = model();
    m.apply(stream_out(100, 50));
    m.apply(port(101, 60, 50, PortDirection::Out));
    m.apply(client(102, 70, Some(4137)));
    m.apply(RegEvent::LinkAdded {
        serial: ser(103),
        id: gid(80),
        endpoints: Some(endpoints(50, 55)),
    });

    m.apply(RegEvent::Removed { id: gid(50) });
    m.apply(RegEvent::Removed { id: gid(60) });
    m.apply(RegEvent::Removed { id: gid(70) });
    m.apply(RegEvent::Removed { id: gid(80) });

    let snap = m.project().snapshot;
    assert_eq!(snap.nodes().count(), 0);
    assert_eq!(snap.ports().count(), 0);
    assert_eq!(snap.clients().count(), 0);
    assert_eq!(snap.links().count(), 0);
}

#[test]
fn model_remove_of_unknown_id_is_harmless() {
    let mut m = model();
    m.apply(stream_out(100, 50));
    m.apply(RegEvent::Removed { id: gid(999) });
    assert_eq!(m.project().snapshot.nodes().count(), 1);
}

// ==========================================================================
// model — recycled global id, oldest generation first (fail closed)
// ==========================================================================

#[test]
fn model_recycled_id_is_ambiguous_until_removal_accounted() {
    let mut m = model();
    // A missed removal: two live nodes claim id 50 (serials 100 then 200).
    m.apply(stream_out(100, 50));
    m.apply(stream_out(200, 50));

    // The snapshot fails closed: id 50 is ambiguous.
    let snap = m.project().snapshot;
    assert_eq!(snap.node_by_id(gid(50)), Some(IdLookup::Ambiguous));
    assert_eq!(snap.nodes().count(), 2);

    // One removal accounts for the OLDEST generation (serial 100); the newer
    // node survives and the id is unambiguous again.
    m.apply(RegEvent::Removed { id: gid(50) });
    let snap = m.project().snapshot;
    assert_eq!(snap.node_by_id(gid(50)), Some(IdLookup::Unique(ser(200))));
    assert!(snap.node(ser(200)).is_some());
    assert!(snap.node(ser(100)).is_none());
}

// ==========================================================================
// model — Link endpoint resolution (bind fallback path)
// ==========================================================================

#[test]
fn model_link_with_endpoints_appears_immediately() {
    let mut m = model();
    m.apply(RegEvent::LinkAdded {
        serial: ser(103),
        id: gid(80),
        endpoints: Some(endpoints(50, 55)),
    });
    assert_eq!(m.project().snapshot.links().count(), 1);
}

#[test]
fn model_link_without_endpoints_is_withheld_until_resolved() {
    let mut m = model();
    // The correctness path: the global carried no endpoint props.
    m.apply(RegEvent::LinkAdded {
        serial: ser(103),
        id: gid(80),
        endpoints: None,
    });
    // Not in the snapshot yet, and it blocks readiness.
    assert_eq!(m.project().snapshot.links().count(), 0);
    m.apply(RegEvent::ServerSynced);
    assert!(!m.graph_ready(), "pending link must hold readiness");

    // The bind fallback resolves it.
    m.apply(RegEvent::LinkEndpointsResolved {
        serial: ser(103),
        endpoints: endpoints(50, 55),
    });
    let snap = m.project().snapshot;
    assert_eq!(snap.links().count(), 1);
    let link = snap.links().next().unwrap();
    assert_eq!(link.output_node, gid(50));
    assert_eq!(link.input_node, gid(55));
    assert!(
        m.graph_ready(),
        "resolving the last obligation completes readiness"
    );
}

#[test]
fn model_stale_link_resolution_is_ignored() {
    let mut m = model();
    m.apply(RegEvent::LinkAdded {
        serial: ser(103),
        id: gid(80),
        endpoints: None,
    });
    // Link removed before the bind returned.
    m.apply(RegEvent::Removed { id: gid(80) });
    // A late resolution for the gone link must not resurrect it.
    m.apply(RegEvent::LinkEndpointsResolved {
        serial: ser(103),
        endpoints: endpoints(50, 55),
    });
    assert_eq!(m.project().snapshot.links().count(), 0);
    m.apply(RegEvent::ServerSynced);
    assert!(
        m.graph_ready(),
        "the obligation cleared when the link was removed"
    );
}

// ==========================================================================
// model — readiness epoch
// ==========================================================================

#[test]
fn model_readiness_waits_for_sync() {
    let mut m = model();
    m.apply(stream_out(100, 50));
    assert_eq!(m.readiness(), Readiness::Waiting);
    assert!(!m.project().graph_ready);
    m.apply(RegEvent::ServerSynced);
    assert_eq!(m.readiness(), Readiness::Complete);
    assert!(m.project().graph_ready);
}

#[test]
fn model_readiness_does_not_release_with_obligation_outstanding() {
    let mut m = model();
    // A node withheld on an unresolved device is an outstanding obligation.
    m.apply(device_node(
        100,
        50,
        MediaRole::Sink,
        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
    ));
    m.apply(RegEvent::ServerSynced);
    // Synced, but the withheld node keeps the epoch shut.
    assert_eq!(m.readiness(), Readiness::Waiting);
    assert!(!m.graph_ready());

    // Resolving the device admits the node and completes readiness.
    m.apply(RegEvent::DeviceAdded { id: gid(42) });
    assert_eq!(m.readiness(), Readiness::Complete);
    assert!(m.graph_ready());
}

#[test]
fn model_readiness_times_out_fail_closed() {
    let mut m = model();
    m.apply(device_node(
        100,
        50,
        MediaRole::Sink,
        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
    ));
    m.apply(RegEvent::ServerSynced);
    assert_eq!(m.readiness(), Readiness::Waiting);

    // The device never resolves; the deadline passes.
    m.apply(RegEvent::Tick { now: 5000 });
    assert_eq!(m.readiness(), Readiness::TimedOut);
    assert!(!m.graph_ready(), "timeout fails closed");
}

#[test]
fn model_tick_before_deadline_does_not_time_out() {
    let mut m = model();
    m.apply(device_node(
        100,
        50,
        MediaRole::Sink,
        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
    ));
    m.apply(RegEvent::Tick { now: 4999 });
    assert_eq!(m.readiness(), Readiness::Waiting);
}

#[test]
fn model_complete_is_sticky_across_later_churn() {
    let mut m = model();
    m.apply(RegEvent::ServerSynced);
    assert_eq!(m.readiness(), Readiness::Complete);
    // A node withheld AFTER completion must not un-complete the epoch — post
    // enumeration, withholding is per-object (the node is simply absent).
    m.apply(device_node(
        100,
        50,
        MediaRole::Sink,
        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
    ));
    assert_eq!(m.readiness(), Readiness::Complete);
    assert!(m.graph_ready());
    // ...and a late timeout Tick is inert once Complete.
    m.apply(RegEvent::Tick { now: 100_000 });
    assert_eq!(m.readiness(), Readiness::Complete);
}

#[test]
fn model_withheld_node_removed_clears_obligation() {
    let mut m = model();
    m.apply(device_node(
        100,
        50,
        MediaRole::Sink,
        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
    ));
    m.apply(RegEvent::ServerSynced);
    assert_eq!(m.readiness(), Readiness::Waiting);
    // The withheld node disappears before its device ever showed up.
    m.apply(RegEvent::Removed { id: gid(50) });
    assert_eq!(m.readiness(), Readiness::Complete);
}

// ==========================================================================
// model — device withholding & session_device flag
// ==========================================================================

#[test]
fn model_device_first_admits_node_immediately() {
    let mut m = model();
    // Device enumerated before the node that references it.
    m.apply(RegEvent::DeviceAdded { id: gid(42) });
    m.apply(device_node(
        100,
        50,
        MediaRole::Sink,
        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
    ));
    let snap = m.project().snapshot;
    let node = snap.node(ser(100)).expect("node admitted immediately");
    assert!(
        node.props.session_device,
        "hardware sink is a session device"
    );
    // No obligation ⇒ a sync completes readiness.
    m.apply(RegEvent::ServerSynced);
    assert!(m.graph_ready());
}

#[test]
fn model_withheld_node_admitted_with_correct_session_device() {
    let mut m = model();
    // A real hardware sink and a card-associated filter share client/device
    // ancestry but classify differently once the device resolves.
    m.apply(device_node(
        100,
        50,
        MediaRole::Sink,
        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
    ));
    m.apply(device_node(
        200,
        51,
        MediaRole::Sink,
        hw_claim(42, "alsa", "support.null-audio-sink"),
    ));
    // Both withheld until the device resolves.
    assert_eq!(m.project().snapshot.nodes().count(), 0);
    m.apply(RegEvent::DeviceAdded { id: gid(42) });

    let snap = m.project().snapshot;
    assert_eq!(
        snap.nodes().count(),
        2,
        "both admitted once the device resolved"
    );
    assert!(
        snap.node(ser(100)).unwrap().props.session_device,
        "the real hardware sink is a session device"
    );
    assert!(
        !snap.node(ser(200)).unwrap().props.session_device,
        "the null sink sharing the same device is not"
    );
}

#[test]
fn model_withheld_filter_admitted_as_not_session_device() {
    let mut m = model();
    m.apply(device_node(
        200,
        51,
        MediaRole::Sink,
        hw_claim(42, "alsa", "support.null-audio-sink"),
    ));
    m.apply(RegEvent::DeviceAdded { id: gid(42) });
    let snap = m.project().snapshot;
    assert!(
        !snap.node(ser(200)).unwrap().props.session_device,
        "a null sink on a card is not a session device"
    );
}