uopi/internal/datasource/modbus/modbus_test.go

package modbus

import (
	"context"
	"encoding/binary"
	"io"
	"net"
	"sync"
	"testing"
	"time"

	"github.com/uopi/uopi/internal/datasource"
)

// mockServer is an in-process Modbus TCP slave for tests. It serves a small
// register/coil store and records writes. Only the function codes exercised by
// the data source are implemented.
type mockServer struct {
	ln net.Listener

	mu        sync.Mutex
	holding   map[uint16]uint16
	input     map[uint16]uint16
	coils     map[uint16]bool
	discrete  map[uint16]bool
	lastWrite []uint16 // registers from the most recent write
}

func newMockServer(t *testing.T) *mockServer {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	s := &mockServer{
		ln:       ln,
		holding:  map[uint16]uint16{},
		input:    map[uint16]uint16{},
		coils:    map[uint16]bool{},
		discrete: map[uint16]bool{},
	}
	go s.serve()
	t.Cleanup(func() { ln.Close() })
	return s
}

func (s *mockServer) addr() string { return s.ln.Addr().String() }

func (s *mockServer) serve() {
	for {
		conn, err := s.ln.Accept()
		if err != nil {
			return
		}
		go s.handle(conn)
	}
}

func (s *mockServer) handle(conn net.Conn) {
	defer conn.Close()
	for {
		head := make([]byte, 7)
		if _, err := io.ReadFull(conn, head); err != nil {
			return
		}
		tx := binary.BigEndian.Uint16(head[0:])
		length := binary.BigEndian.Uint16(head[4:])
		body := make([]byte, length-1)
		if _, err := io.ReadFull(conn, body); err != nil {
			return
		}
		resp := s.respond(body)
		out := make([]byte, 7+len(resp))
		binary.BigEndian.PutUint16(out[0:], tx)
		binary.BigEndian.PutUint16(out[2:], 0)
		binary.BigEndian.PutUint16(out[4:], uint16(1+len(resp)))
		out[6] = head[6]
		copy(out[7:], resp)
		if _, err := conn.Write(out); err != nil {
			return
		}
	}
}

func (s *mockServer) respond(pdu []byte) []byte {
	s.mu.Lock()
	defer s.mu.Unlock()
	fc := pdu[0]
	switch fc {
	case fcReadHolding, fcReadInput:
		addr := binary.BigEndian.Uint16(pdu[1:])
		qty := binary.BigEndian.Uint16(pdu[3:])
		out := []byte{fc, byte(qty * 2)}
		src := s.holding
		if fc == fcReadInput {
			src = s.input
		}
		for i := uint16(0); i < qty; i++ {
			out = binary.BigEndian.AppendUint16(out, src[addr+i])
		}
		return out
	case fcReadCoils, fcReadDiscrete:
		addr := binary.BigEndian.Uint16(pdu[1:])
		qty := binary.BigEndian.Uint16(pdu[3:])
		nbytes := (int(qty) + 7) / 8
		out := []byte{fc, byte(nbytes)}
		bits := make([]byte, nbytes)
		src := s.coils
		if fc == fcReadDiscrete {
			src = s.discrete
		}
		for i := uint16(0); i < qty; i++ {
			if src[addr+i] {
				bits[i/8] |= 1 << (i % 8)
			}
		}
		return append(out, bits...)
	case fcWriteSingleReg:
		addr := binary.BigEndian.Uint16(pdu[1:])
		val := binary.BigEndian.Uint16(pdu[3:])
		s.holding[addr] = val
		s.lastWrite = []uint16{val}
		return pdu // echo
	case fcWriteSingleCoil:
		addr := binary.BigEndian.Uint16(pdu[1:])
		s.coils[addr] = binary.BigEndian.Uint16(pdu[3:]) == 0xFF00
		return pdu
	case fcWriteMultipleRegs:
		addr := binary.BigEndian.Uint16(pdu[1:])
		qty := binary.BigEndian.Uint16(pdu[3:])
		s.lastWrite = nil
		for i := uint16(0); i < qty; i++ {
			v := binary.BigEndian.Uint16(pdu[6+i*2:])
			s.holding[addr+i] = v
			s.lastWrite = append(s.lastWrite, v)
		}
		return append([]byte{fc}, pdu[1:5]...)
	default:
		return []byte{fc | 0x80, 0x01}
	}
}

func (s *mockServer) setHolding(addr, val uint16) {
	s.mu.Lock()
	s.holding[addr] = val
	s.mu.Unlock()
}

func (s *mockServer) setInput(addr, val uint16) {
	s.mu.Lock()
	s.input[addr] = val
	s.mu.Unlock()
}

func (s *mockServer) setDiscrete(addr uint16, on bool) {
	s.mu.Lock()
	s.discrete[addr] = on
	s.mu.Unlock()
}

func testConfig(addr string) Config {
	return Config{
		Enabled:        true,
		PollIntervalMs: 20,
		Devices: []Device{{
			Name:    "dev",
			Address: addr,
			UnitID:  1,
			Registers: []Register{
				{Name: "temp", Kind: "holding", Address: 10, Encoding: "int16", Scale: 0.1, Unit: "C", Writable: true},
				{Name: "count", Kind: "holding", Address: 20, Encoding: "uint16"},
				{Name: "big", Kind: "input", Address: 30, Encoding: "uint32"},
				{Name: "flag", Kind: "discrete", Address: 5},
				{Name: "relay", Kind: "coil", Address: 6, Writable: true},
				{Name: "sp", Kind: "holding", Address: 40, Encoding: "float32", Writable: true},
			},
		}},
	}
}

func TestReadRegisters(t *testing.T) {
	srv := newMockServer(t)
	srv.setHolding(10, 235) // int16, scale 0.1 → 23.5
	srv.setHolding(20, 7)
	srv.setInput(30, 0)
	srv.setInput(31, 1000) // uint32 big-word-first: low word at 31
	srv.setDiscrete(5, true)

	m, err := New(testConfig(srv.addr()))
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer m.Close()

	temp, err := m.readSignal(m.signals["dev:temp"])
	if err != nil {
		t.Fatalf("read temp: %v", err)
	}
	if f, ok := temp.Data.(float64); !ok || f < 23.49 || f > 23.51 {
		t.Errorf("temp = %v (%T), want 23.5", temp.Data, temp.Data)
	}

	count, err := m.readSignal(m.signals["dev:count"])
	if err != nil {
		t.Fatalf("read count: %v", err)
	}
	if v, ok := count.Data.(int64); !ok || v != 7 {
		t.Errorf("count = %v (%T), want int64 7", count.Data, count.Data)
	}

	big, err := m.readSignal(m.signals["dev:big"])
	if err != nil {
		t.Fatalf("read big: %v", err)
	}
	if v, ok := big.Data.(int64); !ok || v != 1000 {
		t.Errorf("big = %v (%T), want int64 1000", big.Data, big.Data)
	}

	flag, err := m.readSignal(m.signals["dev:flag"])
	if err != nil {
		t.Fatalf("read flag: %v", err)
	}
	if b, ok := flag.Data.(bool); !ok || !b {
		t.Errorf("flag = %v, want true", flag.Data)
	}
}

func TestWriteRoundTrip(t *testing.T) {
	srv := newMockServer(t)
	m, err := New(testConfig(srv.addr()))
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	defer m.Close()
	ctx := context.Background()

	// Scaled int16: writing 23.5 with scale 0.1 should store raw 235.
	if err := m.Write(ctx, "dev:temp", 23.5); err != nil {
		t.Fatalf("write temp: %v", err)
	}
	srv.mu.Lock()
	raw := srv.holding[10]
	srv.mu.Unlock()
	if raw != 235 {
		t.Errorf("holding[10] = %d, want 235", raw)
	}

	// Coil write.
	if err := m.Write(ctx, "dev:relay", true); err != nil {
		t.Fatalf("write relay: %v", err)
	}
	srv.mu.Lock()
	on := srv.coils[6]
	srv.mu.Unlock()
	if !on {
		t.Error("coil 6 not set")
	}

	// float32 multi-register write.
	if err := m.Write(ctx, "dev:sp", 12.5); err != nil {
		t.Fatalf("write sp: %v", err)
	}
	got, err := m.readSignal(m.signals["dev:sp"])
	if err != nil {
		t.Fatalf("read sp: %v", err)
	}
	if f, ok := got.Data.(float64); !ok || f < 12.49 || f > 12.51 {
		t.Errorf("sp = %v, want 12.5", got.Data)
	}
}

func TestWriteErrors(t *testing.T) {
	srv := newMockServer(t)
	m, _ := New(testConfig(srv.addr()))
	defer m.Close()
	ctx := context.Background()

	if err := m.Write(ctx, "dev:missing", 1); err != datasource.ErrNotFound {
		t.Errorf("missing write err = %v, want ErrNotFound", err)
	}
	// Input register is read-only.
	if err := m.Write(ctx, "dev:big", 1); err != datasource.ErrNotWritable {
		t.Errorf("input write err = %v, want ErrNotWritable", err)
	}
}

func TestSubscribe(t *testing.T) {
	srv := newMockServer(t)
	srv.setHolding(20, 42)
	m, _ := New(testConfig(srv.addr()))
	defer m.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	ch := make(chan datasource.Value, 4)
	stop, err := m.Subscribe(ctx, "dev:count", ch)
	if err != nil {
		t.Fatalf("subscribe: %v", err)
	}
	defer stop()

	select {
	case v := <-ch:
		if v.Quality != datasource.QualityGood {
			t.Errorf("quality = %v, want good", v.Quality)
		}
		if iv, ok := v.Data.(int64); !ok || iv != 42 {
			t.Errorf("first value = %v, want 42", v.Data)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for first value")
	}
}

func TestSubscribeBadQualityOnError(t *testing.T) {
	// Point at a closed port so reads fail; expect QualityBad, not a hang.
	cfg := testConfig("127.0.0.1:1") // port 1: connection refused
	m, _ := New(cfg)
	defer m.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	ch := make(chan datasource.Value, 1)
	stop, err := m.Subscribe(ctx, "dev:count", ch)
	if err != nil {
		t.Fatalf("subscribe: %v", err)
	}
	defer stop()

	select {
	case v := <-ch:
		if v.Quality != datasource.QualityBad {
			t.Errorf("quality = %v, want bad", v.Quality)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out")
	}
}

func TestNewValidation(t *testing.T) {
	if _, err := New(Config{Devices: []Device{{Name: "", Address: "x"}}}); err == nil {
		t.Error("expected error for missing device name")
	}
	if _, err := New(Config{Devices: []Device{{Name: "a", Address: "x"}, {Name: "a", Address: "y"}}}); err == nil {
		t.Error("expected error for duplicate device")
	}
	if _, err := New(Config{Devices: []Device{{Name: "a", Address: "x", Registers: []Register{{Name: "r", Encoding: "bogus"}}}}}); err == nil {
		t.Error("expected error for bad encoding")
	}
	if _, err := New(Config{Devices: []Device{{Name: "a", Address: "x", Registers: []Register{{Name: "r"}, {Name: "r"}}}}}); err == nil {
		t.Error("expected error for duplicate register")
	}
}

func TestSubscribeUnknownSignal(t *testing.T) {
	m, _ := New(testConfig("127.0.0.1:502"))
	defer m.Close()
	if _, err := m.Subscribe(context.Background(), "nope", nil); err != datasource.ErrNotFound {
		t.Errorf("err = %v, want ErrNotFound", err)
	}
	if _, err := m.GetMetadata(context.Background(), "nope"); err != datasource.ErrNotFound {
		t.Errorf("meta err = %v, want ErrNotFound", err)
	}
}