log_broker_hook_test.go

Documentation: github.com/cybertec-postgresql/pgwatch/v6/internal/log

     1  package log
     2  
     3  import (
     4  	"context"
     5  	"sync"
     6  	"testing"
     7  	"testing/synctest"
     8  	"time"
     9  
    10  	"github.com/sirupsen/logrus"
    11  	"github.com/stretchr/testify/assert"
    12  )
    13  
    14  type errorFormatter struct{}
    15  
    16  func (f *errorFormatter) Format(*logrus.Entry) ([]byte, error) {
    17  	return nil, assert.AnError
    18  }
    19  
    20  func TestFormatterError(t *testing.T) {
    21  	hook := NewBrokerHook(t.Context(), "info")
    22  	hook.SetBrokerFormatter(&errorFormatter{})
    23  	entry := logrus.NewEntry(logrus.New())
    24  	entry.Message = "hello broker"
    25  	entry.Level = logrus.InfoLevel
    26  	assert.Error(t, hook.Fire(entry))
    27  }
    28  
    29  func TestRemoveSubscriber(t *testing.T) {
    30  	hook := NewBrokerHook(t.Context(), "info")
    31  	msgChan1 := make(MessageChanType)
    32  	msgChan2 := make(MessageChanType)
    33  	hook.AddSubscriber(msgChan1)
    34  	hook.AddSubscriber(msgChan2)
    35  	assert.Equal(t, 2, len(hook.subscribers))
    36  
    37  	// Remove the first subscriber
    38  	hook.RemoveSubscriber(msgChan1)
    39  	assert.Equal(t, 1, len(hook.subscribers))
    40  
    41  	// Remove the last subscriber, this is where the "index out of range" error occurs
    42  	hook.RemoveSubscriber(msgChan2)
    43  	assert.Equal(t, 0, len(hook.subscribers))
    44  }
    45  
    46  func TestFire_CancelledContext(t *testing.T) {
    47  	ctx, cancel := context.WithCancel(t.Context())
    48  	cancel()
    49  	hook := NewBrokerHook(ctx, "info")
    50  	err := hook.Fire(logrus.NewEntry(logrus.New()))
    51  	assert.NoError(t, err)
    52  }
    53  
    54  func TestFire_DeliverToSubscriber(t *testing.T) {
    55  	synctest.Test(t, func(t *testing.T) {
    56  		ctx, cancel := context.WithCancel(t.Context())
    57  		hook := NewBrokerHook(ctx, "info")
    58  		msgChan := make(MessageChanType, 1)
    59  		hook.AddSubscriber(msgChan)
    60  
    61  		entry := logrus.NewEntry(logrus.New())
    62  		entry.Message = "hello broker"
    63  		entry.Level = logrus.InfoLevel
    64  		assert.NoError(t, hook.Fire(entry))
    65  
    66  		// Wait for the poll goroutine to drain input and deliver to msgChan.
    67  		synctest.Wait()
    68  
    69  		select {
    70  		case msg := <-msgChan:
    71  			assert.Contains(t, string(msg), "hello broker")
    72  		default:
    73  			t.Fatal("message not delivered")
    74  		}
    75  
    76  		cancel()        // signal poll to exit
    77  		synctest.Wait() // wait for poll goroutine to return
    78  	})
    79  }
    80  
    81  func TestFire_HighLoad(t *testing.T) {
    82  	synctest.Test(t, func(t *testing.T) {
    83  		// Construct without starting poll so the input channel stays full.
    84  		hook := &BrokerHook{
    85  			highLoadTimeout: time.Millisecond,
    86  			input:           make(chan MessageType, 1),
    87  			ctx:             t.Context(),
    88  			level:           "info",
    89  			mu:              &sync.Mutex{},
    90  			formatter:       defaultFormatter,
    91  		}
    92  		hook.input <- "fill" // fill to capacity; no poll goroutine to drain it
    93  
    94  		entry := logrus.NewEntry(logrus.New())
    95  		entry.Message = "to be dropped"
    96  		// The test goroutine is the only goroutine in the bubble.
    97  		// Fire blocks on time.After; fake time advances and takes the drop branch.
    98  		assert.NoError(t, hook.Fire(entry))
    99  	})
   100  }
   101  
   102  func TestSend_NoSubscribersIsNoop(t *testing.T) {
   103  	hook := NewBrokerHook(t.Context(), "info")
   104  	// No subscribers — send must return without blocking or panicking.
   105  	hook.send("msg")
   106  }
   107  
   108  func TestSend_FullSubscriberDrops(t *testing.T) {
   109  	hook := NewBrokerHook(t.Context(), "info")
   110  	// Unbuffered channel with no reader — send must take the default branch and not block.
   111  	msgChan := make(MessageChanType)
   112  	hook.AddSubscriber(msgChan)
   113  	hook.send("dropped")
   114  }
   115  
   116  func TestSetBrokerFormatter_NilResetsDefault(t *testing.T) {
   117  	hook := NewBrokerHook(t.Context(), "info")
   118  	hook.SetBrokerFormatter(&logrus.JSONFormatter{})
   119  	hook.SetBrokerFormatter(nil)
   120  	assert.Equal(t, defaultFormatter, hook.formatter)
   121  }
   122  
   123  func TestLevels(t *testing.T) {
   124  	cases := []struct {
   125  		level    string
   126  		expected []logrus.Level
   127  	}{
   128  		{"none", []logrus.Level{}},
   129  		{"debug", logrus.AllLevels},
   130  		{"info", []logrus.Level{logrus.PanicLevel, logrus.FatalLevel, logrus.ErrorLevel, logrus.WarnLevel, logrus.InfoLevel}},
   131  		{"error", []logrus.Level{logrus.PanicLevel, logrus.FatalLevel, logrus.ErrorLevel}},
   132  	}
   133  	for _, tc := range cases {
   134  		hook := NewBrokerHook(t.Context(), tc.level)
   135  		assert.Equal(t, tc.expected, hook.Levels(), "level=%s", tc.level)
   136  	}
   137  }
   138