hotime/log/capture_test.go

package log

import (
	"bytes"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	logrus "github.com/sirupsen/logrus"
	"github.com/rs/zerolog"
)

// captureBuf 挂到 multiWriter，收集 EmitCaptured 写入的 JSON 行。
type captureBuf struct {
	mu   sync.Mutex
	lines []string
}

func (c *captureBuf) Write(p []byte) (int, error) {
	c.mu.Lock()
	c.lines = append(c.lines, string(bytes.TrimSpace(p)))
	c.mu.Unlock()
	return len(p), nil
}

func (c *captureBuf) lastJSON() map[string]interface{} {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.lines) == 0 {
		return nil
	}
	var m map[string]interface{}
	_ = json.Unmarshal([]byte(c.lines[len(c.lines)-1]), &m)
	return m
}

func newLoggerWithCapture(logLevel int) (*Logger, *captureBuf) {
	buf := &captureBuf{}
	l := NewLogger(logLevel, "", 0)
	l.mw.add(buf)
	return l, buf
}

func startFakeSeqServer(t *testing.T, onBody func(string)) *httptest.Server {
	t.Helper()
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/api/events/raw" {
			http.NotFound(w, r)
			return
		}
		body, _ := io.ReadAll(r.Body)
		onBody(string(body))
		w.WriteHeader(http.StatusCreated)
	}))
}

func TestEmitCaptured_LogLevelZero(t *testing.T) {
	l, buf := newLoggerWithCapture(0)
	l.EmitCaptured(zerolog.InfoLevel, "stdout", "请求参数GetReqMap /api/test")
	m := buf.lastJSON()
	if m == nil {
		t.Fatal("expected captured line")
	}
	if m["source"] != "stdout" {
		t.Fatalf("source=%v want stdout", m["source"])
	}
	if m["level"] != "info" {
		t.Fatalf("level=%v want info", m["level"])
	}
	if !strings.Contains(m["message"].(string), "GetReqMap") {
		t.Fatalf("message=%v", m["message"])
	}
}

func TestEmitCaptured_StderrSource(t *testing.T) {
	l, buf := newLoggerWithCapture(0)
	l.EmitCaptured(zerolog.ErrorLevel, "stderr", "driver error")
	m := buf.lastJSON()
	if m["source"] != "stderr" {
		t.Fatalf("source=%v want stderr", m["source"])
	}
	if m["level"] != "error" {
		t.Fatalf("level=%v want error", m["level"])
	}
}

func TestToClef_PlainText(t *testing.T) {
	sw := &SeqWriter{instance: "127.0.0.1:8081"}
	clef := sw.toClef([]byte("plain log line\n"))
	var m map[string]interface{}
	if err := json.Unmarshal(clef, &m); err != nil {
		t.Fatal(err)
	}
	if m["@mt"] != "plain log line" {
		t.Fatalf("@mt=%v", m["@mt"])
	}
	if m["source"] != "stdout" {
		t.Fatalf("source=%v", m["source"])
	}
	if m["@l"] != "Information" {
		t.Fatalf("@l=%v", m["@l"])
	}
}

func TestStdoutCapture_LongLine(t *testing.T) {
	l, buf := newLoggerWithCapture(1)
	pr, pw := io.Pipe()
	done := make(chan struct{})
	go func() {
		CaptureStream(l, zerolog.InfoLevel, "stdout", pr)
		close(done)
	}()

	longLine := strings.Repeat("x", 70*1024) + "\n"
	shortLine := "after-long-line\n"
	if _, err := pw.Write([]byte(longLine)); err != nil {
		t.Fatal(err)
	}
	if _, err := pw.Write([]byte(shortLine)); err != nil {
		t.Fatal(err)
	}
	time.Sleep(100 * time.Millisecond)

	buf.mu.Lock()
	n := len(buf.lines)
	buf.mu.Unlock()
	if n < 2 {
		t.Fatalf("expected >=2 captured lines, got %d", n)
	}
	last := buf.lastJSON()
	if last == nil || !strings.Contains(last["message"].(string), "after-long-line") {
		t.Fatalf("short line after long line not captured: %v", last)
	}
	_ = pw.Close()
	<-done
}

func TestLogrusBridge_AfterRedirect(t *testing.T) {
	l, buf := newLoggerWithCapture(1)
	pr, pw := io.Pipe()
	go CaptureStream(l, zerolog.InfoLevel, "stdout", pr)

	logrus.SetOutput(pw)
	logrus.SetLevel(logrus.InfoLevel)
	logrus.Info("wechat-sdk-log")
	time.Sleep(100 * time.Millisecond)

	m := buf.lastJSON()
	if m == nil || !strings.Contains(m["message"].(string), "wechat-sdk-log") {
		t.Fatalf("logrus not captured: %v", m)
	}
}

func TestSeqWriter_FakeServer(t *testing.T) {
	var gotBody string
	var mu sync.Mutex
	srv := startFakeSeqServer(t, func(body string) {
		mu.Lock()
		gotBody = body
		mu.Unlock()
	})
	defer srv.Close()

	l := NewLogger(1, "", 0)
	l.SetSeqWriter(srv.URL, "", "test:8081")
	l.Info().Msg("seq-test-message")

	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		mu.Lock()
		b := gotBody
		mu.Unlock()
		if strings.Contains(b, "seq-test-message") {
			return
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatalf("Seq did not receive message, body=%q", gotBody)
}

func TestEmitRecoveredPanic_LogFields(t *testing.T) {
	l, buf := newLoggerWithCapture(0)
	l.EmitRecoveredPanic("test panic reason")
	m := buf.lastJSON()
	if m == nil {
		t.Fatal("no output")
	}
	if m["source"] != "panic" {
		t.Fatalf("source=%v", m["source"])
	}
	if m["level"] != "error" {
		t.Fatalf("level=%v", m["level"])
	}
	if !strings.Contains(m["message"].(string), "test panic reason") {
		t.Fatalf("message=%v", m["message"])
	}
	if stack, ok := m["stack"].(string); !ok || stack == "" {
		t.Fatalf("stack missing: %v", m["stack"])
	}
}