rdp-proxy/agents/rap-node-agent/internal/mesh/fabric_session_manager_test.go

package mesh

import (
	"context"
	"net/http/httptest"
	"testing"
	"time"

	"github.com/example/remote-access-platform/agents/rap-node-agent/internal/fabricproto"
)

func TestFabricSessionPeerManagerReusesPeerPump(t *testing.T) {
	var opened int
	server := httptest.NewServer(Server{
		Local:                PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
		FabricSessionEnabled: true,
		FabricSessionLogger: func(entry FabricSessionEventLogEntry) {
			if entry.Event == "fabric_session_websocket_opened" {
				opened++
			}
		},
	}.Handler())
	defer server.Close()

	manager := NewFabricSessionPeerManager()
	defer manager.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	target := FabricSessionPeerTarget{
		PeerID:  "node-a",
		BaseURL: server.URL,
		Options: FabricSessionDialOptions{
			Token:   "rap_fsn_manager",
			Timeout: time.Second,
		},
		Pump: FabricSessionPumpOptions{
			OutboundBuffer: 4,
			InboundBuffer:  4,
		},
	}

	first, err := manager.Get(ctx, target)
	if err != nil {
		t.Fatalf("first get: %v", err)
	}
	second, err := manager.Get(ctx, target)
	if err != nil {
		t.Fatalf("second get: %v", err)
	}
	if first != second {
		t.Fatal("manager did not reuse peer pump")
	}
	if opened != 1 {
		t.Fatalf("opened sessions = %d, want 1", opened)
	}
	snapshot := manager.Snapshot()
	if snapshot.SchemaVersion != "rap.fabric_session_peer_manager.v1" ||
		snapshot.ActiveCount != 1 ||
		snapshot.ClosedCount != 0 ||
		snapshot.Stats.Opens != 1 ||
		snapshot.Stats.Reuses != 1 {
		t.Fatalf("snapshot = %+v", snapshot)
	}
	if err := first.Send(ctx, fabricproto.Frame{
		Type:     fabricproto.FramePing,
		Sequence: 1,
		Payload:  []byte("manager"),
	}); err != nil {
		t.Fatalf("send ping: %v", err)
	}
	select {
	case frame := <-first.Frames():
		if frame.Type != fabricproto.FramePong || frame.Sequence != 1 || string(frame.Payload) != "manager" {
			t.Fatalf("frame = %+v", frame)
		}
	case err := <-first.Errors():
		t.Fatalf("pump error: %v", err)
	case <-ctx.Done():
		t.Fatal(ctx.Err())
	}
}

func TestFabricSessionPeerManagerClosePeerReopens(t *testing.T) {
	var opened int
	server := httptest.NewServer(Server{
		Local:                PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
		FabricSessionEnabled: true,
		FabricSessionLogger: func(entry FabricSessionEventLogEntry) {
			if entry.Event == "fabric_session_websocket_opened" {
				opened++
			}
		},
	}.Handler())
	defer server.Close()

	manager := NewFabricSessionPeerManager()
	defer manager.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	target := FabricSessionPeerTarget{
		PeerID:  "node-a",
		BaseURL: server.URL,
		Options: FabricSessionDialOptions{
			Token:   "rap_fsn_manager_reopen",
			Timeout: time.Second,
		},
	}

	first, err := manager.Get(ctx, target)
	if err != nil {
		t.Fatalf("first get: %v", err)
	}
	if err := manager.ClosePeer(target); err != nil {
		t.Fatalf("close peer: %v", err)
	}
	second, err := manager.Get(ctx, target)
	if err != nil {
		t.Fatalf("second get: %v", err)
	}
	if first == second {
		t.Fatal("manager reused pump after close peer")
	}
	if opened != 2 {
		t.Fatalf("opened sessions = %d, want 2", opened)
	}
	if snapshot := manager.Snapshot(); snapshot.Stats.ClosePeerCalls != 1 || snapshot.Stats.Opens != 2 {
		t.Fatalf("snapshot = %+v", snapshot)
	}
}

func TestFabricSessionPeerManagerReopensClosedPump(t *testing.T) {
	var opened int
	server := httptest.NewServer(Server{
		Local:                PeerIdentity{ClusterID: "cluster-1", NodeID: "node-a"},
		FabricSessionEnabled: true,
		FabricSessionLogger: func(entry FabricSessionEventLogEntry) {
			if entry.Event == "fabric_session_websocket_opened" {
				opened++
			}
		},
	}.Handler())
	defer server.Close()

	manager := NewFabricSessionPeerManager()
	defer manager.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	target := FabricSessionPeerTarget{
		PeerID:  "node-a",
		BaseURL: server.URL,
		Options: FabricSessionDialOptions{
			Token:   "rap_fsn_manager_closed",
			Timeout: time.Second,
		},
	}

	first, err := manager.Get(ctx, target)
	if err != nil {
		t.Fatalf("first get: %v", err)
	}
	if err := first.Close(); err != nil {
		t.Fatalf("close first pump: %v", err)
	}
	if !first.Closed() {
		t.Fatal("first pump should report closed")
	}
	second, err := manager.Get(ctx, target)
	if err != nil {
		t.Fatalf("second get: %v", err)
	}
	if first == second {
		t.Fatal("manager reused closed pump")
	}
	if opened != 2 {
		t.Fatalf("opened sessions = %d, want 2", opened)
	}
	snapshot := manager.Snapshot()
	if snapshot.ActiveCount != 1 ||
		snapshot.Stats.Opens != 2 ||
		snapshot.Stats.ClosedEvicted != 1 {
		t.Fatalf("snapshot = %+v", snapshot)
	}
}

func TestFabricSessionPeerManagerRejectsIncompleteTarget(t *testing.T) {
	manager := NewFabricSessionPeerManager()
	_, err := manager.Get(context.Background(), FabricSessionPeerTarget{PeerID: "node-a"})
	if err == nil {
		t.Fatal("incomplete target unexpectedly succeeded")
	}
}