package health import ( "context" "errors" "testing" "time" controlplanev1 "proxy-pool/gen/controlplane/v1" healthDomain "proxy-pool/internal/domain/health" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" ) func TestGRPCHandlerReportsAcceptedAndRejectedObservations(t *testing.T) { global := &recordingGlobalStore{} target := &recordingTargetStore{} reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil }) if err != nil { t.Fatalf("NewReducer(): %v", err) } identity := &recordingCheckerIdentity{} handler, err := NewGRPCHandler(reducer, identity, GRPCHandlerOptions{MaxObservationsPerBatch: 3}) if err != nil { t.Fatalf("NewGRPCHandler(): %v", err) } now := time.Date(2026, 7, 31, 14, 0, 0, 0, time.UTC) response, err := handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{ grpcHealthObservation("task-basic", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, now), grpcHealthObservation("task-target", controlplanev1.CheckLevel_CHECK_LEVEL_TARGET, now.Add(time.Second)), {TaskId: "invalid", ProxyId: "proxy-a", Level: controlplanev1.CheckLevel_CHECK_LEVEL_UNSPECIFIED}, }, }) if err != nil || response.GetAccepted() != 2 || response.GetRejected() != 1 { t.Fatalf("ReportObservations() = (%+v, %v)", response, err) } if identity.checkerID != "checker-a" || len(global.commands) != 1 || len(target.commands) != 1 { t.Fatalf("identity=%q global=%d target=%d", identity.checkerID, len(global.commands), len(target.commands)) } } func TestGRPCHandlerRejectsInvalidCallsBeforeStoreMutation(t *testing.T) { global := &recordingGlobalStore{} target := &recordingTargetStore{} reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil }) if err != nil { t.Fatalf("NewReducer(): %v", err) } identity := &recordingCheckerIdentity{err: errors.New("not allowed")} handler, err := NewGRPCHandler(reducer, identity, GRPCHandlerOptions{MaxObservationsPerBatch: 1}) if err != nil { t.Fatalf("NewGRPCHandler(): %v", err) } _, err = handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{ grpcHealthObservation("task-basic", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), }, }) if status.Code(err) != codes.PermissionDenied || len(global.commands) != 0 || len(target.commands) != 0 { t.Fatalf("ReportObservations(denied) = %v; global=%d target=%d", err, len(global.commands), len(target.commands)) } _, err = handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{ grpcHealthObservation("task-one", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), grpcHealthObservation("task-two", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), }, }) if status.Code(err) != codes.InvalidArgument { t.Fatalf("ReportObservations(too large) = %v, want InvalidArgument", err) } } func TestGRPCHandlerReturnsUnavailableForStoreFailure(t *testing.T) { global := &recordingGlobalStore{err: errors.New("redis unavailable")} target := &recordingTargetStore{} reducer, err := NewReducer(global, target, func(context.Context, healthDomain.Observation) (int, error) { return 2, nil }) if err != nil { t.Fatalf("NewReducer(): %v", err) } handler, err := NewGRPCHandler(reducer, &recordingCheckerIdentity{}, GRPCHandlerOptions{MaxObservationsPerBatch: 1}) if err != nil { t.Fatalf("NewGRPCHandler(): %v", err) } _, err = handler.ReportObservations(context.Background(), &controlplanev1.ObservationBatch{ CheckerId: "checker-a", Observations: []*controlplanev1.HealthObservation{ grpcHealthObservation("task-basic", controlplanev1.CheckLevel_CHECK_LEVEL_BASIC, time.Now()), }, }) if status.Code(err) != codes.Unavailable { t.Fatalf("ReportObservations(store failure) = %v, want Unavailable", err) } } func grpcHealthObservation(taskID string, level controlplanev1.CheckLevel, observedAt time.Time) *controlplanev1.HealthObservation { item := &controlplanev1.HealthObservation{ TaskId: taskID, ProxyId: "proxy-a", Level: level, Success: true, Latency: durationpb.New(time.Millisecond), ObservedAt: timestamppb.New(observedAt), } if level == controlplanev1.CheckLevel_CHECK_LEVEL_TARGET { item.RoutingName = "route-a" item.TargetUrl = "https://target.example/check" } return item } type recordingCheckerIdentity struct { checkerID string err error } func (identity *recordingCheckerIdentity) AuthorizeChecker(_ context.Context, checkerID string) error { identity.checkerID = checkerID return identity.err }