3x-ui-agent/internal/agent/ping_test.go

package agent

import (
	"encoding/json"
	"net"
	"testing"
	"time"
)

// probeTCP against a live listener yields a real non-negative
// latency; a closed port and a malformed target both file -1; 0 is
// never a latency (contract: failures are negative or absent).
func TestProbeTCPOutcomes(t *testing.T) {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer ln.Close()
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			conn.Close()
		}
	}()

	ms, sampled := probeTCP(ln.Addr().String(), nil)
	if !sampled || ms < 0 {
		t.Fatalf("live target: ms=%d sampled=%v", ms, sampled)
	}
	ms, sampled = probeTCP("127.0.0.1:1", nil)
	if !sampled || ms != -1 {
		t.Fatalf("closed port: ms=%d sampled=%v", ms, sampled)
	}
	ms, sampled = probeTCP("not-a-target", nil)
	if !sampled || ms != -1 {
		t.Fatalf("malformed: ms=%d sampled=%v", ms, sampled)
	}
	ms, sampled = probeTCP("256.256.256.256:80", nil)
	if !sampled || ms != -1 {
		t.Fatalf("unresolvable: ms=%d sampled=%v", ms, sampled)
	}
}

// applyTasks: an unchanged task keeps its loop; a changed target or
// interval restarts it; a removed task is cancelled; the set is
// truncated at the cap with intervals clamped into range.
func TestProbeRunnerApplyTasks(t *testing.T) {
	r := newProbeRunner()
	defer r.stop()

	r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 60}})
	if len(r.tasks) != 1 {
		t.Fatalf("tasks %v", r.tasks)
	}
	first := r.tasks[1]

	// Unchanged: same loop object survives.
	r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 60}})
	if r.tasks[1] != first {
		t.Fatal("unchanged task restarted")
	}
	// Changed interval: replaced, clamped values are what runs.
	r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 120}})
	if r.tasks[1] == first || r.tasks[1].interval != 120 {
		t.Fatal("changed task not restarted")
	}
	// Removed: cancelled.
	r.applyTasks(nil)
	if len(r.tasks) != 0 {
		t.Fatalf("tasks after clear %v", r.tasks)
	}

	// Clamp + cap.
	many := make([]ProbeTask, 0, 70)
	for i := 1; i <= 70; i++ {
		many = append(many, ProbeTask{ID: int64(i), Target: "127.0.0.1:1", Interval: 1})
	}
	r.applyTasks(many)
	if len(r.tasks) != maxProbeTasksRun {
		t.Fatalf("cap: %d running", len(r.tasks))
	}
	if r.tasks[1].interval != minProbeIntervalSec {
		t.Fatalf("clamp: interval %d", r.tasks[1].interval)
	}
}

// A running loop files samples into the buffer; drain takes them and
// requeue restores order ahead of newer samples.
func TestProbeRunnerResultFlow(t *testing.T) {
	r := newProbeRunner()
	defer r.stop()
	r.applyTasks([]ProbeTask{{ID: 9, Target: "127.0.0.1:1", Interval: 3600}})
	// The loop probes immediately on start.
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if got := r.drainResults(); len(got) > 0 {
			if got[0].TaskID != 9 || got[0].LatencyMs != -1 {
				t.Fatalf("sample %+v", got[0])
			}
			goto drained
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatal("no immediate sample from a fresh task")
drained:

	r.addResult(ProbeResult{TaskID: 1, LatencyMs: 5})
	batch := r.drainResults()
	r.addResult(ProbeResult{TaskID: 1, LatencyMs: 7})
	r.requeueResults(batch)
	got := r.drainResults()
	if len(got) != 2 || got[0].LatencyMs != 5 || got[1].LatencyMs != 7 {
		t.Fatalf("requeue order %+v", got)
	}
	if again := r.drainResults(); again != nil {
		t.Fatalf("second drain %+v", again)
	}
}

// Buffer overflow drops the oldest samples, keeping the newest cap.
func TestProbeRunnerBufferCap(t *testing.T) {
	r := newProbeRunner()
	defer r.stop()
	for i := 0; i < probeResultCap+10; i++ {
		r.addResult(ProbeResult{TaskID: 1, LatencyMs: int64(i)})
	}
	got := r.drainResults()
	if len(got) != probeResultCap {
		t.Fatalf("buffer %d", len(got))
	}
	if got[0].LatencyMs != 10 || got[len(got)-1].LatencyMs != int64(probeResultCap+9) {
		t.Fatalf("kept range %d..%d", got[0].LatencyMs, got[len(got)-1].LatencyMs)
	}
	if r.dropped != 10 {
		t.Fatalf("dropped %d", r.dropped)
	}
}

// The wire shapes: a result marshals as taskId/latencyMs, and the
// config decode distinguishes an absent probe_tasks (old master,
// keep current set) from an empty one (clear all).
func TestProbeWireShapes(t *testing.T) {
	body, err := json.Marshal(ProbeResult{TaskID: 3, LatencyMs: -1})
	if err != nil || string(body) != `{"taskId":3,"latencyMs":-1}` {
		t.Fatalf("result wire: %s %v", body, err)
	}
	decode := func(s string) *[]ProbeTask {
		var result struct {
			ProbeTasks *[]ProbeTask `json:"probe_tasks"`
		}
		if err := json.Unmarshal([]byte(s), &result); err != nil {
			t.Fatal(err)
		}
		return result.ProbeTasks
	}
	if got := decode(`{"config_hash":"x"}`); got != nil {
		t.Fatalf("absent field decoded as %v", got)
	}
	if got := decode(`{"probe_tasks":[]}`); got == nil || len(*got) != 0 {
		t.Fatalf("empty list decoded as %v", got)
	}
	got := decode(`{"probe_tasks":[{"id":2,"target":"1.1.1.1:443","interval":30}]}`)
	if got == nil || len(*got) != 1 || (*got)[0].ID != 2 || (*got)[0].Target != "1.1.1.1:443" || (*got)[0].Interval != 30 {
		t.Fatalf("task decode %+v", got)
	}
}