netconf_rust/

message.rs

use std::fmt;
use std::io::{Cursor, Write};
use std::ops::Deref;
use std::sync::atomic::{AtomicU32, Ordering};

use bytes::Bytes;
use quick_xml::Reader;
use quick_xml::Writer;
use quick_xml::events::{BytesDecl, BytesEnd, BytesStart, Event};

const NETCONF_NS: &str = "urn:ietf:params:xml:ns:netconf:base:1.0";

// Global message-id counter shared across all sessions.
//
// A NETCONF session is a single SSH connection to a device. Each session
// has its own stream of RPCs and replies.
//
// Each RPC needs a unique message-id so the client can match replies to
// requests. This is needed for pipelining, where multiple RPCs are
// sent without waiting for replies — when the replies arrive, the
// message-id is how we know which reply belongs to which request.
//
// IDs only need to be unique within a
// session, and globally unique is a superset of that.
//
// Relaxed ordering: we only need each fetch_add to return a different
// number. No other memory operations depend on this value, so we don't
// need the stronger (and more expensive) ordering guarantees.
static MESSAGE_ID_COUNTER: AtomicU32 = AtomicU32::new(1);

pub fn next_message_id() -> u32 {
    MESSAGE_ID_COUNTER.fetch_add(1, Ordering::Relaxed)
}

/// Zero-copy wrapper around a `Bytes` buffer with span offsets pointing to the
/// `<data>` content within an RPC reply.
///
/// `DataPayload` avoids copying the XML content into a `String`. The underlying
/// `Bytes` is reference-counted (O(1) clone), and `as_str()` returns a view
/// without allocation. Users who need a `String` can call `into_string()`.
///
/// For streaming XML processing, `reader()` returns a `quick_xml::Reader` that
/// borrows directly from the payload's bytes — events reference the same memory
/// with no intermediate copies.
#[derive(Clone)]
pub struct DataPayload {
    bytes: Bytes,
    start: usize,
    end: usize,
}

impl DataPayload {
    /// Create a new `DataPayload` referencing a slice of `bytes`.
    ///
    /// # Safety contract
    /// The caller must ensure `bytes[start..end]` is valid UTF-8.
    /// This is guaranteed when called from the parser, which validates
    /// UTF-8 before parsing.
    pub(crate) fn new(bytes: Bytes, start: usize, end: usize) -> Self {
        debug_assert!(start <= end);
        debug_assert!(end <= bytes.len());
        Self { bytes, start, end }
    }

    /// Create an empty `DataPayload` (for `<data/>` responses).
    pub(crate) fn empty() -> Self {
        Self {
            bytes: Bytes::new(),
            start: 0,
            end: 0,
        }
    }

    /// View the data content as `&str` without copying.
    ///
    /// This is O(1) — no allocation or UTF-8 validation. The bytes were
    /// validated as UTF-8 by the reader task before parsing.
    pub fn as_str(&self) -> &str {
        // SAFETY: The reader task validates UTF-8 (via std::str::from_utf8)
        // before the parser constructs a DataPayload. The bytes are immutable
        // (Bytes is ref-counted), so the UTF-8 invariant is preserved.
        unsafe { std::str::from_utf8_unchecked(&self.bytes[self.start..self.end]) }
    }

    /// Convert to an owned `String`.
    ///
    /// This copies the data content. Use `as_str()` to avoid the copy
    /// when you only need a `&str` view.
    pub fn into_string(self) -> String {
        self.as_str().to_string()
    }

    /// View the data content as `&[u8]`.
    pub fn as_bytes(&self) -> &[u8] {
        &self.bytes[self.start..self.end]
    }

    /// Get the underlying `Bytes` slice covering just the data content.
    pub fn slice(&self) -> Bytes {
        self.bytes.slice(self.start..self.end)
    }

    /// Length of the data content in bytes.
    pub fn len(&self) -> usize {
        self.end - self.start
    }

    /// Returns `true` if the data content is empty.
    pub fn is_empty(&self) -> bool {
        self.start == self.end
    }

    /// Create a `quick_xml::Reader` over the data content.
    ///
    /// Events borrow directly from the payload's bytes — no intermediate copies.
    /// This enables StAX-style streaming processing of large responses.
    pub fn reader(&self) -> Reader<&[u8]> {
        Reader::from_reader(self.as_bytes())
    }
}

impl fmt::Debug for DataPayload {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let s = self.as_str();
        if s.len() > 200 {
            write!(f, "DataPayload({} bytes: {:?}...)", s.len(), &s[..200])
        } else {
            write!(f, "DataPayload({:?})", s)
        }
    }
}

impl fmt::Display for DataPayload {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

impl Deref for DataPayload {
    type Target = str;

    fn deref(&self) -> &str {
        self.as_str()
    }
}

impl AsRef<str> for DataPayload {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl AsRef<[u8]> for DataPayload {
    fn as_ref(&self) -> &[u8] {
        self.as_bytes()
    }
}

impl PartialEq<str> for DataPayload {
    fn eq(&self, other: &str) -> bool {
        self.as_str() == other
    }
}

impl PartialEq<&str> for DataPayload {
    fn eq(&self, other: &&str) -> bool {
        self.as_str() == *other
    }
}

/// Build an RPC envelope around inner XML content using quick-xml's Writer.
pub fn build_rpc(inner_xml: &str) -> (u32, String) {
    let id = next_message_id();
    let id_str = id.to_string();
    let mut writer = Writer::new(Cursor::new(Vec::new()));

    writer
        .write_event(Event::Decl(BytesDecl::new("1.0", Some("UTF-8"), None)))
        .unwrap();
    writer.get_mut().write_all(b"\n").unwrap();

    let mut rpc = BytesStart::new("rpc");
    rpc.push_attribute(("message-id", id_str.as_str()));
    rpc.push_attribute(("xmlns", NETCONF_NS));
    writer.write_event(Event::Start(rpc)).unwrap();

    // Inner XML is written verbatim — it may be a simple <get/> or complex filter XML
    writer.get_mut().write_all(b"\n  ").unwrap();
    writer.get_mut().write_all(inner_xml.as_bytes()).unwrap();
    writer.get_mut().write_all(b"\n").unwrap();

    writer
        .write_event(Event::End(BytesEnd::new("rpc")))
        .unwrap();

    let bytes = writer.into_inner().into_inner();
    (id, String::from_utf8(bytes).unwrap())
}

#[derive(Debug)]
pub struct RpcReply {
    pub message_id: u32,
    pub body: RpcReplyBody,
}

impl RpcReply {
    /// Extract the data payload from this reply, or propagate an error.
    ///
    /// Returns `Ok(DataPayload)` for `Data` replies, `Ok(DataPayload::empty())`
    /// for `Ok` replies, and `Err` for error replies.
    pub fn into_data(self) -> crate::Result<DataPayload> {
        match self.body {
            RpcReplyBody::Data(payload) => Ok(payload),
            RpcReplyBody::Ok => Ok(DataPayload::empty()),
            RpcReplyBody::Error(errors) => Err(crate::Error::Rpc {
                message_id: self.message_id,
                error: errors
                    .first()
                    .map(|e| e.error_message.clone())
                    .unwrap_or_default(),
            }),
        }
    }
}

#[derive(Debug)]
pub enum RpcReplyBody {
    Ok,
    Data(DataPayload),
    Error(Vec<RpcError>),
}

#[derive(Debug, Default)]
pub struct RpcError {
    pub error_type: String,
    pub error_tag: String,
    pub error_severity: String,
    pub error_message: String,
}

/// A message received from the NETCONF server.
#[derive(Debug)]
pub enum ServerMessage {
    RpcReply(RpcReply),
    //Notification(Notification),
}

/// Classify and parse an incoming server message in a single pass.
///
/// Creates one `Reader`, identifies the root element, then continues
/// parsing with the same reader — no double-parsing.
///
/// Takes ownership of `Bytes` from the codec so that `DataPayload` can
/// reference the buffer without copying.
pub fn classify_message(bytes: Bytes) -> crate::Result<ServerMessage> {
    let xml = std::str::from_utf8(&bytes)
        .map_err(|_| crate::Error::UnexpectedResponse("received non-UTF-8 message".into()))?;

    let mut reader = Reader::from_str(xml);
    reader.config_mut().trim_text(true);

    loop {
        match reader.read_event() {
            Ok(Event::Start(e)) => {
                let local = e.local_name();
                match local.as_ref() {
                    b"rpc-reply" => {
                        let reply = parse_rpc_reply_body(xml, &bytes, &mut reader, &e)?;
                        return Ok(ServerMessage::RpcReply(reply));
                    }
                    b"notification" => {
                        unimplemented!("notification not implemented yet")
                    }
                    other => {
                        return Err(crate::Error::UnexpectedResponse(format!(
                            "unknown root element: <{}>",
                            String::from_utf8_lossy(other)
                        )));
                    }
                }
            }
            Ok(Event::Empty(e)) => {
                let local = e.local_name();
                match local.as_ref() {
                    b"rpc-reply" => {
                        let message_id = extract_message_id(&e)?;
                        return Ok(ServerMessage::RpcReply(RpcReply {
                            message_id,
                            body: RpcReplyBody::Ok,
                        }));
                    }
                    other => {
                        return Err(crate::Error::UnexpectedResponse(format!(
                            "unknown root element: <{}>",
                            String::from_utf8_lossy(other)
                        )));
                    }
                }
            }
            Ok(Event::Decl(_)) | Ok(Event::Comment(_)) | Ok(Event::PI(_)) => continue,
            Ok(Event::Eof) => {
                return Err(crate::Error::UnexpectedResponse(
                    "empty message: no root element".into(),
                ));
            }
            Err(e) => {
                return Err(crate::Error::UnexpectedResponse(format!(
                    "XML parse error: {e}"
                )));
            }
            _ => continue,
        }
    }
}

/// Standalone entry point for parsing an `<rpc-reply>` (used by tests).
pub fn parse_rpc_reply(xml: &str) -> crate::Result<RpcReply> {
    let bytes = Bytes::from(xml.to_string());
    let mut reader = Reader::from_str(xml);
    reader.config_mut().trim_text(true);

    loop {
        match reader.read_event() {
            Ok(Event::Start(e)) if e.local_name().as_ref() == b"rpc-reply" => {
                return parse_rpc_reply_body(xml, &bytes, &mut reader, &e);
            }
            Ok(Event::Empty(e)) if e.local_name().as_ref() == b"rpc-reply" => {
                let message_id = extract_message_id(&e)?;
                return Ok(RpcReply {
                    message_id,
                    body: RpcReplyBody::Ok,
                });
            }
            Ok(Event::Eof) => {
                return Err(crate::Error::UnexpectedResponse(
                    "no rpc-reply element found".into(),
                ));
            }
            Err(e) => {
                return Err(crate::Error::UnexpectedResponse(format!(
                    "XML parse error: {e}"
                )));
            }
            _ => continue,
        }
    }
}

/// Extract `message-id` from a `<rpc-reply>` element using `try_get_attribute`.
fn extract_message_id(e: &BytesStart<'_>) -> crate::Result<u32> {
    let attr = e
        .try_get_attribute("message-id")
        .map_err(|err| {
            crate::Error::UnexpectedResponse(format!("invalid rpc-reply attributes: {err}"))
        })?
        .ok_or_else(|| {
            crate::Error::UnexpectedResponse("missing message-id in rpc-reply".into())
        })?;

    let val = attr.unescape_value().map_err(|err| {
        crate::Error::UnexpectedResponse(format!("invalid message-id attr: {err}"))
    })?;

    val.parse::<u32>()
        .map_err(|e| crate::Error::UnexpectedResponse(format!("invalid message-id '{val}': {e}")))
}

/// Parse the body of an `<rpc-reply>` after the root Start event has been consumed.
///
/// Uses span-based slicing for `<data>` content (zero-copy from the original XML)
/// and inline parsing for `<rpc-error>` (no re-parse of the document).
///
/// The `raw` parameter is the ref-counted `Bytes` buffer from the codec, used
/// to construct `DataPayload` without copying.
fn parse_rpc_reply_body(
    xml: &str,
    raw: &Bytes,
    reader: &mut Reader<&[u8]>,
    root: &BytesStart<'_>,
) -> crate::Result<RpcReply> {
    let message_id = extract_message_id(root)?;
    let mut body: Option<RpcReplyBody> = None;

    loop {
        match reader.read_event() {
            Ok(Event::Start(e)) => {
                let local = e.local_name();
                match local.as_ref() {
                    b"data" => {
                        // Span-based: read_to_end returns byte offsets into the original XML.
                        // Slice the original string directly — zero-copy, exact preservation.
                        let span = reader.read_to_end(e.name()).map_err(|e| {
                            crate::Error::UnexpectedResponse(format!(
                                "XML parse error in <data>: {e}"
                            ))
                        })?;
                        let inner = xml[span.start as usize..span.end as usize].trim();
                        let trimmed_start = inner.as_ptr() as usize - xml.as_ptr() as usize;
                        let trimmed_end = trimmed_start + inner.len();
                        body = Some(RpcReplyBody::Data(DataPayload::new(
                            raw.clone(),
                            trimmed_start,
                            trimmed_end,
                        )));
                    }
                    b"rpc-error" => {
                        // Inline error parsing: continue with the current reader
                        // instead of re-parsing the entire document.
                        let first_error = parse_single_rpc_error(reader)?;
                        let mut errors = vec![first_error];

                        // Check for additional <rpc-error> elements
                        loop {
                            match reader.read_event() {
                                Ok(Event::Start(e2))
                                    if e2.local_name().as_ref() == b"rpc-error" =>
                                {
                                    errors.push(parse_single_rpc_error(reader)?);
                                }
                                Ok(Event::End(_)) | Ok(Event::Eof) => break,
                                Err(e) => {
                                    return Err(crate::Error::UnexpectedResponse(format!(
                                        "XML parse error: {e}"
                                    )));
                                }
                                _ => {}
                            }
                        }

                        body = Some(RpcReplyBody::Error(errors));
                        break;
                    }
                    _ => {}
                }
            }
            Ok(Event::Empty(e)) => match e.local_name().as_ref() {
                b"ok" => body = Some(RpcReplyBody::Ok),
                b"data" => body = Some(RpcReplyBody::Data(DataPayload::empty())),
                _ => {}
            },
            Ok(Event::Eof) => break,
            Err(e) => {
                return Err(crate::Error::UnexpectedResponse(format!(
                    "XML parse error: {e}"
                )));
            }
            _ => {}
        }
    }

    let body = body.unwrap_or(RpcReplyBody::Ok);
    Ok(RpcReply { message_id, body })
}

/// Fields within an `<rpc-error>` element.
#[derive(Clone, Copy)]
enum ErrorField {
    Type,
    Tag,
    Severity,
    Message,
}

/// Parse a single `<rpc-error>` block from the current reader position.
///
/// Called after the `<rpc-error>` Start event has been consumed.
/// Reads until the matching `</rpc-error>` End event.
fn parse_single_rpc_error(reader: &mut Reader<&[u8]>) -> crate::Result<RpcError> {
    let mut error = RpcError::default();
    let mut current_field: Option<ErrorField> = None;

    loop {
        match reader.read_event() {
            Ok(Event::Start(e)) => {
                current_field = match e.local_name().as_ref() {
                    b"error-type" => Some(ErrorField::Type),
                    b"error-tag" => Some(ErrorField::Tag),
                    b"error-severity" => Some(ErrorField::Severity),
                    b"error-message" => Some(ErrorField::Message),
                    _ => None,
                };
            }
            Ok(Event::Text(e)) => {
                if let Some(field) = current_field {
                    let text = e.xml_content().unwrap_or_default().to_string();
                    match field {
                        ErrorField::Type => error.error_type = text,
                        ErrorField::Tag => error.error_tag = text,
                        ErrorField::Severity => error.error_severity = text,
                        ErrorField::Message => error.error_message = text,
                    }
                }
            }
            Ok(Event::End(e)) => {
                if e.local_name().as_ref() == b"rpc-error" {
                    break;
                }
                current_field = None;
            }
            Ok(Event::Eof) => break,
            Err(e) => {
                return Err(crate::Error::UnexpectedResponse(format!(
                    "XML parse error: {e}"
                )));
            }
            _ => {}
        }
    }

    Ok(error)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_build_rpc() {
        let (id, xml) = build_rpc("<get/>");
        assert!(id > 0);
        assert!(xml.contains(&format!("message-id=\"{id}\"")));
        assert!(xml.contains("<get/>"));
        assert!(xml.contains("<rpc"));
        assert!(xml.contains("</rpc>"));
    }

    #[test]
    fn test_parse_ok_reply() {
        let xml = r#"<rpc-reply message-id="1" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <ok/>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        assert_eq!(reply.message_id, 1);
        assert!(matches!(reply.body, RpcReplyBody::Ok));
    }

    #[test]
    fn test_parse_data_reply() {
        let xml = r#"<rpc-reply message-id="2" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data>
    <interfaces xmlns="urn:example:interfaces">
      <interface>
        <name>eth0</name>
      </interface>
    </interfaces>
  </data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        assert_eq!(reply.message_id, 2);
        match &reply.body {
            RpcReplyBody::Data(data) => {
                assert!(data.contains("<interfaces"));
                assert!(data.contains("eth0"));
            }
            _ => panic!("expected Data reply"),
        }
    }

    #[test]
    fn test_parse_error_reply() {
        let xml = r#"<rpc-reply message-id="3" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <rpc-error>
    <error-type>application</error-type>
    <error-tag>invalid-value</error-tag>
    <error-severity>error</error-severity>
    <error-message>Invalid input</error-message>
  </rpc-error>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        assert_eq!(reply.message_id, 3);
        match &reply.body {
            RpcReplyBody::Error(errors) => {
                assert_eq!(errors.len(), 1);
                assert_eq!(errors[0].error_type, "application");
                assert_eq!(errors[0].error_tag, "invalid-value");
                assert_eq!(errors[0].error_severity, "error");
                assert_eq!(errors[0].error_message, "Invalid input");
            }
            _ => panic!("expected Error reply"),
        }
    }

    #[test]
    fn test_parse_multiple_errors() {
        let xml = r#"<rpc-reply message-id="10" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <rpc-error>
    <error-type>application</error-type>
    <error-tag>invalid-value</error-tag>
    <error-severity>error</error-severity>
    <error-message>First error</error-message>
  </rpc-error>
  <rpc-error>
    <error-type>protocol</error-type>
    <error-tag>bad-element</error-tag>
    <error-severity>error</error-severity>
    <error-message>Second error</error-message>
  </rpc-error>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        assert_eq!(reply.message_id, 10);
        match &reply.body {
            RpcReplyBody::Error(errors) => {
                assert_eq!(errors.len(), 2);
                assert_eq!(errors[0].error_message, "First error");
                assert_eq!(errors[1].error_message, "Second error");
                assert_eq!(errors[1].error_type, "protocol");
            }
            _ => panic!("expected Error reply"),
        }
    }

    #[test]
    fn test_parse_empty_data_reply() {
        let xml = r#"<rpc-reply message-id="4" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data/>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        assert_eq!(reply.message_id, 4);
        assert!(matches!(reply.body, RpcReplyBody::Data(ref s) if s.is_empty()));
    }

    #[test]
    fn test_message_ids_increment() {
        let (id1, _) = build_rpc("<get/>");
        let (id2, _) = build_rpc("<get/>");
        assert_eq!(id2, id1 + 1);
    }

    #[test]
    fn test_data_preserves_inner_xml_exactly() {
        let xml = r#"<rpc-reply message-id="5" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data>
    <!-- comment preserved -->
    <config xmlns="urn:example">
      <value attr="x &amp; y">text</value>
    </config>
  </data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        match &reply.body {
            RpcReplyBody::Data(data) => {
                // Span-based extraction preserves comments, entities, and attributes exactly
                assert!(data.contains("<!-- comment preserved -->"));
                assert!(data.contains("x &amp; y"));
                assert!(data.contains("<config"));
            }
            _ => panic!("expected Data reply"),
        }
    }

    // ── classify_message tests ──────────────────────────────────────────

    #[test]
    fn classify_rpc_reply() {
        let xml = r#"<rpc-reply message-id="1" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"><ok/></rpc-reply>"#;
        let msg = classify_message(Bytes::from(xml)).unwrap();
        assert!(matches!(msg, ServerMessage::RpcReply(_)));
    }

    #[test]
    fn classify_with_xml_declaration() {
        let xml = r#"<?xml version="1.0" encoding="UTF-8"?><rpc-reply message-id="5" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0"><ok/></rpc-reply>"#;
        let msg = classify_message(Bytes::from(xml)).unwrap();
        assert!(matches!(msg, ServerMessage::RpcReply(_)));
    }

    #[test]
    fn classify_unknown_root() {
        let xml = r#"<unknown-element/>"#;
        let result = classify_message(Bytes::from(xml));
        assert!(result.is_err());
    }

    #[test]
    fn classify_empty_message() {
        let result = classify_message(Bytes::from(""));
        assert!(result.is_err());
    }

    // ── DataPayload tests ──────────────────────────────────────────────

    #[test]
    fn data_payload_as_str() {
        let xml = r#"<rpc-reply message-id="20" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data>
    <config><hostname>router1</hostname></config>
  </data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        match &reply.body {
            RpcReplyBody::Data(payload) => {
                let s = payload.as_str();
                assert!(s.contains("<config>"));
                assert!(s.contains("router1"));
                // Deref<Target=str> works
                assert!(payload.contains("router1"));
            }
            _ => panic!("expected Data reply"),
        }
    }

    #[test]
    fn data_payload_into_string() {
        let xml = r#"<rpc-reply message-id="21" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data><value>hello</value></data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        match reply.body {
            RpcReplyBody::Data(payload) => {
                let s = payload.into_string();
                assert!(s.contains("<value>hello</value>"));
            }
            _ => panic!("expected Data reply"),
        }
    }

    #[test]
    fn data_payload_reader() {
        let xml = r#"<rpc-reply message-id="22" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data><item>one</item><item>two</item></data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        match &reply.body {
            RpcReplyBody::Data(payload) => {
                let mut reader = payload.reader();
                let mut buf = Vec::new();
                let mut items = Vec::new();
                loop {
                    match reader.read_event_into(&mut buf) {
                        Ok(Event::Start(e)) if e.local_name().as_ref() == b"item" => {}
                        Ok(Event::Text(e)) => {
                            items.push(e.xml_content().unwrap().to_string());
                        }
                        Ok(Event::Eof) => break,
                        _ => {}
                    }
                    buf.clear();
                }
                assert_eq!(items, vec!["one", "two"]);
            }
            _ => panic!("expected Data reply"),
        }
    }

    #[test]
    fn data_payload_empty() {
        let payload = DataPayload::empty();
        assert!(payload.is_empty());
        assert_eq!(payload.len(), 0);
        assert_eq!(payload.as_str(), "");
        assert_eq!(payload.into_string(), "");
    }

    #[test]
    fn data_payload_len_and_is_empty() {
        let xml = r#"<rpc-reply message-id="23" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data>abc</data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        match &reply.body {
            RpcReplyBody::Data(payload) => {
                assert_eq!(payload.len(), 3);
                assert!(!payload.is_empty());
            }
            _ => panic!("expected Data reply"),
        }
    }

    #[test]
    fn data_payload_large_preserves_content() {
        // Generate a large data payload to verify no truncation
        let inner = "<item>x</item>".repeat(1000);
        let xml = format!(
            r#"<rpc-reply message-id="24" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data>{inner}</data>
</rpc-reply>"#
        );
        let reply = parse_rpc_reply(&xml).unwrap();
        match &reply.body {
            RpcReplyBody::Data(payload) => {
                assert_eq!(payload.as_str(), inner);
                assert_eq!(payload.len(), inner.len());
            }
            _ => panic!("expected Data reply"),
        }
    }

    #[test]
    fn data_payload_partial_eq() {
        let xml = r#"<rpc-reply message-id="25" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data>hello</data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        match &reply.body {
            RpcReplyBody::Data(payload) => {
                assert!(*payload == *"hello");
                assert!(*payload != *"world");
            }
            _ => panic!("expected Data reply"),
        }
    }

    #[test]
    fn rpc_reply_into_data() {
        let xml = r#"<rpc-reply message-id="26" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <data><config/></data>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        let payload = reply.into_data().unwrap();
        assert!(payload.contains("<config/>"));
    }

    #[test]
    fn rpc_reply_into_data_ok() {
        let xml = r#"<rpc-reply message-id="27" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <ok/>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        let payload = reply.into_data().unwrap();
        assert!(payload.is_empty());
    }

    #[test]
    fn rpc_reply_into_data_error() {
        let xml = r#"<rpc-reply message-id="28" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  <rpc-error>
    <error-type>application</error-type>
    <error-tag>invalid-value</error-tag>
    <error-severity>error</error-severity>
    <error-message>bad</error-message>
  </rpc-error>
</rpc-reply>"#;
        let reply = parse_rpc_reply(xml).unwrap();
        assert!(reply.into_data().is_err());
    }
}