converge/pkg/comms/agentserver_test.go

package comms

import (
	"context"
	"git.wamblee.org/converge/pkg/testsupport"
	"github.com/stretchr/testify/suite"
	"io"
	"log"
	"net/http"
	"testing"
	"time"
)

type AgentServerTestSuite struct {
	suite.Suite

	ctx         context.Context
	cancelFunc  context.CancelFunc
	pprofServer *http.Server

	agentConnection  io.ReadWriteCloser
	serverConnection io.ReadWriteCloser
}

func (s *AgentServerTestSuite) SetupSuite() {
	s.pprofServer = testsupport.StartPprof("")
}

func (s *AgentServerTestSuite) TearDownSuite() {
	testsupport.StopPprof(s.ctx, s.pprofServer)
}

func (s *AgentServerTestSuite) SetupTest() {
	ctx, cancelFunc := testsupport.CreateTestContext(context.Background(), 10*time.Second)
	s.ctx = ctx
	s.cancelFunc = cancelFunc

	// Could have also used net.Pipe but net.Pipe uses synchronous communication
	// by default and the bitpipe implementation can become asynchronous when
	// a channels ize > 0 is passed in. Also the test utility respects the context
	// so also deals with cancellation much better than net.Pipe.
	bitpipe := testsupport.NewInmemoryConnection(s.ctx, "inmemory", 10)
	agentConnection := bitpipe.Front()
	serverConnection := bitpipe.Back()
	s.serverConnection = serverConnection
	s.agentConnection = agentConnection
}

func (suite *AgentServerTestSuite) TearDownTest() {
}

func TestAgentServerTestSuite(t *testing.T) {
	suite.Run(t, &AgentServerTestSuite{})
}

func (s *AgentServerTestSuite) createCommChannel() (CommChannel, CommChannel) {
	commChannels := testsupport.RunAndWait(
		&s.Suite,
		func() any {
			log.Println("Agent initializing")
			commChannel, err := NewCommChannel(Agent, s.agentConnection)
			s.Nil(err)
			return commChannel
		},
		func() any {
			log.Println("Server initializing")
			commChannel, err := NewCommChannel(ConvergeServer, s.serverConnection)
			s.Nil(err)
			return commChannel
		},
	)

	s.Equal(2, len(commChannels))
	agentCommChannel := commChannels[0].(CommChannel)
	serverCommChannel := commChannels[1].(CommChannel)
	return agentCommChannel, serverCommChannel
}

func (s *AgentServerTestSuite) TestNewCommChannel() {
	// Setup Comm channel
	agentCommChannel, serverCommChannel := s.createCommChannel()

	// verify the side channel is working by sending an object
	testsupport.RunAndWait(
		&s.Suite,
		func() any {
			protocolVersion := ProtocolVersion{Version: 10}
			err := SendWithTimeout[ProtocolVersion](agentCommChannel.SideChannel, protocolVersion)
			s.Nil(err)
			log.Printf("Sent one message %v", protocolVersion)
			return nil
		},
		func() any {
			protocolVersion, err := ReceiveWithTimeout[ProtocolVersion](serverCommChannel.SideChannel)
			s.Nil(err)
			log.Printf("Received one message %v", protocolVersion)
			return nil
		},
	)

	log.Printf("%v %v", agentCommChannel, serverCommChannel)
}

func (s *AgentServerTestSuite) Test_ConnectThroughYamux() {
	agentCommChannel, serverCommChannel := s.createCommChannel()

	dataAgentToServer := "hello"
	dataServerToAgent := "bye"
	testsupport.RunAndWait(
		&s.Suite,
		func() any {
			conn, err := agentCommChannel.Session.OpenStream()
			s.Nil(err)
			n, err := conn.Write([]byte(dataAgentToServer))
			s.Nil(err)
			s.Equal(len(dataAgentToServer), n)

			buf := make([]byte, len(dataServerToAgent))
			n, err = conn.Read(buf)
			s.Nil(err)
			s.Equal(len(dataServerToAgent), n)
			s.Equal([]byte(dataServerToAgent), buf[:n])

			return nil
		},
		func() any {
			conn, err := serverCommChannel.Session.Accept()

			s.Nil(err)
			buf := make([]byte, len(dataAgentToServer))
			n, err := conn.Read(buf)
			s.Nil(err)
			s.Equal(len(dataAgentToServer), n)
			s.Equal([]byte(dataAgentToServer), buf[:n])

			n, err = conn.Write([]byte(dataServerToAgent))
			s.Nil(err)
			s.Equal(len(dataServerToAgent), n)

			return nil
		})
}

func (s *AgentServerTestSuite) Test_ListenForAgentEvents() {

}

func (s *AgentServerTestSuite) Test_LIstenForServerEvents() {

}