//! Integration test backfill for the 0.19 cycle.
//!
//! The plan audit (post-cycle review) surfaced kernel-touching
//! APIs from Plans 188 / 196 / 199 / 200 / 202 that shipped
//! with only unit-level test coverage. This file fills the gap
//! so the privileged-CI gate exercises the kernel round-trip
//! for each.
//!
//! All tests root-gated via `nlink::require_root!()` so they
//! skip cleanly on non-root developer machines and run for
//! real under `.github/workflows/integration-tests.yml`.

use std::time::Duration;

use nlink::Result;
use nlink::netlink::{
    Connection, Route,
    config::NetworkConfig,
    namespace,
};

use crate::common::TestNamespace;

/// 30-second timeout wrapper. Same shape as the existing
/// `network_config_apply.rs` helper — if a backfill test hangs
/// (likely the kernel-side surface broke), the CI gate fires
/// `Error::Timeout` rather than a 60-minute job timeout.
async fn with_timeout<F>(body: F) -> Result<()>
where
    F: std::future::Future<Output = Result<()>>,
{
    match tokio::time::timeout(Duration::from_secs(30), body).await {
        Ok(result) => result,
        Err(_elapsed) => Err(nlink::Error::Timeout),
    }
}

fn route_in_ns(ns: &TestNamespace) -> Result<Connection<Route>> {
    namespace::connection_for::<Route>(ns.name())
}

// =============================================================================
// Plan 188 — declarative apply parity
// =============================================================================

/// Plan 188 §2.1 — `ConfigDiff::apply` happy path.
///
/// Build a NetworkConfig, compute the diff, then call
/// `diff.apply()` (NOT `cfg.apply()`) so the diff-side method
/// gets the round-trip test the §4.7 acceptance criteria asked
/// for.
#[tokio::test]
async fn plan_188_config_diff_apply_round_trips() -> Result<()> {
    nlink::require_root!();
    with_timeout(async {
        let ns = TestNamespace::new("p188-cd-apply")?;
        let conn = route_in_ns(&ns)?;

        let cfg = NetworkConfig::new().link("eth0", |b| b.dummy());
        let diff = cfg.diff(&conn).await?;
        assert!(!diff.is_empty(), "fresh ns should have a non-empty diff");

        let result = diff
            .apply(&conn, nlink::netlink::config::ApplyOptions::default())
            .await?;
        assert_eq!(result.changes_made, 1);

        // Re-diff after apply — should be empty (idempotent).
        let cfg2 = NetworkConfig::new().link("eth0", |b| b.dummy());
        let diff2 = cfg2.diff(&conn).await?;
        assert!(
            diff2.is_empty(),
            "post-apply diff must be empty; got {diff2}"
        );
        Ok(())
    })
    .await
}

/// Plan 188 §2.4 — `NetworkConfig::apply_reconcile` happy path
/// (no transient errors → first apply succeeds, single attempt).
#[tokio::test]
async fn plan_188_apply_reconcile_first_attempt_succeeds() -> Result<()> {
    nlink::require_root!();
    with_timeout(async {
        let ns = TestNamespace::new("p188-reconcile")?;
        let conn = route_in_ns(&ns)?;

        let cfg = NetworkConfig::new().link("eth0", |b| b.dummy());
        let report = cfg
            .apply_reconcile(
                &conn,
                nlink::netlink::nftables::config::ReconcileOptions::default(),
            )
            .await?;
        assert_eq!(report.attempts, 1, "no transient errors expected");
        assert_eq!(report.change_count, 1);
        Ok(())
    })
    .await
}

/// Plan 188 §2.7 — `Connection<Nftables>::del_table_if_exists`
/// is idempotent: calling on a non-existent table returns
/// `Ok(())`, calling on an existing one deletes + returns
/// `Ok(())`, and a second call on the just-deleted table
/// still returns `Ok(())`.
#[tokio::test]
async fn plan_188_del_table_if_exists_is_idempotent() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("nf_tables");
    with_timeout(async {
        use nlink::netlink::Nftables;
        use nlink::netlink::nftables::types::Family;

        let ns = TestNamespace::new("p188-del-table")?;
        let conn = namespace::connection_for::<Nftables>(ns.name())?;

        // Cold: delete non-existent — must NOT error.
        conn.del_table_if_exists("absent-test", Family::Inet).await?;

        // Warm: add it, then delete via if_exists.
        conn.transaction()
            .add_table("test-table", Family::Inet)
            .commit(&conn)
            .await?;

        conn.del_table_if_exists("test-table", Family::Inet).await?;

        // Cold again: just-deleted table — must NOT error.
        conn.del_table_if_exists("test-table", Family::Inet).await?;

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 202 — RTA_MULTIPATH parser round-trip
// =============================================================================

/// Plan 202 §2.3 — the headline regression test. Write a
/// multipath route, dump it back, verify the nexthop list
/// survives the round-trip (pre-Plan-202 it was silently
/// dropped on the parse side).
#[tokio::test]
async fn plan_202_multipath_route_round_trips() -> Result<()> {
    nlink::require_root!();
    with_timeout(async {
        use std::net::{IpAddr, Ipv4Addr};
        use nlink::netlink::addr::Ipv4Address;
        use nlink::netlink::link::DummyLink;
        use nlink::netlink::route::{Ipv4Route, NextHop};

        let ns = TestNamespace::new("p202-mp")?;
        let conn = route_in_ns(&ns)?;

        // Two dummy egress interfaces + addresses for the
        // nexthops to land on.
        conn.add_link(DummyLink::new("eth0")).await?;
        conn.add_link(DummyLink::new("eth1")).await?;
        conn.set_link_up("eth0").await?;
        conn.set_link_up("eth1").await?;
        conn.add_address(Ipv4Address::new(
            "eth0",
            Ipv4Addr::new(10, 0, 0, 1),
            24,
        ))
        .await?;
        conn.add_address(Ipv4Address::new(
            "eth1",
            Ipv4Addr::new(10, 0, 1, 1),
            24,
        ))
        .await?;

        let r = Ipv4Route::new("192.0.2.0", 24).multipath(vec![
            NextHop::new()
                .gateway_v4(Ipv4Addr::new(10, 0, 0, 254))
                .dev("eth0"),
            NextHop::new()
                .gateway_v4(Ipv4Addr::new(10, 0, 1, 254))
                .dev("eth1"),
        ]);
        conn.add_route(r).await?;

        // Dump and find the route.
        let routes = conn.get_routes().await?;
        let target = Ipv4Addr::new(192, 0, 2, 0);
        let dumped = routes
            .iter()
            .find(|r| r.destination() == Some(&IpAddr::V4(target)))
            .expect("multipath route must appear in dump");

        // The headline assertion: nexthops survive parsing.
        let nhs = dumped
            .multipath()
            .expect("Plan 202 — multipath nexthops must NOT be dropped on parse");
        assert_eq!(nhs.len(), 2, "expected 2 nexthops, got {}", nhs.len());

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 200 — facade
// =============================================================================

/// Plan 200 §2.1 — `nlink::facade::apply::network_in_namespace`
/// composes correctly with `NetworkConfig`. Same coverage as
/// the direct `apply` call but goes through the one-liner.
#[tokio::test]
async fn plan_200_facade_apply_network_in_namespace() -> Result<()> {
    nlink::require_root!();
    with_timeout(async {
        let ns = TestNamespace::new("p200-facade")?;
        let cfg = NetworkConfig::new().link("eth0", |b| b.dummy());

        let result = nlink::facade::apply::network_in_namespace(ns.name(), &cfg).await?;
        assert_eq!(result.changes_made, 1);

        // Diff via facade should be empty post-apply.
        let diff = nlink::facade::diff::network_in_namespace(ns.name(), &cfg).await?;
        assert!(diff.is_empty(), "post-apply facade diff must be empty");

        Ok(())
    })
    .await
}

/// Plan 200 §2.4 — `Stack` orchestrates layers in dependency
/// order. This test only exercises the NetworkConfig layer
/// (so we don't require nftables / WireGuard modules in CI).
#[tokio::test]
async fn plan_200_stack_apply_network_only_layer() -> Result<()> {
    nlink::require_root!();
    with_timeout(async {
        let ns = TestNamespace::new("p200-stack")?;
        let stack = nlink::facade::Stack::new()
            .network(NetworkConfig::new().link("eth0", |b| b.dummy()));

        let report = stack.apply_in_namespace(ns.name()).await?;
        assert!(!report.is_noop(), "Stack with a non-empty layer is not a no-op");
        assert!(
            report.network.as_ref().is_some_and(|r| r.changes_made == 1),
            "network layer should report 1 change"
        );
        assert!(report.nftables_change_count.is_none());
        assert!(report.wireguard.is_none());

        // Re-applying the same Stack must be a no-op.
        let report2 = stack.apply_in_namespace(ns.name()).await?;
        assert!(report2.is_noop(), "re-apply must be no-op; got {report2:?}");

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 196 — declarative WireguardConfig
// =============================================================================

/// Plan 196 §2.3 — declarative WG round-trip happy path.
/// Creates a `wg`-kind link via NetworkConfig, then declares
/// a peer via WireguardConfig, verifies the kernel-side
/// `get_device_by_name` returns matching state.
///
/// Gated by `require_module!("wireguard")` so the test skips
/// cleanly on a kernel without the WG module loaded.
#[tokio::test]
async fn plan_196_wireguard_config_round_trips() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("wireguard");
    with_timeout(async {
        use std::net::{IpAddr, Ipv4Addr, SocketAddr};
        use nlink::netlink::Wireguard;
        use nlink::netlink::genl::wireguard::{AllowedIp, WireguardConfig};

        let ns = TestNamespace::new("p196-wg")?;

        // Pre-create wg0 via `ip` — NetworkConfig doesn't yet
        // have a wireguard() builder, and the integration
        // gate's Debian container ships iproute2.
        let status = std::process::Command::new("ip")
            .args([
                "netns", "exec", ns.name(),
                "ip", "link", "add", "wg0", "type", "wireguard",
            ])
            .status();
        let created = matches!(status, Ok(s) if s.success());
        if !created {
            eprintln!(
                "plan_196_wireguard_config_round_trips: skipped — couldn't ip link add wg0 type wireguard (kernel without wg mod or no iproute2)"
            );
            return Ok(());
        }

        // Connect to the WG GENL family inside the ns.
        let conn = namespace::connection_for_async::<Wireguard>(ns.name()).await?;

        let private_key = [0xaau8; 32];
        let peer_pk = [0xbbu8; 32];
        let endpoint = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)), 51820);

        let cfg = WireguardConfig::new().device("wg0", |d| {
            d.private_key(private_key)
                .listen_port(51820)
                .peer(peer_pk, |p| {
                    p.endpoint(endpoint)
                        .persistent_keepalive(Duration::from_secs(25))
                        .allowed_ip(AllowedIp::v4(Ipv4Addr::new(10, 0, 0, 0), 24))
                })
        });

        let result = cfg.apply(&conn).await?;
        assert!(
            result.total_writes() >= 2,
            "expected at least 1 device write + 1 peer write; got {result:?}"
        );

        // Verify kernel-side state.
        let device = conn.get_device_by_name("wg0").await?;
        assert_eq!(device.listen_port, Some(51820));
        assert_eq!(device.peers.len(), 1);
        assert_eq!(device.peers[0].public_key, peer_pk);
        assert_eq!(device.peers[0].endpoint, Some(endpoint));

        // Second apply must be idempotent at the peer level
        // (private_key always rewrites by design — see Plan
        // 196's `private_key` caveat in the module rustdoc).
        let result2 = cfg.apply(&conn).await?;
        assert_eq!(
            result2.peer_writes, 0,
            "no peer mutations on re-apply; got {result2:?}"
        );
        assert_eq!(
            result2.peer_removals, 0,
            "no peer removals on re-apply"
        );

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 199 — WireguardWatcher polling primitive
// =============================================================================

/// Plan 199 — `WireguardWatcher` first-poll emits PeerAdded
/// for every existing peer (initial-inventory semantics). The
/// kernel side is the same as Plan 196's test; this one
/// verifies the watcher exposes the kernel state correctly.
#[tokio::test]
async fn plan_199_watcher_first_poll_emits_initial_inventory() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("wireguard");
    with_timeout(async {
        use nlink::netlink::Wireguard;
        use nlink::netlink::genl::wireguard::{
            WireguardEvent, WireguardWatchOptions, WireguardWatcher,
        };

        let ns = TestNamespace::new("p199-watcher")?;
        let status = std::process::Command::new("ip")
            .args([
                "netns", "exec", ns.name(),
                "ip", "link", "add", "wg0", "type", "wireguard",
            ])
            .status();
        if !matches!(status, Ok(s) if s.success()) {
            eprintln!("plan_199_watcher: skipped — couldn't create wg0");
            return Ok(());
        }

        let conn = namespace::connection_for_async::<Wireguard>(ns.name()).await?;
        let opts = WireguardWatchOptions::default()
            .interval(Duration::from_millis(100))
            .interface("wg0");
        let mut watcher = WireguardWatcher::new(conn, opts)?;

        // First poll on an empty device emits NO events
        // (no peers yet).
        let events = watcher.next_events().await?;
        assert!(events.is_empty(), "no peers yet; expected empty, got {events:?}");

        // Add a peer out-of-band via `ip`.
        let _ = std::process::Command::new("ip")
            .args([
                "netns", "exec", ns.name(),
                "wg", "set", "wg0",
                "peer", "fE/wpxQ6/M6OmF5j4dvbY3FbCEXc3KlBL2QqAYjE0WI=",
                "allowed-ips", "10.0.0.0/24",
            ])
            .status();
        // `wg` tool may not be present in the CI container —
        // in that case skip the post-event verification.

        // Second poll should now emit PeerAdded IF the peer
        // was successfully added.
        let events = watcher.next_events().await?;
        if !events.is_empty() {
            assert!(
                events.iter().any(|e| matches!(e, WireguardEvent::PeerAdded { .. })),
                "second poll should fire PeerAdded; got {events:?}"
            );
        }

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 204 — wire-format CRITICAL fixes — root-gated integration tests
// =============================================================================

/// Plan 204 C4 regression — devlink mcast subscribe used to fail
/// with FamilyNotFound because nlink looked up `"devlink"` but the
/// kernel registers the group as `"config"`. Now subscribe must
/// succeed (or fail with a clear "no devlink kernel module" error,
/// not a name mismatch).
#[tokio::test]
async fn plan_204_c4_devlink_subscribe_resolves_config_group() -> Result<()> {
    nlink::require_root!();
    with_timeout(async {
        // Devlink kernel module is part of the kernel core on
        // every modern Linux; no module-load required. The
        // family resolution itself may still fail in restricted
        // CI containers — that's a different failure, not the
        // Plan 204 bug class.
        let conn = match nlink::netlink::Connection::<nlink::netlink::Devlink>::new_async().await {
            Ok(c) => c,
            Err(e) => {
                eprintln!(
                    "devlink family unavailable in this kernel/netns: {e}; \
                     skipping (Plan 204 C4 still pinned by the unit test)"
                );
                return Ok(());
            }
        };
        // Pre-Plan 204: this returned FamilyNotFound { name: "devlink::devlink" }.
        // Post-fix: resolves to the kernel's "config" group.
        conn.subscribe()?;
        Ok(())
    })
    .await
}

/// Plan 204 C2 regression — XfrmUserpolicyInfo was 4 bytes short
/// and `add_sp` rejected by every kernel with EINVAL. Now the body
/// is the kernel-expected 168 bytes and an SP add round-trips.
///
/// Note: requires `xfrm_user` kernel module and the test policy
/// must be syntactically valid for the kernel to accept.
#[tokio::test]
async fn plan_204_c2_xfrm_add_sp_round_trips() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("xfrm_user");
    with_timeout(async {
        let ns = TestNamespace::new("plan204-xfrm-sp")?;
        let _route = route_in_ns(&ns)?;

        let conn: nlink::netlink::Connection<nlink::netlink::Xfrm> =
            namespace::connection_for(ns.name())?;

        // Build a minimal SP and just verify list/dump succeeds.
        // Full add_sp wiring still requires builder integration
        // tests in xfrm.rs that aren't yet root-gated.
        // This test mainly proves the dump-side recv loop now
        // has timeout + seq filter (Plan 208 also applied here).
        let policies = conn.get_security_policies().await?;
        // Fresh netns has no policies — assert dump returns empty.
        assert!(policies.is_empty(), "fresh netns should have 0 SPs, got {}", policies.len());

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 207 — NetworkConfig correctness pass (kernel-level backfill)
// =============================================================================

/// Plan 207a M10 — `cfg.diff()` reads admin UP from `IFF_UP` on
/// the kernel-dumped link flags (not from `OperState`, which
/// only reflects operational L2 state).
///
/// Test: bring an iface UP out-of-band, then build a config
/// that declares the same iface as `.up()`. Assert the diff
/// is **empty** (pre-Plan-207a the diff thought it was DOWN
/// because OperState on a dummy stays UNKNOWN, even when
/// administratively UP).
#[tokio::test]
async fn plan_207a_diff_reads_iff_up_from_admin_flags() -> Result<()> {
    use nlink::netlink::link::DummyLink;
    nlink::require_root!();
    with_timeout(async {
        let ns = TestNamespace::new("p207a-iff-up")?;
        let conn = route_in_ns(&ns)?;

        // Bring d0 administratively UP via the imperative API.
        conn.add_link(DummyLink::new("d0")).await?;
        let idx = conn
            .get_link_by_name("d0")
            .await?
            .ok_or_else(|| nlink::Error::InvalidMessage("d0 not found".into()))?
            .ifindex();
        conn.set_link_up(idx).await?;

        // Now declare the same iface UP.
        let cfg = NetworkConfig::new().link("d0", |l| l.dummy().up());

        let diff = cfg.diff(&conn).await?;
        // 207a: the diff must see the existing UP state and
        // produce NO updates. Pre-fix the diff didn't read
        // IFF_UP and thought d0 was DOWN.
        assert!(
            diff.links_to_modify.is_empty(),
            "diff produced updates for an iface that's already in the declared state: {:?}",
            diff.links_to_modify
        );

        Ok(())
    })
    .await
}

/// Plan 207b H2 — `cfg.diff()` resolves a declared
/// `master = "br0"` against the kernel's ifindex-valued
/// IFLA_MASTER attribute on the slave link.
///
/// Test: create a bridge + a slave attached to it out-of-band,
/// then declare the same setup. Assert the diff is **empty**
/// (pre-Plan-207b the diff compared the declared name "br0"
/// against the parsed-as-u32 master ifindex on the slave, so
/// it always reported a "master changed" update).
#[tokio::test]
async fn plan_207b_diff_resolves_master_ifindex_to_name() -> Result<()> {
    use nlink::netlink::link::{BridgeLink, DummyLink};
    nlink::require_root!();
    with_timeout(async {
        let ns = TestNamespace::new("p207b-master")?;
        let conn = route_in_ns(&ns)?;

        conn.add_link(BridgeLink::new("br0")).await?;
        conn.add_link(DummyLink::new("slave0")).await?;
        let br_idx = conn
            .get_link_by_name("br0")
            .await?
            .ok_or_else(|| nlink::Error::InvalidMessage("br0 not found".into()))?
            .ifindex();
        let slave_idx = conn
            .get_link_by_name("slave0")
            .await?
            .ok_or_else(|| nlink::Error::InvalidMessage("slave0 not found".into()))?
            .ifindex();
        conn.set_link_master(slave_idx, br_idx).await?;

        // Declare the same setup.
        let cfg = NetworkConfig::new()
            .link("br0", |l| l.bridge())
            .link("slave0", |l| l.dummy().master("br0"));

        let diff = cfg.diff(&conn).await?;
        assert!(
            diff.links_to_modify.is_empty(),
            "207b regression — diff reports updates when master is already correct: {:?}",
            diff.links_to_modify
        );

        Ok(())
    })
    .await
}

/// Plan 207f M18 — `replace_qdisc` is atomic (NLM_F_REPLACE).
///
/// Test: install a netem qdisc, then replace it with a
/// different netem. Spin in a tight dump loop while the
/// replace runs; assert there is no transient window where
/// the qdisc dump returns empty. Pre-Plan-207f the lib did
/// `del_qdisc` + `add_qdisc` as two separate netlink ops; a
/// concurrent observer would see "no qdisc" between them.
///
/// The 207f fix sets `NLM_F_REPLACE` so the kernel swaps the
/// qdisc atomically. The atomicity is implicit in the kernel —
/// this test mainly proves no error path or extra round-trip
/// snuck a window in.
#[tokio::test]
async fn plan_207f_replace_qdisc_is_atomic() -> Result<()> {
    use nlink::netlink::link::DummyLink;
    use nlink::netlink::tc::NetemConfig;
    use nlink::TcHandle;
    nlink::require_root!();
    nlink::require_module!("sch_netem");
    with_timeout(async {
        let ns = TestNamespace::new("p207f-replace")?;
        let conn = route_in_ns(&ns)?;

        conn.add_link(DummyLink::new("d0")).await?;
        let idx = conn
            .get_link_by_name("d0")
            .await?
            .ok_or_else(|| nlink::Error::InvalidMessage("d0 not found".into()))?
            .ifindex();
        conn.set_link_up(idx).await?;

        // Install the original netem qdisc.
        let v1 = NetemConfig::new().delay(Duration::from_millis(50));
        conn.add_qdisc_full("d0", TcHandle::ROOT, Some(TcHandle::major_only(1)), v1.clone())
            .await?;

        // Replace with a new netem config; assert the result.
        let v2 = NetemConfig::new().delay(Duration::from_millis(100));
        conn.replace_qdisc("d0", v2.clone()).await?;

        // After replace, the qdisc must still be present.
        let qdiscs = conn.get_qdiscs().await?;
        let netem = qdiscs.iter().any(|q| q.kind() == Some("netem"));
        assert!(netem, "replace_qdisc left no netem qdisc on d0");

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 204 C1 — NFT_JUMP / NFT_GOTO verdict constants round-trip
// =============================================================================

/// Plan 204 C1 kernel-level regression test (test-coverage gap
/// agent flagged as CRITICAL — only unit-level constant checks
/// existed pre-fix). Pre-Plan-204 `Verdict::Jump(chain)` wrote
/// `-2` (which is `NFT_BREAK`) on the wire. The kernel either
/// silently treated it as BREAK (rule terminates evaluation)
/// or rejected the rule entirely, depending on attribute
/// validation ordering. Either way every jump rule was broken
/// since the `Verdict` enum shipped.
///
/// Post-fix: `Verdict::Jump = -3` and `Verdict::Goto = -4`
/// (matching kernel UAPI `nft_verdicts`).
///
/// The test installs a parent chain that jumps to a child
/// chain and verifies:
/// 1. The kernel ACCEPTS the rule (no commit error).
/// 2. `list_rules` returns the rule.
/// 3. The raw `expression_bytes` contain the i32 LE encoding
///    of `NFT_JUMP = -3` (`0xfd 0xff 0xff 0xff`), proving the
///    correct constant survived to the kernel.
///
/// A regression that re-introduces the old `-2` constant
/// would: (a) likely fail step 1 (kernel rejects), or (b)
/// fail step 3 (the verdict bytes contain `0xfe` instead of
/// `0xfd`).
#[tokio::test]
async fn plan_204_c1_verdict_jump_round_trips_through_kernel() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("nf_tables");
    with_timeout(async {
        use nlink::netlink::Nftables;
        use nlink::netlink::nftables::types::{Chain, Family, Hook, Rule};

        let ns = TestNamespace::new("p204-c1-jump")?;
        let conn = namespace::connection_for::<Nftables>(ns.name())?;

        // Stand up a table with a parent chain (hooks input,
        // base chain) and a child chain (regular, no hook —
        // jump target).
        let parent = Chain::new("t", "parent")
            .family(Family::Inet)
            .hook(Hook::Input)
            .priority(nlink::netlink::nftables::Priority::Custom(0));
        let child = Chain::new("t", "child").family(Family::Inet);

        conn.transaction()
            .add_table("t", Family::Inet)
            .add_chain(parent)
            .add_chain(child)
            .commit(&conn)
            .await?;

        // The rule: when a packet arrives at parent's input hook,
        // jump to child. Plan 204 fixes mean this should install
        // without error.
        let jump_rule = Rule::new("t", "parent")
            .family(Family::Inet)
            .try_jump("child")?;

        conn.transaction()
            .add_rule(jump_rule)
            .commit(&conn)
            .await?;

        // Step 2: rule must dump back.
        let rules = conn.list_rules("t", Family::Inet).await?;
        let parent_rules: Vec<_> = rules
            .iter()
            .filter(|r| r.chain == "parent")
            .collect();
        assert_eq!(
            parent_rules.len(),
            1,
            "expected 1 rule in parent chain after jump install, got {}",
            parent_rules.len()
        );

        // Step 3: the raw expression bytes must contain
        // `NFT_JUMP = -3` encoded *big-endian* — empirical
        // observation from CI shows the kernel emits
        // `NFTA_VERDICT_CODE` as `__be32` (the netfilter rule
        // attributes use BE for verdict codes, even though most
        // nftables NLA_U32 are native). The verdict bytes appear
        // as `[ff, ff, ff, fd]` in the dumped NFTA_RULE_EXPRESSIONS.
        // Pre-Plan-204 this would be `[ff, ff, ff, fe]` (NFT_BREAK).
        let expr_bytes = &parent_rules[0].expression_bytes;
        let jump_marker_be = (-3_i32).to_be_bytes();
        let break_marker_be = (-2_i32).to_be_bytes();
        let contains_jump = expr_bytes
            .windows(4)
            .any(|w| w == jump_marker_be.as_slice());
        let contains_break_with_chain = expr_bytes
            .windows(4)
            .any(|w| w == break_marker_be.as_slice());
        assert!(
            contains_jump,
            "rule expression bytes must contain NFT_JUMP marker (-3 BE): bytes = {expr_bytes:02x?}"
        );
        assert!(
            !contains_break_with_chain,
            "rule expression bytes must NOT contain NFT_BREAK marker (-2 BE) — \
             pre-Plan-204 bug regressed: bytes = {expr_bytes:02x?}"
        );

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 211 M1 — Hook::InetIngress kernel acceptance
// =============================================================================

/// Plan 211 M1 kernel-level regression (test-coverage gap agent
/// flagged as CRITICAL — only unit-level constant checks pre-fix).
///
/// Pre-Plan-211, `Hook::Ingress` on `Family::Inet` silently
/// aliased to the wrong hook value (PREROUTING instead of the
/// kernel-5.10+ INGRESS hook value 5). After the family-aware
/// `Hook::Ingress` → `Hook::InetIngress` split, installing a
/// chain on the `Inet` family with `Hook::InetIngress` must:
/// 1. Be accepted by a kernel 5.10+
/// 2. NOT collide with any existing Prerouting chain on the
///    same family (proves the hook value is genuinely different)
///
/// The test installs both a `Prerouting` chain AND an
/// `InetIngress` chain on the same Inet table at the same
/// priority. If `InetIngress` were aliasing to `Prerouting` (the
/// pre-fix shape), the kernel would reject the second chain
/// with EEXIST/EBUSY. Both succeeding proves the hook values
/// are distinct.
///
/// Skips cleanly on kernels older than 5.10 (the kernel
/// returns EOPNOTSUPP / EINVAL on `NF_INET_INGRESS` — we
/// detect the error and `Ok(())` out).
#[tokio::test]
async fn plan_211_m1_inet_ingress_chain_installs_on_correct_hook() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("nf_tables");
    with_timeout(async {
        use nlink::netlink::Nftables;
        use nlink::netlink::nftables::types::{Chain, Family, Hook};

        let ns = TestNamespace::new("p211-m1-ingress")?;
        let conn = namespace::connection_for::<Nftables>(ns.name())?;

        // Install the Prerouting chain first — always supported.
        let prerouting = Chain::new("t", "pre")
            .family(Family::Inet)
            .hook(Hook::Prerouting)
            .priority(nlink::netlink::nftables::Priority::Custom(0));
        conn.transaction()
            .add_table("t", Family::Inet)
            .add_chain(prerouting)
            .commit(&conn)
            .await?;

        // Now try the InetIngress chain on the same table at the
        // same priority. Pre-Plan-211 this would EEXIST against
        // the Prerouting chain because both aliased to the same
        // hook value. Post-fix InetIngress = NF_INET_INGRESS (5).
        //
        // NetdevIngress chains require a `device` attribute;
        // InetIngress on the Inet family doesn't (it's process-
        // global ingress, no per-iface bind).
        let inet_ingress = Chain::new("t", "ingress")
            .family(Family::Inet)
            .hook(Hook::InetIngress)
            .priority(nlink::netlink::nftables::Priority::Custom(0));

        match conn
            .transaction()
            .add_chain(inet_ingress)
            .commit(&conn)
            .await
        {
            Ok(()) => {
                // Verify both chains live in the table.
                let chains = conn
                    .list_chains_in("t", Family::Inet)
                    .await?;
                let pre_count = chains.iter().filter(|c| c.name == "pre").count();
                let ing_count = chains.iter().filter(|c| c.name == "ingress").count();
                assert_eq!(pre_count, 1, "prerouting chain must persist");
                assert_eq!(ing_count, 1, "inet ingress chain must persist");
            }
            Err(e) if e.is_invalid_argument() || e.is_not_supported() => {
                // Kernel < 5.10 — NF_INET_INGRESS not supported.
                // Skip with a tracing note so CI logs show why.
                tracing::warn!(
                    error = %e,
                    "kernel doesn't support NF_INET_INGRESS — skipping (need >= 5.10)"
                );
            }
            Err(e) => return Err(e),
        }

        Ok(())
    })
    .await
}

// =============================================================================
// Plan 191 — Route subscribe_all_with_resync — initial inventory walk
// =============================================================================

/// Plan 191 kernel-level regression (test-coverage gap agent
/// flagged as HIGH — only the Nftables-side of the resync
/// shape had a kernel test).
///
/// `subscribe_all_with_resync` returns a `ResyncStream` that:
/// - wraps live multicast events as `ResyncedEvent::Event(_)`
/// - on `ENOBUFS`, replays the snapshot wrapped as
///   `ResyncedEvent::Marker(ResyncStart)` → `Resynced(_)`* →
///   `Marker(ResyncEnd)`.
///
/// This test exercises the live-event path: after subscribing,
/// add a dummy link out-of-band, and assert the stream yields a
/// `ResyncedEvent::Event(NewLink)` for it. Pre-Plan-191 the
/// Route-side `subscribe_all_with_resync` didn't exist and
/// users had no way to compose route multicast with ENOBUFS
/// recovery; this test guards against a silent regression that
/// would, e.g., drop the wrapper around live events.
///
/// (The ENOBUFS-driven snapshot walk path itself is exercised
/// by `into_events_with_resync_recovers_from_enobufs` in
/// `nftables_reconcile.rs`; the route side uses the same
/// `ResyncStream` glue so the ENOBUFS path is covered
/// transitively.)
#[tokio::test]
async fn plan_191_route_subscribe_with_resync_emits_live_events() -> Result<()> {
    use nlink::netlink::events::NetworkEvent;
    use nlink::netlink::link::DummyLink;
    use nlink::netlink::resync::ResyncedEvent;
    use std::sync::Arc;
    use tokio_stream::StreamExt;

    nlink::require_root!();
    with_timeout(async {
        let ns = TestNamespace::new("p191-resync")?;

        // Subscribe FIRST so the upcoming link-add fires as a
        // live multicast event.
        let conn = route_in_ns(&ns)?;
        let ns_name = ns.name().to_string();
        let factory: nlink::netlink::resync::ConnectionFactory<Route> = Arc::new(move || {
            let ns = ns_name.clone();
            Box::pin(async move { namespace::connection_for::<Route>(&ns) })
        });

        let mut stream = conn
            .subscribe_all_with_resync(factory)
            .await?;

        // Yield to the runtime so the multicast subscription is
        // actually registered with the kernel before the mutator
        // runs.
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Add a dummy link via a SECOND connection (the first is
        // pinned to the events stream by the request lock).
        let mutator = route_in_ns(&ns)?;
        mutator.add_link(DummyLink::new("d0")).await?;
        drop(mutator);

        // Drain the stream looking for a `Event(NewLink("d0"))`.
        // Bound at 5s — multicast events arrive within ms in
        // practice.
        let mut saw_d0 = false;
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        while !saw_d0 {
            let remaining = deadline
                .checked_duration_since(tokio::time::Instant::now())
                .unwrap_or_default();
            if remaining.is_zero() {
                break;
            }
            match tokio::time::timeout(remaining, stream.next()).await {
                Ok(Some(Ok(ResyncedEvent::Event(NetworkEvent::NewLink(link)))))
                    if link.name() == Some("d0") =>
                {
                    saw_d0 = true;
                }
                Ok(Some(Ok(_))) => continue, // unrelated event; keep looking
                Ok(Some(Err(e))) => return Err(e),
                Ok(None) => break,
                Err(_) => break,
            }
        }

        assert!(
            saw_d0,
            "subscribe_all_with_resync stream did not emit a live NewLink event for the d0 add — \
             the wrapper around live multicast events may have regressed"
        );

        Ok(())
    })
    .await
}

/// Plan 204 C1 sibling — same shape for `Verdict::Goto`. Goto
/// pre-fix wrote `-3` which is now `NFT_JUMP` post-fix. The
/// roundtrip must show `NFT_GOTO = -4` instead.
#[tokio::test]
async fn plan_204_c1_verdict_goto_round_trips_through_kernel() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("nf_tables");
    with_timeout(async {
        use nlink::netlink::Nftables;
        use nlink::netlink::nftables::types::{Chain, Family, Hook, Rule};

        let ns = TestNamespace::new("p204-c1-goto")?;
        let conn = namespace::connection_for::<Nftables>(ns.name())?;

        let parent = Chain::new("t", "parent")
            .family(Family::Inet)
            .hook(Hook::Input)
            .priority(nlink::netlink::nftables::Priority::Custom(0));
        let child = Chain::new("t", "child").family(Family::Inet);

        conn.transaction()
            .add_table("t", Family::Inet)
            .add_chain(parent)
            .add_chain(child)
            .commit(&conn)
            .await?;

        let goto_rule = Rule::new("t", "parent")
            .family(Family::Inet)
            .try_goto("child")?;

        conn.transaction()
            .add_rule(goto_rule)
            .commit(&conn)
            .await?;

        let rules = conn.list_rules("t", Family::Inet).await?;
        let parent_rules: Vec<_> = rules
            .iter()
            .filter(|r| r.chain == "parent")
            .collect();
        assert_eq!(parent_rules.len(), 1);

        let expr_bytes = &parent_rules[0].expression_bytes;
        // 0.19 fix — NFTA_VERDICT_CODE is BE on the wire; see Jump test above.
        let goto_marker_be = (-4_i32).to_be_bytes();
        let contains_goto = expr_bytes
            .windows(4)
            .any(|w| w == goto_marker_be.as_slice());
        assert!(
            contains_goto,
            "rule expression bytes must contain NFT_GOTO marker (-4 BE): bytes = {expr_bytes:02x?}"
        );

        Ok(())
    })
    .await
}

/// WireGuard private-key readback round-trip.
///
/// Verifies that `get_device_by_name` returns `Some(private_key)` for a
/// privileged caller after `set_device` wrote the key. This is the kernel
/// acceptance gate for the fix to `parse_device_attrs` (the `PrivateKey`
/// attribute was previously silently dropped).
#[tokio::test]
async fn wg_private_key_readback_round_trips() -> Result<()> {
    nlink::require_root!();
    nlink::require_module!("wireguard");
    with_timeout(async {
        use nlink::netlink::Wireguard;

        let ns = TestNamespace::new("wg-pkey-rb")?;

        // The module is present (require_module! above), so a failure here is
        // a genuine environment fault, not a skip condition — fail hard rather
        // than green-washing a broken parser.
        let status = std::process::Command::new("ip")
            .args([
                "netns", "exec", ns.name(),
                "ip", "link", "add", "wg0", "type", "wireguard",
            ])
            .status()?;
        assert!(
            status.success(),
            "ip link add wg0 type wireguard failed in netns {}",
            ns.name()
        );

        let conn = namespace::connection_for_async::<Wireguard>(ns.name()).await?;

        let private_key = [0xCDu8; 32];
        conn.set_device("wg0", |b| b.private_key(private_key)).await?;

        // The kernel clamps the X25519 secret on store
        // (curve25519_clamp_secret in wg_noise_set_static_identity_private_key),
        // so GET_DEVICE returns the clamped key, not the bytes we set.
        let mut expected = private_key;
        expected[0] &= 248;
        expected[31] = (expected[31] & 127) | 64;

        let device = conn.get_device_by_name("wg0").await?;
        assert_eq!(
            device.private_key,
            Some(expected),
            "clamped private key must round-trip through GET_DEVICE for a privileged caller"
        );

        Ok(())
    })
    .await
}
