mxaccessgw/clients/go/mxgateway/alarms_test.go

package mxgateway

import (
	"context"
	"errors"
	"io"
	"net"
	"testing"

	pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
)

// PR E.4 — pins the Go SDK surface for the new alarm RPCs:
// AcknowledgeAlarm + QueryActiveAlarms.

func TestAcknowledgeAlarmSendsRequestAndReturnsReply(t *testing.T) {
	fake := &fakeGatewayWithAlarms{
		acknowledgeReply: &pb.AcknowledgeAlarmReply{
			CorrelationId: "corr-1",
			ProtocolStatus: &pb.ProtocolStatus{
				Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
			},
			Status: &pb.MxStatusProxy{
				Success:  1,
				Category: pb.MxStatusCategory_MX_STATUS_CATEGORY_OK,
			},
		},
	}
	client, cleanup := newBufconnClientWithAlarms(t, fake)
	defer cleanup()

	reply, err := client.AcknowledgeAlarm(context.Background(), &pb.AcknowledgeAlarmRequest{
		ClientCorrelationId: "corr-1",
		AlarmFullReference:  "Tank01.Level.HiHi",
		Comment:             "investigating",
		OperatorUser:        "alice",
	})
	if err != nil {
		t.Fatalf("AcknowledgeAlarm() error = %v", err)
	}
	if reply.GetProtocolStatus().GetCode() != pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK {
		t.Fatalf("protocol status = %v", reply.GetProtocolStatus().GetCode())
	}
	if got := fake.acknowledgeRequest.GetAlarmFullReference(); got != "Tank01.Level.HiHi" {
		t.Fatalf("captured alarm reference = %q", got)
	}
	if got := fake.acknowledgeRequest.GetComment(); got != "investigating" {
		t.Fatalf("captured comment = %q", got)
	}
	if got := fake.acknowledgeAuth; got != "Bearer test-api-key" {
		t.Fatalf("authorization metadata = %q", got)
	}
}

func TestAcknowledgeAlarmRejectsNilRequest(t *testing.T) {
	fake := &fakeGatewayWithAlarms{}
	client, cleanup := newBufconnClientWithAlarms(t, fake)
	defer cleanup()

	_, err := client.AcknowledgeAlarm(context.Background(), nil)
	if err == nil || !errors.Is(err, errors.Unwrap(err)) && err.Error() != "mxgateway: acknowledge alarm request is required" {
		// Accept either: the helper returned the literal sentinel, or the
		// generic transport error — both prove nil was rejected.
	}
	if err == nil {
		t.Fatalf("AcknowledgeAlarm(nil) returned no error")
	}
}

func TestAcknowledgeAlarmMapsUnauthenticated(t *testing.T) {
	fake := &fakeGatewayWithAlarms{
		acknowledgeError: status.Error(codes.Unauthenticated, "expired key"),
	}
	client, cleanup := newBufconnClientWithAlarms(t, fake)
	defer cleanup()

	_, err := client.AcknowledgeAlarm(context.Background(), &pb.AcknowledgeAlarmRequest{
		AlarmFullReference: "Tank01.Level.HiHi",
		OperatorUser:       "alice",
	})
	if err == nil {
		t.Fatalf("AcknowledgeAlarm() returned no error on Unauthenticated")
	}
	var gwErr *GatewayError
	if !errors.As(err, &gwErr) {
		t.Fatalf("error %T does not unwrap to *GatewayError", err)
	}
	if got, _ := status.FromError(gwErr.Err); got.Code() != codes.Unauthenticated {
		t.Fatalf("inner status code = %v", got.Code())
	}
}

func TestQueryActiveAlarmsStreamsSnapshots(t *testing.T) {
	fake := &fakeGatewayWithAlarms{
		activeSnapshots: []*pb.ActiveAlarmSnapshot{
			{
				AlarmFullReference: "Tank01.Level.HiHi",
				CurrentState:       pb.AlarmConditionState_ALARM_CONDITION_STATE_ACTIVE,
				Severity:           750,
			},
			{
				AlarmFullReference: "Tank02.Level.HiHi",
				CurrentState:       pb.AlarmConditionState_ALARM_CONDITION_STATE_ACTIVE_ACKED,
				Severity:           750,
			},
		},
	}
	client, cleanup := newBufconnClientWithAlarms(t, fake)
	defer cleanup()

	stream, err := client.QueryActiveAlarms(context.Background(), &pb.QueryActiveAlarmsRequest{
		SessionId: "session-1",
	})
	if err != nil {
		t.Fatalf("QueryActiveAlarms() error = %v", err)
	}

	var received []*pb.ActiveAlarmSnapshot
	for {
		snap, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			t.Fatalf("stream.Recv() error = %v", err)
		}
		received = append(received, snap)
	}
	if len(received) != 2 {
		t.Fatalf("snapshot count = %d, want 2", len(received))
	}
	if received[0].GetAlarmFullReference() != "Tank01.Level.HiHi" {
		t.Fatalf("snapshot[0] ref = %q", received[0].GetAlarmFullReference())
	}
	if received[1].GetCurrentState() != pb.AlarmConditionState_ALARM_CONDITION_STATE_ACTIVE_ACKED {
		t.Fatalf("snapshot[1] state = %v", received[1].GetCurrentState())
	}
}

func TestQueryActiveAlarmsPassesFilterPrefix(t *testing.T) {
	fake := &fakeGatewayWithAlarms{}
	client, cleanup := newBufconnClientWithAlarms(t, fake)
	defer cleanup()

	stream, err := client.QueryActiveAlarms(context.Background(), &pb.QueryActiveAlarmsRequest{
		SessionId:         "session-1",
		AlarmFilterPrefix: "Tank01.",
	})
	if err != nil {
		t.Fatalf("QueryActiveAlarms() error = %v", err)
	}
	for {
		_, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			t.Fatalf("stream.Recv() error = %v", err)
		}
	}

	if got := fake.queryRequest.GetAlarmFilterPrefix(); got != "Tank01." {
		t.Fatalf("captured filter prefix = %q", got)
	}
}

func TestStreamAlarmsPassesFilterPrefixAndReceivesFeedMessages(t *testing.T) {
	fake := &fakeGatewayWithAlarms{
		feedMessages: []*pb.AlarmFeedMessage{
			{
				Payload: &pb.AlarmFeedMessage_ActiveAlarm{
					ActiveAlarm: &pb.ActiveAlarmSnapshot{
						AlarmFullReference: "Tank01.Level.HiHi",
						CurrentState:       pb.AlarmConditionState_ALARM_CONDITION_STATE_ACTIVE,
					},
				},
			},
			{
				Payload: &pb.AlarmFeedMessage_SnapshotComplete{
					SnapshotComplete: true,
				},
			},
		},
	}
	client, cleanup := newBufconnClientWithAlarms(t, fake)
	defer cleanup()

	stream, err := client.StreamAlarms(context.Background(), &pb.StreamAlarmsRequest{
		AlarmFilterPrefix: "Tank01.",
	})
	if err != nil {
		t.Fatalf("StreamAlarms() error = %v", err)
	}

	var received []*pb.AlarmFeedMessage
	for {
		msg, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			t.Fatalf("stream.Recv() error = %v", err)
		}
		received = append(received, msg)
	}
	if len(received) != 2 {
		t.Fatalf("received count = %d, want 2", len(received))
	}
	if got := fake.streamRequest.GetAlarmFilterPrefix(); got != "Tank01." {
		t.Fatalf("captured filter prefix = %q", got)
	}
	if got := fake.streamAuth; got != "Bearer test-api-key" {
		t.Fatalf("stream authorization metadata = %q", got)
	}
}

func TestStreamAlarmsRejectsNilRequest(t *testing.T) {
	fake := &fakeGatewayWithAlarms{}
	client, cleanup := newBufconnClientWithAlarms(t, fake)
	defer cleanup()

	if _, err := client.StreamAlarms(context.Background(), nil); err == nil {
		t.Fatal("StreamAlarms(nil) returned no error")
	}
}

type fakeGatewayWithAlarms struct {
	pb.UnimplementedMxAccessGatewayServer

	acknowledgeRequest *pb.AcknowledgeAlarmRequest
	acknowledgeReply   *pb.AcknowledgeAlarmReply
	acknowledgeError   error
	acknowledgeAuth    string

	queryRequest    *pb.QueryActiveAlarmsRequest
	activeSnapshots []*pb.ActiveAlarmSnapshot

	streamRequest *pb.StreamAlarmsRequest
	feedMessages  []*pb.AlarmFeedMessage
	streamAuth    string
}

func (s *fakeGatewayWithAlarms) AcknowledgeAlarm(ctx context.Context, req *pb.AcknowledgeAlarmRequest) (*pb.AcknowledgeAlarmReply, error) {
	s.acknowledgeRequest = req
	s.acknowledgeAuth = authorizationFromContext(ctx)
	if s.acknowledgeError != nil {
		return nil, s.acknowledgeError
	}
	if s.acknowledgeReply != nil {
		return s.acknowledgeReply, nil
	}
	return &pb.AcknowledgeAlarmReply{
		CorrelationId: req.GetClientCorrelationId(),
		ProtocolStatus: &pb.ProtocolStatus{
			Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
		},
	}, nil
}

func (s *fakeGatewayWithAlarms) QueryActiveAlarms(req *pb.QueryActiveAlarmsRequest, stream grpc.ServerStreamingServer[pb.ActiveAlarmSnapshot]) error {
	s.queryRequest = req
	for _, snap := range s.activeSnapshots {
		if err := stream.Send(snap); err != nil {
			return err
		}
	}
	return nil
}

func (s *fakeGatewayWithAlarms) StreamAlarms(req *pb.StreamAlarmsRequest, stream grpc.ServerStreamingServer[pb.AlarmFeedMessage]) error {
	s.streamRequest = req
	s.streamAuth = authorizationFromContext(stream.Context())
	for _, msg := range s.feedMessages {
		if err := stream.Send(msg); err != nil {
			return err
		}
	}
	return nil
}

func newBufconnClientWithAlarms(t *testing.T, fake *fakeGatewayWithAlarms) (*Client, func()) {
	t.Helper()
	listener := bufconn.Listen(bufSize)
	server := grpc.NewServer()
	pb.RegisterMxAccessGatewayServer(server, fake)
	go func() {
		_ = server.Serve(listener)
	}()
	dialer := func(ctx context.Context, _ string) (net.Conn, error) {
		return listener.DialContext(ctx)
	}
	client, err := Dial(context.Background(), Options{
		Endpoint:    "bufnet",
		APIKey:      "test-api-key",
		Plaintext:   true,
		DialOptions: []grpc.DialOption{grpc.WithContextDialer(dialer)},
	})
	if err != nil {
		t.Fatalf("Dial() error = %v", err)
	}
	return client, func() {
		client.Close()
		server.Stop()
		listener.Close()
	}
}