DebTik/internal/bwtest/telemetry_test.go

package bwtest

import (
	"context"
	"net"
	"testing"
	"time"
)

func TestSummaryCoversSamplesBeyondRetentionWindow(t *testing.T) {
	c := NewClient()
	for i := 1; i <= 300; i++ {
		rate := 1.0
		if i == 1 {
			rate = 50
		}
		c.recordSampleLocked(&LiveSample{ElapsedSec: float64(i), TxBytes: int64(i) * 125000, RxBytes: int64(i) * 250000, TxMbps: rate, RxMbps: 2, JitterMs: float64(i), TxPackets: int64(i) * 10, RxPackets: int64(i) * 10, LostPackets: int64(i)})
	}
	if len(c.samples) != 200 {
		t.Fatal("unbounded history")
	}
	c.finalizeSummary()
	s := c.status.Summary
	if s.AvgTxMbps != 1 || s.AvgRxMbps != 2 || s.PeakTxMbps != 50 || s.AvgJitterMs != 150.5 {
		t.Fatalf("summary lost early samples: %+v", s)
	}
	if s.AvgLossPct < 4.76 || s.AvgLossPct > 4.77 {
		t.Fatal("loss not computed from full totals")
	}
}

func TestObserversReceiveBroadcastAndCanUnsubscribe(t *testing.T) {
	c := NewClient()
	a, stopA := c.Subscribe()
	b, stopB := c.Subscribe()
	defer stopB()
	c.recordSampleLocked(&LiveSample{ElapsedSec: 1})
	if (<-a).ElapsedSec != 1 || (<-b).ElapsedSec != 1 {
		t.Fatal("observers competed for samples")
	}
	stopA()
	stopA()
	if _, ok := <-a; ok {
		t.Fatal("subscription was not closed")
	}
	c.recordSampleLocked(&LiveSample{ElapsedSec: 2})
	if (<-b).ElapsedSec != 2 {
		t.Fatal("other observer stopped")
	}
}

func TestZeroDurationRemainsContinuous(t *testing.T) {
	// A closed local port makes the background connection end immediately; this
	// checks normalization without sending traffic or waiting ten seconds.
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	port := l.Addr().(*net.TCPAddr).Port
	l.Close()
	c := NewClient()
	if err := c.Run(context.Background(), SessionRequest{Target: "127.0.0.1", Port: port, Duration: 0}); err != nil {
		t.Fatal(err)
	}
	defer c.Stop()
	if c.GetStatus().Duration != 0 {
		t.Fatal("continuous duration was replaced")
	}
}

func TestContinuousSessionSurvivesTenSecondsAndStops(t *testing.T) {
	if testing.Short() {
		t.Skip("continuous session integration test")
	}
	s := testServer(t)
	c := NewClient()
	defer c.Stop()
	if err := c.Run(context.Background(), SessionRequest{Target: "127.0.0.1", Port: s.config.Port, Protocol: "udp", Direction: "both", Duration: 0, BandwidthLimit: 100000}); err != nil {
		t.Fatal(err)
	}
	timer := time.NewTimer(11 * time.Second)
	defer timer.Stop()
	<-timer.C
	if st := c.GetStatus(); !st.Running || st.Error != "" || st.ElapsedSec < 10 {
		t.Fatalf("continuous session ended: %+v", st)
	}
	c.Stop()
	deadline := time.Now().Add(2 * time.Second)
	for c.GetStatus().Running && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}
	st := c.GetStatus()
	if st.Running || st.Summary == nil || st.Summary.TotalTxBytes == 0 || st.Summary.TotalRxBytes == 0 {
		t.Fatal("continuous session failed to stop or exchange data")
	}
}