worker/internal/distworker/results_test.go

package distworker

import (
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"rocketgit.ru/rsmon/worker/internal/wire"
)

func TestResultBufferSnapshotEmpty(t *testing.T) {
	b := newResultBuffer()
	assert.Nil(t, b.snapshot(0))
	assert.Nil(t, b.snapshot(10))
}

func TestResultBufferAppendAndSnapshot(t *testing.T) {
	b := newResultBuffer()
	for i := 0; i < 12; i++ {
		b.add(&ResultRow{MonitorID: int64(i), State: "OK"})
	}
	got := b.snapshot(20)
	// Buffer is fixed-size so we cap at recentResultsSize; here we
	// added only 12 entries so the buffer still holds all of them.
	require.Len(t, got, 12)
	assert.Equal(t, int64(0), got[0].MonitorID)
	assert.Equal(t, int64(11), got[11].MonitorID)
}

func TestResultBufferEvictsOldest(t *testing.T) {
	b := newResultBuffer()
	total := recentResultsSize + 5
	for i := 0; i < total; i++ {
		b.add(&ResultRow{MonitorID: int64(i)})
	}
	got := b.snapshot(recentResultsSize)
	require.Len(t, got, recentResultsSize)
	// The first surviving entry is i=5 (the (N+1)-th add).
	assert.Equal(t, int64(5), got[0].MonitorID)
	assert.Equal(t, int64(total-1), got[recentResultsSize-1].MonitorID)
}

func TestResultBufferConcurrentAccess(t *testing.T) {
	b := newResultBuffer()
	var wg sync.WaitGroup
	for w := 0; w < 4; w++ {
		wg.Add(1)
		go func(off int) {
			defer wg.Done()
			for i := 0; i < 100; i++ {
				b.add(&ResultRow{MonitorID: int64(off*100 + i)})
			}
		}(w)
	}
	for w := 0; w < 4; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < 50; i++ {
				_ = b.snapshot(10)
			}
		}()
	}
	wg.Wait()
	got := b.snapshot(recentResultsSize)
	assert.NotNil(t, got)
}

func TestNotificationBufferAppendAndSnapshot(t *testing.T) {
	b := newNotificationBuffer()
	for i := 0; i < 3; i++ {
		b.add(&NotificationRow{Subject: "s" + itoaForTest(i), At: time.Now()})
	}
	got := b.snapshot(10)
	require.Len(t, got, 3)
	assert.Equal(t, "s0", got[0].Subject)
}

func TestRunnerRecentResultsBeforeStart(t *testing.T) {
	r := NewRunner(&Config{MaxConcurrency: 2})
	assert.Nil(t, r.RecentResults(5),
		"RecentResults must be safe before Start")
}

func TestRunnerRecentNotificationsBeforeStart(t *testing.T) {
	r := NewRunner(&Config{MaxConcurrency: 2})
	assert.Nil(t, r.RecentNotifications(5))
}

func TestRunnerRecordNotification(t *testing.T) {
	r := NewRunner(&Config{MaxConcurrency: 2})
	r.RecordNotification(&NotificationRow{Subject: "x", OK: true})
	r.RecordNotification(&NotificationRow{Subject: "y", OK: false, Error: "boom"})
	got := r.RecentNotifications(10)
	require.Len(t, got, 2)
	assert.Equal(t, "x", got[0].Subject)
	assert.True(t, got[0].OK)
	assert.False(t, got[1].OK)
	assert.Equal(t, "boom", got[1].Error)
}

func TestRunnerIdentityFieldsFromInit(t *testing.T) {
	executor := func(interface{}) interface{} { return []wire.CheckResultReport{} }
	r := newTestRunner(t, 4, 1, executor)
	r.applyInit(&wire.WorkerInit{
		WorkerID:     "w-1",
		RegionCode:   "eu",
		Version:      "v1.2.3",
		Capabilities: []string{"http", "ssl"},
	})
	assert.Equal(t, "w-1", r.WorkerID())
	assert.Equal(t, "eu", r.RegionCode())
	assert.Equal(t, "v1.2.3", r.WorkerVersion())
	assert.Equal(t, []string{"http", "ssl"}, r.WorkerCapabilities())
}

func TestRunnerTokenAccessors(t *testing.T) {
	r := NewRunner(&Config{MaxConcurrency: 1, Token: "tok-1"})
	assert.Equal(t, "tok-1", r.Token())
	assert.True(t, r.TokenRotatedAt().IsZero(), "no rotation yet")
	r.tokenRotatedMu.Lock()
	r.tokenRotatedAt = time.Now().UTC()
	r.tokenRotatedMu.Unlock()
	assert.False(t, r.TokenRotatedAt().IsZero())
}

func itoaForTest(n int) string {
	if n == 0 {
		return "0"
	}
	const d = "0123456789"
	out := ""
	for n > 0 {
		out = string(d[n%10]) + out
		n /= 10
	}
	return out
}