tutti/packages/device-link/networkchange/monitor_test.go

package networkchange

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"
)

func TestMonitorStartsAtOneAndAdvancesOnlyForDistinctSuccessfulSamples(t *testing.T) {
	source := newTestSource(Fingerprint{1})
	monitor, err := NewMonitorWithConfig(source, Config{
		Debounce:      time.Millisecond,
		PollInterval:  time.Hour,
		SafetyRecheck: 100 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	changes, unsubscribe, err := monitor.Subscribe(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer unsubscribe()
	runDone := make(chan error, 1)
	go func() { runDone <- monitor.Run(ctx) }()
	source.WaitForSamples(t, 1)
	if got := monitor.Generation(); got != 1 {
		t.Fatalf("initial generation = %d, want 1", got)
	}
	waitForMode(t, monitor, ObservationWatching)
	status := monitor.Status()
	if status.Mode != ObservationWatching || status.PollReason != PollReasonNone || status.Generation != 1 ||
		!status.SampleHealthy || status.ConsecutiveSampleFailures != 0 || status.LastSampleAt.IsZero() {
		t.Fatalf("watching status = %#v, want active watcher at generation 1", status)
	}

	source.Trigger()
	assertNoChange(t, changes)

	source.SetFingerprint(Fingerprint{2})
	source.Trigger()
	change := waitForChange(t, changes)
	if change.PreviousGeneration != 1 || change.Generation != 2 {
		t.Fatalf("change = %#v, want 1 -> 2", change)
	}
	if got := monitor.Generation(); got != 2 {
		t.Fatalf("generation after first change = %d, want 2", got)
	}

	source.SetError(errors.New("sample unavailable"))
	source.SetFingerprint(Fingerprint{3})
	source.Trigger()
	assertNoChange(t, changes)
	if got := monitor.Generation(); got != 2 {
		t.Fatalf("generation after failed sample = %d, want 2", got)
	}
	status = monitor.Status()
	if status.SampleHealthy || status.ConsecutiveSampleFailures != 1 {
		t.Fatalf("failed sample status = %#v, want one consecutive failure", status)
	}

	source.SetError(nil)
	source.Trigger()
	change = waitForChange(t, changes)
	if change.PreviousGeneration != 2 || change.Generation != 3 {
		t.Fatalf("change after recovery = %#v, want 2 -> 3", change)
	}
	status = monitor.Status()
	if !status.SampleHealthy || status.ConsecutiveSampleFailures != 0 {
		t.Fatalf("recovered sample status = %#v, want healthy", status)
	}
	cancel()
	if err := <-runDone; err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if status := monitor.Status(); status.Mode != ObservationStopped {
		t.Fatalf("stopped status = %#v, want stopped", status)
	}
}

func TestMonitorFallsBackToPollingAfterWatcherFailure(t *testing.T) {
	source := newTestSource(Fingerprint{1})
	source.watch = make(chan struct{})
	close(source.watch)
	monitor, err := NewMonitorWithConfig(source, Config{
		Debounce:      time.Millisecond,
		PollInterval:  2 * time.Millisecond,
		SafetyRecheck: 10 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	changes, unsubscribe, err := monitor.Subscribe(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer unsubscribe()
	runDone := make(chan error, 1)
	go func() { runDone <- monitor.Run(ctx) }()
	source.WaitForSamples(t, 1)
	waitForMode(t, monitor, ObservationPolling)
	status := monitor.Status()
	if status.Mode != ObservationPolling || status.PollReason != PollReasonWatcherClosed {
		t.Fatalf("fallback status = %#v, want closed-watcher polling", status)
	}
	source.SetFingerprint(Fingerprint{2})
	change := waitForChange(t, changes)
	if change.Generation != 2 {
		t.Fatalf("polling change generation = %d, want 2", change.Generation)
	}
	cancel()
	if err := <-runDone; err != nil {
		t.Fatalf("Run() error = %v", err)
	}
	if status := monitor.Status(); status.Mode != ObservationStopped {
		t.Fatalf("stopped fallback status = %#v, want stopped", status)
	}
}

func TestMonitorCoalescesSlowSubscriberToNewestGeneration(t *testing.T) {
	source := newTestSource(Fingerprint{1})
	monitor, err := NewMonitorWithConfig(source, Config{
		Debounce:      time.Millisecond,
		PollInterval:  time.Hour,
		SafetyRecheck: time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	changes, unsubscribe, err := monitor.Subscribe(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer unsubscribe()
	runDone := make(chan error, 1)
	go func() { runDone <- monitor.Run(ctx) }()
	source.WaitForSamples(t, 1)
	for value := byte(2); value < 12; value++ {
		source.SetFingerprint(Fingerprint{value})
		source.Trigger()
		// Each trigger is debounced and sampled synchronously by the monitor;
		// waiting for the generation makes this test independent of scheduler
		// ordering while intentionally leaving the subscription unread.
		waitForGeneration(t, monitor, uint64(value))
	}
	change := waitForChange(t, changes)
	if change.Generation < 2 || change.Generation > 11 {
		t.Fatalf("first queued generation = %d, want a retained generation", change.Generation)
	}
	for {
		select {
		case change = <-changes:
		default:
			if change.Generation != 11 {
				t.Fatalf("newest queued generation = %d, want 11", change.Generation)
			}
			cancel()
			if err := <-runDone; err != nil {
				t.Fatalf("Run() error = %v", err)
			}
			return
		}
	}
}

type testSource struct {
	mu          sync.Mutex
	fingerprint Fingerprint
	err         error
	watch       chan struct{}
	samples     int
	sampleSeen  chan struct{}
}

func newTestSource(fingerprint Fingerprint) *testSource {
	return &testSource{
		fingerprint: fingerprint,
		watch:       make(chan struct{}, 32),
		sampleSeen:  make(chan struct{}, 32),
	}
}

func (s *testSource) Sample(context.Context) (Fingerprint, error) {
	s.mu.Lock()
	s.samples++
	fingerprint, err := s.fingerprint, s.err
	s.mu.Unlock()
	select {
	case s.sampleSeen <- struct{}{}:
	default:
	}
	return fingerprint, err
}

func (s *testSource) Watch(context.Context) (<-chan struct{}, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.watch, nil
}

func (s *testSource) SetFingerprint(fingerprint Fingerprint) {
	s.mu.Lock()
	s.fingerprint = fingerprint
	s.mu.Unlock()
}

func (s *testSource) SetError(err error) {
	s.mu.Lock()
	s.err = err
	s.mu.Unlock()
}

func (s *testSource) Trigger() {
	s.mu.Lock()
	watch := s.watch
	s.mu.Unlock()
	select {
	case watch <- struct{}{}:
	default:
	}
}

func (s *testSource) WaitForSamples(t *testing.T, count int) {
	t.Helper()
	for i := 0; i < count; i++ {
		select {
		case <-s.sampleSeen:
		case <-time.After(time.Second):
			t.Fatal("timed out waiting for network sample")
		}
	}
}

func waitForChange(t *testing.T, changes <-chan Change) Change {
	t.Helper()
	select {
	case change := <-changes:
		return change
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for network change")
		return Change{}
	}
}

func waitForGeneration(t *testing.T, monitor *Monitor, generation uint64) {
	t.Helper()
	deadline := time.Now().Add(time.Second)
	for time.Now().Before(deadline) {
		if monitor.Generation() == generation {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("generation = %d, want %d", monitor.Generation(), generation)
}

func waitForMode(t *testing.T, monitor *Monitor, mode ObservationMode) {
	t.Helper()
	deadline := time.Now().Add(time.Second)
	for time.Now().Before(deadline) {
		if monitor.Status().Mode == mode {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("monitor mode = %q, want %q", monitor.Status().Mode, mode)
}

func assertNoChange(t *testing.T, changes <-chan Change) {
	t.Helper()
	select {
	case change := <-changes:
		t.Fatalf("unexpected network change: %#v", change)
	case <-time.After(15 * time.Millisecond):
	}
}