(mynewt-newtmgr) 18/20: nmxact/udp: Add unit tests for UDP session

[ja...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sjanc pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mynewt-newtmgr.git

commit a040631682424c9e5c0bba34d38ee86f860f5691
Author: Szymon Janc <sz...@codecoup.pl>
AuthorDate: Thu Oct 22 00:37:29 2026 +0200

    nmxact/udp: Add unit tests for UDP session
    
    Drive UdpSesn against a fake device listening on an ephemeral
    127.0.0.1 UDP port that answers NMP or OMP echo requests: Open,
    TxRxMgmt, Close, reopening a closed session, Open/Close twice,
    TxRxMgmtAsync (regression test for 8473977), response timeouts,
    oversized requests, an invalid peer address and UdpXport start/stop.
    
    Assisted-by: Claude:claude-opus-5-5
---
 nmxact/udp/udp_test.go | 446 +++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 446 insertions(+)

diff --git a/nmxact/udp/udp_test.go b/nmxact/udp/udp_test.go
new file mode 100644
index 0000000..4b1b12f
--- /dev/null
+++ b/nmxact/udp/udp_test.go
@@ -0,0 +1,446 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package udp
+
+import (
+	"fmt"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/runtimeco/go-coap"
+	"github.com/ugorji/go/codec"
+
+	"mynewt.apache.org/newtmgr/nmxact/nmp"
+	"mynewt.apache.org/newtmgr/nmxact/nmxutil"
+	"mynewt.apache.org/newtmgr/nmxact/sesn"
+)
+
+const testTimeout = 5 * time.Second
+
+// fakeUdpDevice is a management server on a loopback UDP socket that
+// answers NMP or OMP echo requests.
+type fakeUdpDevice struct {
+	conn    *net.UDPConn
+	proto   sesn.MgmtProto
+	respond bool
+
+	mtx  sync.Mutex
+	errs []error
+	reqs int
+
+	wg sync.WaitGroup
+}
+
+func newFakeUdpDevice(t *testing.T, proto sesn.MgmtProto,
+	respond bool) *fakeUdpDevice {
+
+	t.Helper()
+
+	conn, err := net.ListenUDP("udp",
+		&net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0})
+	if err != nil {
+		t.Fatalf("ListenUDP failed: %v", err)
+	}
+
+	f := &fakeUdpDevice{
+		conn:    conn,
+		proto:   proto,
+		respond: respond,
+	}
+
+	f.wg.Add(1)
+	go f.serve()
+
+	t.Cleanup(func() {
+		conn.Close()
+		f.wg.Wait()
+
+		f.mtx.Lock()
+		defer f.mtx.Unlock()
+		for _, err := range f.errs {
+			t.Errorf("fake device: %v", err)
+		}
+	})
+
+	return f
+}
+
+func (f *fakeUdpDevice) addr() string {
+	return f.conn.LocalAddr().String()
+}
+
+func (f *fakeUdpDevice) numReqs() int {
+	f.mtx.Lock()
+	defer f.mtx.Unlock()
+
+	return f.reqs
+}
+
+func (f *fakeUdpDevice) serve() {
+	defer f.wg.Done()
+
+	buf := make([]byte, 4096)
+	for {
+		n, src, err := f.conn.ReadFromUDP(buf)
+		if err != nil {
+			return
+		}
+
+		f.mtx.Lock()
+		f.reqs++
+		f.mtx.Unlock()
+
+		rsp, err := f.process(buf[:n])
+		if err != nil {
+			f.mtx.Lock()
+			f.errs = append(f.errs, err)
+			f.mtx.Unlock()
+			continue
+		}
+
+		if f.respond {
+			f.conn.WriteToUDP(rsp, src)
+		}
+	}
+}
+
+func cborDecodeMap(b []byte) (map[string]interface{}, error) {
+	m := map[string]interface{}{}
+	dec := codec.NewDecoderBytes(b, new(codec.CborHandle))
+	if err := dec.Decode(&m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func echoRspHdr(seq uint8) nmp.NmpHdr {
+	return nmp.NmpHdr{
+		Op:    nmp.NMP_OP_WRITE_RSP,
+		Group: nmp.NMP_GROUP_DEFAULT,
+		Id:    nmp.NMP_ID_DEF_ECHO,
+		Seq:   seq,
+	}
+}
+
+func (f *fakeUdpDevice) process(req []byte) ([]byte, error) {
+	if f.proto == sesn.MGMT_PROTO_NMP {
+		hdr, err := nmp.DecodeNmpHdr(req)
+		if err != nil {
+			return nil, err
+		}
+		body, err := cborDecodeMap(req[nmp.NMP_HDR_SIZE:])
+		if err != nil {
+			return nil, err
+		}
+		return nmp.EncodeNmpPlain(&nmp.NmpMsg{
+			Hdr:  echoRspHdr(hdr.Seq),
+			Body: map[string]interface{}{"r": body["d"], "rc": 0},
+		})
+	}
+
+	m, err := coap.ParseDgramMessage(req)
+	if err != nil {
+		return nil, err
+	}
+	body, err := cborDecodeMap(m.Payload())
+	if err != nil {
+		return nil, err
+	}
+	hb, ok := body["_h"].([]byte)
+	if !ok {
+		return nil, fmt.Errorf("OMP request without NMP header")
+	}
+	hdr, err := nmp.DecodeNmpHdr(hb)
+	if err != nil {
+		return nil, err
+	}
+
+	rh := echoRspHdr(hdr.Seq)
+	var payload []byte
+	enc := codec.NewEncoderBytes(&payload, new(codec.CborHandle))
+	err = enc.Encode(map[string]interface{}{
+		"_h": rh.Bytes(),
+		"r":  body["d"],
+		"rc": 0,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	rsp := coap.NewDgramMessage(coap.MessageParams{
+		Type:      coap.Acknowledgement,
+		Code:      coap.Content,
+		MessageID: m.MessageID(),
+		Token:     m.Token(),
+		Payload:   payload,
+	})
+	return rsp.MarshalBinary()
+}
+
+func newTestSesn(t *testing.T, proto sesn.MgmtProto,
+	addr string) *UdpSesn {
+
+	t.Helper()
+
+	cfg := sesn.NewSesnCfg()
+	cfg.MgmtProto = proto
+	cfg.PeerSpec.Udp = addr
+
+	s, err := NewUdpSesn(cfg)
+	if err != nil {
+		t.Fatalf("NewUdpSesn failed: %v", err)
+	}
+	return s
+}
+
+func echoMsg(payload string) *nmp.NmpMsg {
+	r := nmp.NewEchoReq()
+	r.Payload = payload
+	return r.Msg()
+}
+
+func checkEcho(t *testing.T, rsp nmp.NmpRsp, seq uint8, payload string) {
+	t.Helper()
+
+	echo, ok := rsp.(*nmp.EchoRsp)
+	if !ok {
+		t.Fatalf("got %T, want *nmp.EchoRsp", rsp)
+	}
+	if echo.Payload != payload || echo.Hdr().Seq != seq {
+		t.Errorf("got payload %q seq %d, want %q seq %d",
+			echo.Payload, echo.Hdr().Seq, payload, seq)
+	}
+}
+
+var udpProtos = []sesn.MgmtProto{sesn.MGMT_PROTO_NMP, sesn.MGMT_PROTO_OMP}
+
+func TestUdpSesnTxRxMgmt(t *testing.T) {
+	for _, proto := range udpProtos {
+		t.Run(proto.String(), func(t *testing.T) {
+			dev := newFakeUdpDevice(t, proto, true)
+			s := newTestSesn(t, proto, dev.addr())
+
+			if s.IsOpen() {
+				t.Fatalf("new session is open")
+			}
+			if s.MgmtProto() != proto {
+				t.Errorf("MgmtProto() = %v, want %v", s.MgmtProto(), proto)
+			}
+			if err := s.Open(); err != nil {
+				t.Fatalf("Open failed: %v", err)
+			}
+			if !s.IsOpen() {
+				t.Fatalf("session not open after Open")
+			}
+
+			for i := 0; i < 3; i++ {
+				req := echoMsg(fmt.Sprintf("ping %d", i))
+				rsp, err := s.TxRxMgmt(req, testTimeout)
+				if err != nil {
+					t.Fatalf("TxRxMgmt failed: %v", err)
+				}
+				checkEcho(t, rsp, req.Hdr.Seq, fmt.Sprintf("ping %d", i))
+			}
+
+			if err := s.Close(); err != nil {
+				t.Fatalf("Close failed: %v", err)
+			}
+			if s.IsOpen() {
+				t.Errorf("session open after Close")
+			}
+		})
+	}
+}
+
+func TestUdpSesnOpenCloseTwice(t *testing.T) {
+	dev := newFakeUdpDevice(t, sesn.MGMT_PROTO_NMP, true)
+	s := newTestSesn(t, sesn.MGMT_PROTO_NMP, dev.addr())
+
+	if err := s.Close(); !nmxutil.IsSesnClosed(err) {
+		t.Errorf("Close of unopened session: got %v", err)
+	}
+	if err := s.Open(); err != nil {
+		t.Fatalf("Open failed: %v", err)
+	}
+	if err := s.Open(); !nmxutil.IsSesnAlreadyOpen(err) {
+		t.Errorf("second Open: got %v", err)
+	}
+	if err := s.Close(); err != nil {
+		t.Fatalf("Close failed: %v", err)
+	}
+	if err := s.Close(); !nmxutil.IsSesnClosed(err) {
+		t.Errorf("second Close: got %v", err)
+	}
+
+	if _, err := s.TxRxMgmt(echoMsg("x"), testTimeout); err == nil {
+		t.Errorf("TxRxMgmt on closed session succeeded")
+	}
+}
+
+func TestUdpSesnReopen(t *testing.T) {
+	for _, proto := range udpProtos {
+		t.Run(proto.String(), func(t *testing.T) {
+			dev := newFakeUdpDevice(t, proto, true)
+			s := newTestSesn(t, proto, dev.addr())
+
+			for i := 0; i < 2; i++ {
+				if err := s.Open(); err != nil {
+					t.Fatalf("Open %d failed: %v", i, err)
+				}
+				req := echoMsg("again")
+				rsp, err := s.TxRxMgmt(req, testTimeout)
+				if err != nil {
+					t.Fatalf("TxRxMgmt %d failed: %v", i, err)
+				}
+				checkEcho(t, rsp, req.Hdr.Seq, "again")
+				if err := s.Close(); err != nil {
+					t.Fatalf("Close %d failed: %v", i, err)
+				}
+			}
+		})
+	}
+}
+
+// Regression test for 8473977: TxRxMgmtAsync must deliver the response
+// on ch (UDP sessions implement it synchronously).
+func TestUdpSesnTxRxMgmtAsync(t *testing.T) {
+	for _, proto := range udpProtos {
+		t.Run(proto.String(), func(t *testing.T) {
+			dev := newFakeUdpDevice(t, proto, true)
+			s := newTestSesn(t, proto, dev.addr())
+			if err := s.Open(); err != nil {
+				t.Fatal(err)
+			}
+			defer s.Close()
+
+			ch := make(chan nmp.NmpRsp, 1)
+			errc := make(chan error, 1)
+			req := echoMsg("async")
+
+			if err := s.TxRxMgmtAsync(req, testTimeout, ch, errc); err != nil {
+				t.Fatalf("TxRxMgmtAsync failed: %v", err)
+			}
+
+			select {
+			case rsp := <-ch:
+				checkEcho(t, rsp, req.Hdr.Seq, "async")
+			case err := <-errc:
+				t.Fatalf("unexpected error: %v", err)
+			case <-time.After(testTimeout):
+				t.Fatalf("timeout waiting for async response")
+			}
+		})
+	}
+}
+
+func TestUdpSesnTimeout(t *testing.T) {
+	for _, proto := range udpProtos {
+		t.Run(proto.String(), func(t *testing.T) {
+			dev := newFakeUdpDevice(t, proto, false)
+			s := newTestSesn(t, proto, dev.addr())
+			if err := s.Open(); err != nil {
+				t.Fatal(err)
+			}
+			defer s.Close()
+
+			_, err := s.TxRxMgmt(echoMsg("x"), 50*time.Millisecond)
+			if !nmxutil.IsRspTimeout(err) {
+				t.Fatalf("got %v (%T), want RspTimeoutError", err, err)
+			}
+
+			// The async variant reports the timeout on errc.
+			ch := make(chan nmp.NmpRsp, 1)
+			errc := make(chan error, 1)
+			s.TxRxMgmtAsync(echoMsg("y"), 50*time.Millisecond, ch, errc)
+			select {
+			case err := <-errc:
+				if !nmxutil.IsRspTimeout(err) {
+					t.Errorf("got %v, want RspTimeoutError", err)
+				}
+			case rsp := <-ch:
+				t.Fatalf("unexpected response %+v", rsp)
+			case <-time.After(testTimeout):
+				t.Fatalf("timeout waiting for async error")
+			}
+
+			if n := dev.numReqs(); n != 2 {
+				t.Errorf("device saw %d requests, want 2", n)
+			}
+		})
+	}
+}
+
+func TestUdpSesnRequestTooBig(t *testing.T) {
+	dev := newFakeUdpDevice(t, sesn.MGMT_PROTO_NMP, true)
+	s := newTestSesn(t, sesn.MGMT_PROTO_NMP, dev.addr())
+	if err := s.Open(); err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	req := echoMsg(strings.Repeat("x", MAX_PACKET_SIZE))
+	if _, err := s.TxRxMgmt(req, testTimeout); err == nil {
+		t.Errorf("oversized request succeeded")
+	}
+}
+
+func TestUdpSesnBadPeer(t *testing.T) {
+	s := newTestSesn(t, sesn.MGMT_PROTO_NMP, "not an address")
+	if err := s.Open(); err == nil {
+		s.Close()
+		t.Fatalf("Open with invalid peer succeeded")
+	}
+	if s.IsOpen() {
+		t.Errorf("session open after failed Open")
+	}
+}
+
+func TestUdpXportStartStop(t *testing.T) {
+	x := NewUdpXport()
+
+	if err := x.Stop(); err == nil {
+		t.Errorf("Stop of unstarted xport succeeded")
+	}
+	if err := x.Start(); err != nil {
+		t.Fatalf("Start failed: %v", err)
+	}
+	if err := x.Start(); err == nil {
+		t.Errorf("second Start succeeded")
+	}
+	if err := x.Stop(); err != nil {
+		t.Fatalf("Stop failed: %v", err)
+	}
+	if err := x.Tx([]byte{1}); err == nil {
+		t.Errorf("Tx succeeded")
+	}
+
+	cfg := sesn.NewSesnCfg()
+	cfg.PeerSpec.Udp = "127.0.0.1:1"
+	s, err := x.BuildSesn(cfg)
+	if err != nil {
+		t.Fatalf("BuildSesn failed: %v", err)
+	}
+	if _, ok := s.(*UdpSesn); !ok {
+		t.Errorf("BuildSesn returned %T, want *UdpSesn", s)
+	}
+}