fix(observer): remove dead nil checks from udpConn, add unit tests

The WrapUDPConn constructor returns nil when either arg is nil, so c.stats is always non-nil in all udpConn methods. Remove 8 redundant nil guards that were inconsistent with packetConn (which already skipped them). Add 65 unit tests across plugin, stats, and wrapper packages covering nil safety, race conditions, double-close, event marshaling, and reset-traffic semantics. Replace magic numbers in tests with named Kind constants.
2026-05-24 23:47:48 +08:00
parent 7c95d40e05
commit 689ba36e92
7 changed files with 1267 additions and 24 deletions
@@ -0,0 +1,203 @@
+package observer
+
+import (
+	"context"
+	"io"
+	"os"
+	"testing"
+
+	corelogger "github.com/go-gost/core/logger"
+	"github.com/go-gost/core/observer"
+	"github.com/go-gost/plugin/observer/proto"
+	xlogger "github.com/go-gost/x/logger"
+	xstats "github.com/go-gost/x/observer/stats"
+	"github.com/go-gost/x/service"
+	"google.golang.org/grpc"
+)
+
+func TestMain(m *testing.M) {
+	corelogger.SetDefault(xlogger.Nop())
+	os.Exit(m.Run())
+}
+
+// mockGrpcClient implements proto.ObserverClient for testing.
+type mockGrpcClient struct {
+	observeReply *proto.ObserveReply
+	observeErr   error
+}
+
+func (m *mockGrpcClient) Observe(ctx context.Context, in *proto.ObserveRequest, opts ...grpc.CallOption) (*proto.ObserveReply, error) {
+	if m.observeErr != nil {
+		return nil, m.observeErr
+	}
+	return m.observeReply, nil
+}
+
+type mockGrpcConn struct {
+	closed bool
+}
+
+func (m *mockGrpcConn) Invoke(ctx context.Context, method string, args, reply interface{}, opts ...grpc.CallOption) error {
+	return nil
+}
+
+func (m *mockGrpcConn) NewStream(ctx context.Context, desc *grpc.StreamDesc, method string, opts ...grpc.CallOption) (grpc.ClientStream, error) {
+	return nil, nil
+}
+
+func (m *mockGrpcConn) Close() error {
+	m.closed = true
+	return nil
+}
+
+func TestNewGRPCPlugin_ReturnsNonNil(t *testing.T) {
+	// NewGRPCPlugin should always return a non-nil observer. Even if the
+	// underlying gRPC dial would eventually fail, modern gRPC uses lazy
+	// connect so the constructor always returns a valid instance.
+	p := NewGRPCPlugin("test", "127.0.0.1:65535")
+	if p == nil {
+		t.Fatal("NewGRPCPlugin should never return nil")
+	}
+}
+
+func TestGrpcPlugin_Observe_NoOpClient_DoesNotPanic(t *testing.T) {
+	// When client is nil (e.g., from connection error or deliberate no-op),
+	// Observe should return nil without panicking.
+	p := &grpcPlugin{client: nil}
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err != nil {
+		t.Errorf("no-op observer should return nil on Observe, got %v", err)
+	}
+}
+
+func TestGrpcPlugin_Observe_NilClient(t *testing.T) {
+	p := &grpcPlugin{
+		conn:   nil,
+		client: nil,
+	}
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err != nil {
+		t.Errorf("nil client Observe should return nil, got %v", err)
+	}
+}
+
+func TestGrpcPlugin_Observe_EmptyEvents(t *testing.T) {
+	p := &grpcPlugin{
+		client: &mockGrpcClient{},
+	}
+	// Empty events should return nil without calling the client
+	err := p.Observe(context.Background(), nil)
+	if err != nil {
+		t.Errorf("empty events should return nil, got %v", err)
+	}
+}
+
+func TestGrpcPlugin_Observe_StatusEvent(t *testing.T) {
+	p := &grpcPlugin{
+		client: &mockGrpcClient{
+			observeReply: &proto.ObserveReply{Ok: true},
+		},
+	}
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{
+			Kind:    "service",
+			Service: "test-svc",
+			State:   "running",
+			Msg:     "started",
+		},
+	})
+	if err != nil {
+		t.Errorf("Observe with status event failed: %v", err)
+	}
+}
+
+func TestGrpcPlugin_Observe_StatsEvent(t *testing.T) {
+	p := &grpcPlugin{
+		client: &mockGrpcClient{
+			observeReply: &proto.ObserveReply{Ok: true},
+		},
+	}
+	err := p.Observe(context.Background(), []observer.Event{
+		xstats.StatsEvent{
+			Kind:         "service",
+			Service:      "test-svc",
+			Client:       "client1",
+			TotalConns:   10,
+			CurrentConns: 3,
+			InputBytes:   1024,
+			OutputBytes:  2048,
+			TotalErrs:    1,
+		},
+	})
+	if err != nil {
+		t.Errorf("Observe with stats event failed: %v", err)
+	}
+}
+
+func TestGrpcPlugin_Observe_ServerError(t *testing.T) {
+	p := &grpcPlugin{
+		client: &mockGrpcClient{
+			observeErr: io.ErrUnexpectedEOF,
+		},
+		log: xlogger.Nop(),
+	}
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err == nil {
+		t.Error("Observe should return error when server fails")
+	}
+}
+
+func TestGrpcPlugin_Observe_ReplyNotOk(t *testing.T) {
+	p := &grpcPlugin{
+		client: &mockGrpcClient{
+			observeReply: &proto.ObserveReply{Ok: false},
+		},
+	}
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err == nil {
+		t.Error("Observe should return error when reply is not ok")
+	}
+}
+
+func TestGrpcPlugin_Observe_NilReply(t *testing.T) {
+	p := &grpcPlugin{
+		client: &mockGrpcClient{
+			observeReply: nil,
+		},
+	}
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err == nil {
+		t.Error("Observe should return error when reply is nil")
+	}
+}
+
+func TestGrpcPlugin_Close_NilConn(t *testing.T) {
+	p := &grpcPlugin{conn: nil}
+	if err := p.Close(); err != nil {
+		t.Errorf("Close with nil conn should return nil, got %v", err)
+	}
+}
+
+func TestGrpcPlugin_Close_WithConn(t *testing.T) {
+	conn := &mockGrpcConn{}
+	p := &grpcPlugin{conn: conn}
+	if err := p.Close(); err != nil {
+		t.Errorf("Close failed: %v", err)
+	}
+	if !conn.closed {
+		t.Error("Close should close the underlying connection")
+	}
+}
+
+// Test that grpcPlugin satisfies observer.Observer
+var _ observer.Observer = (*grpcPlugin)(nil)
@@ -0,0 +1,244 @@
+package observer
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/go-gost/core/observer"
+	"github.com/go-gost/x/internal/plugin"
+	xstats "github.com/go-gost/x/observer/stats"
+	"github.com/go-gost/x/service"
+)
+
+func TestNewHTTPPlugin(t *testing.T) {
+	p := NewHTTPPlugin("test", "localhost:8080")
+	if p == nil {
+		t.Fatal("NewHTTPPlugin should not return nil")
+	}
+	hp, ok := p.(*httpPlugin)
+	if !ok {
+		t.Fatalf("NewHTTPPlugin returned %T, want *httpPlugin", p)
+	}
+	if hp.url != "http://localhost:8080" {
+		t.Errorf("url = %s, want http://localhost:8080", hp.url)
+	}
+	if hp.client == nil {
+		t.Error("client should not be nil")
+	}
+}
+
+func TestNewHTTPPlugin_NoScheme(t *testing.T) {
+	p := NewHTTPPlugin("test", "example.com:9999")
+	hp := p.(*httpPlugin)
+	if !strings.HasPrefix(hp.url, "http://") {
+		t.Errorf("expected http:// prefix, got %s", hp.url)
+	}
+}
+
+func TestNewHTTPPlugin_KeepsHTTPSScheme(t *testing.T) {
+	p := NewHTTPPlugin("test", "https://example.com/api")
+	hp := p.(*httpPlugin)
+	if !strings.HasPrefix(hp.url, "https://") {
+		t.Errorf("expected https:// prefix, got %s", hp.url)
+	}
+}
+
+func TestHTTPPlugin_Observe_StatusEvent(t *testing.T) {
+	var receivedBody json.RawMessage
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			t.Errorf("expected POST, got %s", r.Method)
+		}
+		if ct := r.Header.Get("Content-Type"); ct != "application/json" {
+			t.Errorf("Content-Type = %s, want application/json", ct)
+		}
+		json.NewDecoder(r.Body).Decode(&receivedBody)
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(`{"ok": true}`))
+	}))
+	defer srv.Close()
+
+	p := NewHTTPPlugin("test", srv.URL)
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{
+			Kind:    "service",
+			Service: "test-svc",
+			State:   "running",
+			Msg:     "started",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify the event was sent correctly
+	var req observeRequest
+	if err := json.Unmarshal(receivedBody, &req); err != nil {
+		t.Fatal(err)
+	}
+	if len(req.Events) != 1 {
+		t.Fatalf("expected 1 event, got %d", len(req.Events))
+	}
+	if req.Events[0].Kind != "service" {
+		t.Errorf("event kind = %s, want service", req.Events[0].Kind)
+	}
+	if req.Events[0].Service != "test-svc" {
+		t.Errorf("event service = %s, want test-svc", req.Events[0].Service)
+	}
+	if req.Events[0].Status == nil {
+		t.Fatal("status event should have Status field")
+	}
+	if req.Events[0].Status.State != "running" {
+		t.Errorf("state = %s, want running", req.Events[0].Status.State)
+	}
+}
+
+func TestHTTPPlugin_Observe_StatsEvent(t *testing.T) {
+	var receivedBody json.RawMessage
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		json.NewDecoder(r.Body).Decode(&receivedBody)
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(`{"ok": true}`))
+	}))
+	defer srv.Close()
+
+	p := NewHTTPPlugin("test", srv.URL)
+	err := p.Observe(context.Background(), []observer.Event{
+		xstats.StatsEvent{
+			Kind:         "traffic",
+			Service:      "svc",
+			Client:       "c1",
+			TotalConns:   100,
+			CurrentConns: 5,
+			InputBytes:   4096,
+			OutputBytes:  8192,
+			TotalErrs:    2,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var req observeRequest
+	if err := json.Unmarshal(receivedBody, &req); err != nil {
+		t.Fatal(err)
+	}
+	if len(req.Events) != 1 {
+		t.Fatalf("expected 1 event, got %d", len(req.Events))
+	}
+	if req.Events[0].Stats == nil {
+		t.Fatal("stats event should have Stats field")
+	}
+	if req.Events[0].Stats.TotalConns != 100 {
+		t.Errorf("totalConns = %d, want 100", req.Events[0].Stats.TotalConns)
+	}
+	if req.Events[0].Stats.CurrentConns != 5 {
+		t.Errorf("currentConns = %d, want 5", req.Events[0].Stats.CurrentConns)
+	}
+	if req.Events[0].Stats.InputBytes != 4096 {
+		t.Errorf("inputBytes = %d, want 4096", req.Events[0].Stats.InputBytes)
+	}
+	if req.Events[0].Stats.OutputBytes != 8192 {
+		t.Errorf("outputBytes = %d, want 8192", req.Events[0].Stats.OutputBytes)
+	}
+	if req.Events[0].Stats.TotalErrs != 2 {
+		t.Errorf("totalErrs = %d, want 2", req.Events[0].Stats.TotalErrs)
+	}
+	if req.Events[0].Client != "c1" {
+		t.Errorf("client = %s, want c1", req.Events[0].Client)
+	}
+}
+
+func TestHTTPPlugin_Observe_EmptyEvents(t *testing.T) {
+	p := NewHTTPPlugin("test", "http://localhost:9999")
+	err := p.Observe(context.Background(), nil)
+	if err != nil {
+		t.Errorf("empty events should return nil, got %v", err)
+	}
+}
+
+func TestHTTPPlugin_Observe_ServerError(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	p := NewHTTPPlugin("test", srv.URL)
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err == nil {
+		t.Error("Observe should return error on non-200 response")
+	}
+}
+
+func TestHTTPPlugin_Observe_NotOk(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(`{"ok": false}`))
+	}))
+	defer srv.Close()
+
+	p := NewHTTPPlugin("test", srv.URL)
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err == nil {
+		t.Error("Observe should return error when ok is false")
+	}
+}
+
+func TestHTTPPlugin_Observe_InvalidJSON(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(`not json`))
+	}))
+	defer srv.Close()
+
+	p := NewHTTPPlugin("test", srv.URL)
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err == nil {
+		t.Error("Observe should return error on invalid JSON response")
+	}
+}
+
+func TestHTTPPlugin_CustomHeader(t *testing.T) {
+	var receivedHeader string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		receivedHeader = r.Header.Get("X-Custom")
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte(`{"ok": true}`))
+	}))
+	defer srv.Close()
+
+	p := NewHTTPPlugin("test", srv.URL, plugin.HeaderOption(http.Header{
+		"X-Custom": []string{"my-value"},
+	}))
+	err := p.Observe(context.Background(), []observer.Event{
+		service.ServiceEvent{Kind: "service", Service: "test"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if receivedHeader != "my-value" {
+		t.Errorf("X-Custom header = %s, want my-value", receivedHeader)
+	}
+}
+
+func TestHTTPPlugin_Observe_WithClient(t *testing.T) {
+	// Ensure p.client is properly set by constructor and Observe doesn't
+	// hit the dead nil check.
+	p := NewHTTPPlugin("test", "http://example.com")
+	if p.(*httpPlugin).client == nil {
+		t.Fatal("client should not be nil after construction")
+	}
+}
+
+// Test that httpPlugin satisfies observer.Observer
+var _ observer.Observer = (*httpPlugin)(nil)