EvoBGP/internal/telemetry/collector_test.go

package telemetry

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"evobgp/internal/store"
)

func TestNextDelayBounded(t *testing.T) {
	cfg := Config{Interval: 20 * time.Second, BackoffMax: 2 * time.Minute}
	for _, tc := range []struct {
		failures int
		failed   bool
		min, max time.Duration
	}{
		{0, false, 20 * time.Second, 24 * time.Second},
		{0, true, 40 * time.Second, 48 * time.Second},
		{8, true, 2 * time.Minute, 144 * time.Second},
	} {
		got := nextDelay(cfg, tc.failures, tc.failed)
		if got < tc.min || got > tc.max {
			t.Fatalf("delay(%d,%v)=%s outside [%s,%s]", tc.failures, tc.failed, got, tc.min, tc.max)
		}
	}
}

func TestCollectorNilStoreCloses(t *testing.T) {
	c := Start(context.Background(), nil, Config{})
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := c.Close(ctx); err != nil {
		t.Fatal(err)
	}
}

type collectorTestStore struct {
	mu        sync.Mutex
	claims    []store.TelemetryClaim
	claimCall int
	finished  chan struct{}
}

func (s *collectorTestStore) ClaimTelemetry(_ context.Context, owner string, limit int, _ time.Duration, _ string) ([]store.TelemetryClaim, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.claimCall++
	if len(s.claims) == 0 {
		return nil, nil
	}
	claim := s.claims[0]
	s.claims = s.claims[1:]
	claim.OwnerID = owner
	return []store.TelemetryClaim{claim}, nil
}
func (s *collectorTestStore) CompleteTelemetry(_ context.Context, _ store.TelemetryClaim, _ store.TelemetryResult, _ time.Time) error {
	s.finished <- struct{}{}
	return nil
}
func (s *collectorTestStore) ListTelemetry(context.Context, string) ([]store.TelemetrySnapshot, error) {
	return nil, nil
}
func (s *collectorTestStore) RequestTelemetryRefresh(context.Context, string, time.Duration) (bool, error) {
	return false, nil
}

type roundTripFunc func(*http.Request) (*http.Response, error)

func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }

func TestCollectorBoundsPollsAndStopsClaims(t *testing.T) {
	var active, maxActive, requests atomic.Int32
	client := &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
		current := active.Add(1)
		for {
			old := maxActive.Load()
			if current <= old || maxActive.CompareAndSwap(old, current) {
				break
			}
		}
		defer active.Add(-1)
		requests.Add(1)
		time.Sleep(100 * time.Millisecond)
		body := `{"sessions":[]}`
		if strings.HasSuffix(r.URL.Path, "/health") {
			body = `{"ok":true}`
		}
		return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header)}, nil
	})}
	host := "node.test"
	st := &collectorTestStore{finished: make(chan struct{}, 4)}
	for i := 0; i < 4; i++ {
		st.claims = append(st.claims, store.TelemetryClaim{TelemetrySnapshot: store.TelemetrySnapshot{TenantID: "tenant", TargetID: fmt.Sprint(i), Kind: "remote_agent"}, Speaker: &store.Speaker{ID: fmt.Sprint(i), Role: "replica", MetaJSON: fmt.Sprintf(`{"agent_domain":%q,"agent_secret":"secret"}`, host)}})
	}
	c := Start(context.Background(), st, Config{Workers: 2, Interval: 20 * time.Second, Timeout: 3 * time.Second, Lease: 8 * time.Second, BackoffMax: time.Minute, HTTPClient: client})
	for i := 0; i < 4; i++ {
		select {
		case <-st.finished:
		case <-time.After(6 * time.Second):
			t.Fatal("collector did not finish bounded claims")
		}
	}
	if maxActive.Load() > 4 || requests.Load() != 8 {
		t.Fatalf("requests=%d maxActive=%d", requests.Load(), maxActive.Load())
	}
	if err := c.Close(context.Background()); err != nil {
		t.Fatal(err)
	}
	st.mu.Lock()
	before := st.claimCall
	st.mu.Unlock()
	time.Sleep(1100 * time.Millisecond)
	st.mu.Lock()
	after := st.claimCall
	st.mu.Unlock()
	if after != before {
		t.Fatalf("claims continued after Close: %d -> %d", before, after)
	}
}