refactor: remove legacy server-driven SSE test path
ci / test (push) Successful in 10s
ci / release-artifacts (push) Skipped

The browser and CLI flows both use download/upload/complete; the
/api/tests/{id}/events endpoint and its Runner plumbing were dead code
that could bypass the single-active-test lock via expireActiveLocked
promoting the queue while an SSE stream was still running.

Remove the endpoint, Runner option, DefaultRunner, probe.Run engine and
the frontend subscribe/closeSource leftovers. Add a regression test
asserting the endpoint returns 404 and create responses omit eventsUrl.
This commit is contained in:
Test User
2026-09-22 14:25:30 +08:00
parent 2575c932c5
commit 4d8a295504
9 changed files with 42 additions and 595 deletions
+2 -3
View File
@@ -60,9 +60,7 @@ git ls-remote origin refs/heads/main
- `internal/web/server.go`
- HTTP API、网页静态资源、队列、全局锁、记录完成、下载/上传接口、iperf3 会话。
- 这是生产行为最关键的文件,修改前必须读相关测试。
- `internal/web/runner.go`
- 服务端 SSE/探测 runner。
- 测试 SSE 探测端点(`/api/tests/{id}/events`)已移除;浏览器和 CLI 测速统一走 download/upload/complete,页面只保留 `/api/health/events` 心跳。
- `internal/client/client.go`
- HTTP CLI 客户端。
@@ -207,6 +205,7 @@ make update-ip-db
- `TestRunIperf3CompletesOneOffServerSession`
- `TestCancelActiveTestPromotesQueuedSession`
- `TestCancelQueuedTestRemovesItAndUpdatesPositions`
- `TestEventsEndpointIsRemoved`
- `TestBrowserSummaryDoesNotTreatShortBurstsAsSustainedMax`
- `TestStoreListsLargeRecordsWithFullSampleCurves`
-1
View File
@@ -72,7 +72,6 @@ func runServer(args []string) {
options := appweb.Options{
Store: store,
Runner: appweb.DefaultRunner,
ClientResolver: resolver,
TrustProxyHeaders: *trustProxyHeaders,
BandwidthLimitMbps: *bandwidthLimitMbps,
+6 -10
View File
@@ -21,7 +21,7 @@ import (
func TestRunCompletesBrowserCompatibleSpeedTest(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner})
server := appweb.New(appweb.Options{Store: store})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
@@ -53,7 +53,7 @@ func TestRunCompletesBrowserCompatibleSpeedTest(t *testing.T) {
func TestRunWaitsInQueueUntilBusySessionCompletes(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner})
server := appweb.New(appweb.Options{Store: store})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
@@ -86,7 +86,7 @@ func TestRunWaitsInQueueUntilBusySessionCompletes(t *testing.T) {
func TestRunCancelsQueuedSessionWhenContextStops(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner})
server := appweb.New(appweb.Options{Store: store})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
@@ -133,7 +133,7 @@ func TestRunCancelsQueuedSessionWhenContextStops(t *testing.T) {
func TestRunPersistsPartialRecordWhenCanceledAfterSamples(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner})
server := appweb.New(appweb.Options{Store: store})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
@@ -171,7 +171,7 @@ func TestRunPersistsPartialRecordWhenCanceledAfterSamples(t *testing.T) {
func TestRunIperf3CompletesOneOffServerSession(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner, Iperf3Path: fakeIperf3(t)})
server := appweb.New(appweb.Options{Store: store, Iperf3Path: fakeIperf3(t)})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
@@ -195,7 +195,7 @@ func TestRunIperf3CompletesOneOffServerSession(t *testing.T) {
}
func TestRunIperf3FailedCommandReleasesOneOffServerSession(t *testing.T) {
server := appweb.New(appweb.Options{Store: &memoryStore{}, Runner: noopRunner, Iperf3Path: fakeIperf3(t)})
server := appweb.New(appweb.Options{Store: &memoryStore{}, Iperf3Path: fakeIperf3(t)})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
@@ -285,10 +285,6 @@ func (s *memoryStore) snapshot() []recordstore.Record {
return out
}
func noopRunner(ctx context.Context, request appweb.TestRequest, emit func(appweb.Event) error) (probe.Summary, error) {
return probe.Summary{Target: request.Target.Address, StartedAt: time.Now(), FinishedAt: time.Now()}, nil
}
func fakeIperf3(t *testing.T) string {
t.Helper()
path := filepath.Join(t.TempDir(), "iperf3")
-130
View File
@@ -1,13 +1,10 @@
package probe
import (
"context"
"errors"
"fmt"
"io"
"math"
"net"
"net/http"
"net/url"
"strconv"
"strings"
@@ -54,14 +51,6 @@ type Summary struct {
Score float64 `json:"score"`
}
type Config struct {
Target Target
HTTPURL string
Duration time.Duration
Interval time.Duration
Timeout time.Duration
}
func ParseTarget(input string) (Target, error) {
raw := strings.TrimSpace(input)
if raw == "" {
@@ -84,60 +73,6 @@ func ParseTarget(input string) (Target, error) {
}, nil
}
func Run(ctx context.Context, config Config, emit func(Sample) error) (Summary, error) {
config = normalizeConfig(config)
startedAt := time.Now().UTC()
deadline := startedAt.Add(config.Duration)
var samples []Sample
for {
if err := ctx.Err(); err != nil {
return Summarize(config.Target.Address, startedAt, time.Now().UTC(), samples), err
}
tcpSample := probeTCP(ctx, config.Target, config.Timeout)
samples = append(samples, tcpSample)
if emit != nil {
if err := emit(tcpSample); err != nil {
return Summarize(config.Target.Address, startedAt, time.Now().UTC(), samples), err
}
}
if config.HTTPURL != "" {
httpSample := probeHTTP(ctx, config.HTTPURL, config.Timeout)
samples = append(samples, httpSample)
if emit != nil {
if err := emit(httpSample); err != nil {
return Summarize(config.Target.Address, startedAt, time.Now().UTC(), samples), err
}
}
}
if time.Now().UTC().After(deadline) {
break
}
wait := config.Interval
remaining := time.Until(deadline)
if remaining < wait {
wait = remaining
}
if wait <= 0 {
break
}
select {
case <-ctx.Done():
return Summarize(config.Target.Address, startedAt, time.Now().UTC(), samples), ctx.Err()
case <-time.After(wait):
}
}
finishedAt := time.Now().UTC()
summary := Summarize(config.Target.Address, startedAt, finishedAt, samples)
summary.HTTPURL = config.HTTPURL
return summary, nil
}
func Summarize(target string, startedAt, finishedAt time.Time, samples []Sample) Summary {
summary := Summary{
Target: target,
@@ -195,71 +130,6 @@ func Summarize(target string, startedAt, finishedAt time.Time, samples []Sample)
return summary
}
func normalizeConfig(config Config) Config {
if config.Duration <= 0 {
config.Duration = 30 * time.Second
}
if config.Interval <= 0 {
config.Interval = time.Second
}
if config.Timeout <= 0 {
config.Timeout = 3 * time.Second
}
return config
}
func probeTCP(ctx context.Context, target Target, timeout time.Duration) Sample {
start := time.Now()
dialer := net.Dialer{Timeout: timeout}
conn, err := dialer.DialContext(ctx, "tcp", target.Address)
latency := time.Since(start)
sample := newSample("tcp", latency)
if err != nil {
sample.Success = false
sample.Error = err.Error()
return sample
}
_ = conn.Close()
sample.Success = true
return sample
}
func probeHTTP(ctx context.Context, httpURL string, timeout time.Duration) Sample {
start := time.Now()
request, err := http.NewRequestWithContext(ctx, http.MethodGet, httpURL, nil)
if err != nil {
sample := newSample("http", 0)
sample.Error = err.Error()
return sample
}
client := http.Client{Timeout: timeout}
response, err := client.Do(request)
latency := time.Since(start)
sample := newSample("http", latency)
if err != nil {
sample.Error = err.Error()
return sample
}
defer response.Body.Close()
_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 1024))
if response.StatusCode >= 500 {
sample.Error = response.Status
return sample
}
sample.Success = true
return sample
}
func newSample(kind string, latency time.Duration) Sample {
return Sample{
At: time.Now().UTC(),
Kind: kind,
Latency: latency,
LatencyMS: round2(float64(latency.Microseconds()) / 1000),
}
}
func parseHostPort(raw string) (string, int, error) {
if strings.Contains(raw, "://") {
parsed, err := url.Parse(raw)
-86
View File
@@ -1,11 +1,7 @@
package probe
import (
"context"
"math"
"net"
"net/http"
"net/http/httptest"
"testing"
"time"
)
@@ -103,88 +99,6 @@ func TestSummarizeHandlesNoSamples(t *testing.T) {
}
}
func TestRunEmitsSuccessfulTCPSamplesAgainstLocalServer(t *testing.T) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("Listen: %v", err)
}
defer listener.Close()
done := make(chan struct{})
go func() {
defer close(done)
for {
conn, err := listener.Accept()
if err != nil {
return
}
_ = conn.Close()
}
}()
target, err := ParseTarget(listener.Addr().String())
if err != nil {
t.Fatalf("ParseTarget: %v", err)
}
var emitted []Sample
summary, err := Run(context.Background(), Config{
Target: target,
Duration: 80 * time.Millisecond,
Interval: 20 * time.Millisecond,
Timeout: 100 * time.Millisecond,
}, func(sample Sample) error {
emitted = append(emitted, sample)
return nil
})
if err != nil {
t.Fatalf("Run returned error: %v", err)
}
if len(emitted) == 0 {
t.Fatal("no samples emitted")
}
if summary.Successes == 0 || summary.Failures != 0 {
t.Fatalf("summary successes/failures = %d/%d, want successes and no failures", summary.Successes, summary.Failures)
}
if emitted[0].Kind != "tcp" || !emitted[0].Success || emitted[0].LatencyMS <= 0 {
t.Fatalf("first sample = %#v, want successful tcp sample with latency", emitted[0])
}
}
func TestRunEmitsHTTPSamplesWhenHTTPURLConfigured(t *testing.T) {
httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNoContent)
}))
defer httpServer.Close()
target, err := ParseTarget(httpServer.URL)
if err != nil {
t.Fatalf("ParseTarget: %v", err)
}
var sawHTTP bool
summary, err := Run(context.Background(), Config{
Target: target,
HTTPURL: httpServer.URL,
Duration: 60 * time.Millisecond,
Interval: 20 * time.Millisecond,
Timeout: 100 * time.Millisecond,
}, func(sample Sample) error {
if sample.Kind == "http" && sample.Success && sample.LatencyMS > 0 {
sawHTTP = true
}
return nil
})
if err != nil {
t.Fatalf("Run returned error: %v", err)
}
if !sawHTTP {
t.Fatal("did not emit a successful http sample")
}
if summary.Samples == 0 || summary.Successes == 0 {
t.Fatalf("summary = %#v, want successful samples", summary)
}
}
func nearlyEqual(got, want, tolerance float64) bool {
return math.Abs(got-want) <= tolerance
}
-19
View File
@@ -1,19 +0,0 @@
package web
import (
"context"
"speedtest/internal/probe"
)
func DefaultRunner(ctx context.Context, request TestRequest, emit func(Event) error) (probe.Summary, error) {
return probe.Run(ctx, probe.Config{
Target: request.Target,
HTTPURL: request.HTTPURL,
Duration: request.Duration,
Interval: request.Interval,
Timeout: request.Timeout,
}, func(sample probe.Sample) error {
return emit(Event{Type: "sample", Sample: &sample})
})
}
-146
View File
@@ -32,12 +32,10 @@ type RecordStore interface {
List(ctx context.Context, limit int) ([]recordstore.Record, error)
}
type Runner func(ctx context.Context, request TestRequest, emit func(Event) error) (probe.Summary, error)
type ClientResolver func(ctx context.Context, ip string) (recordstore.ClientInfo, error)
type Options struct {
Store RecordStore
Runner Runner
ClientResolver ClientResolver
TrustProxyHeaders bool
BandwidthLimitMbps float64
@@ -49,7 +47,6 @@ type Options struct {
type Server struct {
store RecordStore
runner Runner
clientResolver ClientResolver
trustProxyHeaders bool
bandwidthLimiter limit.Limiter
@@ -79,13 +76,6 @@ type TestRequest struct {
TimeoutMillis int
}
type Event struct {
Type string `json:"type"`
Sample *probe.Sample `json:"sample,omitempty"`
Summary *probe.Summary `json:"summary,omitempty"`
Error string `json:"error,omitempty"`
}
type recordsResponse struct {
Records []recordstore.Record `json:"records"`
}
@@ -131,7 +121,6 @@ type createTestResponse struct {
Status string `json:"status"`
QueuePosition int `json:"queuePosition"`
QueueURL string `json:"queueUrl"`
EventsURL string `json:"eventsUrl"`
DownloadURL string `json:"downloadUrl"`
UploadURL string `json:"uploadUrl"`
CompleteURL string `json:"completeUrl"`
@@ -195,7 +184,6 @@ func New(options Options) *Server {
}
return &Server{
store: options.Store,
runner: options.Runner,
clientResolver: options.ClientResolver,
trustProxyHeaders: options.TrustProxyHeaders,
bandwidthLimiter: limit.New(options.BandwidthLimitMbps),
@@ -698,7 +686,6 @@ func (s *Server) handleCreateTest(w http.ResponseWriter, r *http.Request) {
Status: status,
QueuePosition: position,
QueueURL: "/api/tests/" + request.ID + "/queue",
EventsURL: "/api/tests/" + request.ID + "/events",
DownloadURL: "/api/download?testId=" + request.ID,
UploadURL: "/api/upload?testId=" + request.ID,
CompleteURL: "/api/tests/" + request.ID + "/complete",
@@ -710,8 +697,6 @@ func (s *Server) handleTestRoute(w http.ResponseWriter, r *http.Request) {
switch {
case strings.HasSuffix(r.URL.Path, "/queue"):
s.handleQueueStatus(w, r)
case strings.HasSuffix(r.URL.Path, "/events"):
s.handleTestEvents(w, r)
case strings.HasSuffix(r.URL.Path, "/complete"):
s.handleCompleteTest(w, r)
case strings.HasSuffix(r.URL.Path, "/cancel"):
@@ -803,94 +788,6 @@ func (s *Server) handleCancelTest(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "canceled"})
}
func (s *Server) handleTestEvents(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
return
}
id, ok := eventSessionID(r.URL.Path)
if !ok {
writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
return
}
s.mu.Lock()
_, exists := s.sessions[id]
active := s.activeID == id
s.mu.Unlock()
if !exists {
writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
return
}
if !active {
writeJSON(w, http.StatusConflict, errorResponse{Error: "该测速会话正在排队,请等待轮到后再开始。"})
return
}
s.mu.Lock()
request, exists := s.sessions[id]
delete(s.sessions, id)
s.mu.Unlock()
if !exists {
writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
return
}
w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("X-Accel-Buffering", "no")
flusher, _ := w.(http.Flusher)
var samples []probe.Sample
emit := func(event Event) error {
if event.Sample != nil {
sample := *event.Sample
if sample.LatencyMS == 0 && sample.Latency > 0 {
sample.LatencyMS = float64(sample.Latency.Microseconds()) / 1000
}
samples = append(samples, sample)
event.Sample = &sample
}
if err := writeSSE(w, event); err != nil {
return err
}
if flusher != nil {
flusher.Flush()
}
return nil
}
if s.runner == nil {
_ = emit(Event{Type: "error", Error: "test runner is not configured"})
s.releaseActive(request.ID)
return
}
defer s.releaseActive(request.ID)
summary, err := s.runner(r.Context(), request, emit)
if err != nil {
_ = emit(Event{Type: "error", Error: err.Error()})
return
}
summary.ID = request.ID
summary.HTTPURL = request.HTTPURL
if s.store != nil {
if err := s.store.Append(r.Context(), recordstore.Record{
ID: request.ID,
Client: request.Client,
Summary: summary,
Samples: samples,
CreatedAt: request.CreatedAt,
}); err != nil {
_ = emit(Event{Type: "error", Error: "failed to persist test record"})
return
}
}
_ = emit(Event{Type: "summary", Summary: &summary})
}
func normalizeTestRequest(raw rawTestRequest) (TestRequest, error) {
target, err := probe.ParseTarget(raw.Target)
if err != nil {
@@ -941,15 +838,6 @@ func normalizeTestRequest(raw rawTestRequest) (TestRequest, error) {
}, nil
}
func (s *Server) releaseActive(id string) {
s.mu.Lock()
defer s.mu.Unlock()
if s.activeID == id {
s.activeID = ""
s.promoteNextLocked(time.Now().UTC())
}
}
func (s *Server) takeSession(id string) (TestRequest, bool) {
s.mu.Lock()
defer s.mu.Unlock()
@@ -1210,15 +1098,6 @@ func clientIPFromRequest(r *http.Request, trustProxyHeaders bool) string {
return strings.TrimSpace(r.RemoteAddr)
}
func eventSessionID(path string) (string, bool) {
if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/events") {
return "", false
}
id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/tests/"), "/events")
id = strings.Trim(id, "/")
return id, id != ""
}
func completeSessionID(path string) (string, bool) {
if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/complete") {
return "", false
@@ -1551,31 +1430,6 @@ func round2(value float64) float64 {
return math.Round(value*100) / 100
}
func writeSSE(w http.ResponseWriter, event Event) error {
eventType := event.Type
if eventType == "" {
eventType = "message"
}
var data any
switch {
case event.Sample != nil:
data = event.Sample
case event.Summary != nil:
data = event.Summary
case event.Error != "":
data = errorResponse{Error: event.Error}
default:
data = event
}
payload, err := json.Marshal(data)
if err != nil {
return err
}
return writeSSEPayload(w, eventType, payload)
}
func writeSSEData(w http.ResponseWriter, eventType string, data any) error {
payload, err := json.Marshal(data)
if err != nil {
+34 -161
View File
@@ -6,7 +6,6 @@ import (
"encoding/json"
"errors"
"io/fs"
"net"
"net/http"
"net/http/httptest"
"os"
@@ -38,7 +37,7 @@ func TestRecordsEndpointReturnsRecords(t *testing.T) {
},
},
}
server := New(Options{Store: store, Runner: noopRunner})
server := New(Options{Store: store})
req := httptest.NewRequest(http.MethodGet, "/api/records?limit=2", nil)
res := httptest.NewRecorder()
@@ -75,7 +74,7 @@ func TestRecordsEndpointAnonymizesClientIPAndFillsMissingISP(t *testing.T) {
},
},
}
server := New(Options{Store: store, Runner: noopRunner})
server := New(Options{Store: store})
req := httptest.NewRequest(http.MethodGet, "/api/records?limit=1", nil)
res := httptest.NewRecorder()
@@ -98,7 +97,7 @@ func TestRecordsEndpointAnonymizesClientIPAndFillsMissingISP(t *testing.T) {
}
func TestConfigEndpointReturnsBandwidthLimit(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, BandwidthLimitMbps: 80})
server := New(Options{Store: &memoryStore{}, BandwidthLimitMbps: 80})
req := httptest.NewRequest(http.MethodGet, "/api/config", nil)
res := httptest.NewRecorder()
@@ -120,7 +119,7 @@ func TestConfigEndpointReturnsBandwidthLimit(t *testing.T) {
}
func TestHealthEventsStreamsHeartbeat(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, BandwidthLimitMbps: 30})
server := New(Options{Store: &memoryStore{}, BandwidthLimitMbps: 30})
req := httptest.NewRequest(http.MethodGet, "/api/health/events?limit=1", nil)
res := httptest.NewRecorder()
@@ -139,7 +138,7 @@ func TestHealthEventsStreamsHeartbeat(t *testing.T) {
}
func TestDownloadEndpointStreamsRequestedBytesAndAdvertisesLimit(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, BandwidthLimitMbps: 80})
server := New(Options{Store: &memoryStore{}, BandwidthLimitMbps: 80})
createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
createRes := httptest.NewRecorder()
@@ -168,7 +167,7 @@ func TestDownloadEndpointStreamsRequestedBytesAndAdvertisesLimit(t *testing.T) {
}
func TestDownloadEndpointRequiresActiveTestID(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
req := httptest.NewRequest(http.MethodGet, "/api/download?bytes=1024", nil)
res := httptest.NewRecorder()
@@ -180,7 +179,7 @@ func TestDownloadEndpointRequiresActiveTestID(t *testing.T) {
}
func TestUploadEndpointAcceptsActiveSession(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, BandwidthLimitMbps: 80})
server := New(Options{Store: &memoryStore{}, BandwidthLimitMbps: 80})
createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
createRes := httptest.NewRecorder()
@@ -210,7 +209,7 @@ func TestUploadEndpointAcceptsActiveSession(t *testing.T) {
}
func TestUploadEndpointRequiresActiveTestID(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
req := httptest.NewRequest(http.MethodPost, "/api/upload?bytes=1024", bytes.NewBuffer(make([]byte, 1024)))
res := httptest.NewRecorder()
@@ -222,7 +221,7 @@ func TestUploadEndpointRequiresActiveTestID(t *testing.T) {
}
func TestDownloadEndpointRejectsInvalidByteCount(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
req := httptest.NewRequest(http.MethodGet, "/api/download?bytes=0", nil)
res := httptest.NewRecorder()
@@ -234,7 +233,7 @@ func TestDownloadEndpointRejectsInvalidByteCount(t *testing.T) {
}
func TestCreateTestRejectsInvalidTarget(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
req := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com:abc"}`))
res := httptest.NewRecorder()
@@ -262,8 +261,8 @@ func TestClientIPUsesRemoteAddrUnlessProxyHeadersAreTrusted(t *testing.T) {
}
}
func TestCreateTestReturnsSessionIDAndEventsURL(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, ClientResolver: staticResolver})
func TestCreateTestReturnsSessionIDAndRouteURLs(t *testing.T) {
server := New(Options{Store: &memoryStore{}, ClientResolver: staticResolver})
req := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1,"intervalMillis":200,"timeoutMillis":300}`))
req.RemoteAddr = "198.51.100.10:54321"
@@ -283,9 +282,6 @@ func TestCreateTestReturnsSessionIDAndEventsURL(t *testing.T) {
if payload.Status != "ready" || payload.QueuePosition != 0 || payload.QueueURL != "/api/tests/"+payload.ID+"/queue" {
t.Fatalf("queue fields = %q/%d/%q, want ready/0/queue URL", payload.Status, payload.QueuePosition, payload.QueueURL)
}
if payload.EventsURL != "/api/tests/"+payload.ID+"/events" {
t.Fatalf("EventsURL = %q, want /api/tests/%s/events", payload.EventsURL, payload.ID)
}
if payload.DownloadURL != "/api/download?testId="+payload.ID {
t.Fatalf("DownloadURL = %q, want /api/download?testId=%s", payload.DownloadURL, payload.ID)
}
@@ -301,7 +297,7 @@ func TestCreateTestReturnsSessionIDAndEventsURL(t *testing.T) {
}
func TestCreateTestQueuesWhenAnotherSessionIsActive(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
firstRes := httptest.NewRecorder()
@@ -326,7 +322,7 @@ func TestCreateTestQueuesWhenAnotherSessionIsActive(t *testing.T) {
}
func TestQueuedTestPromotesAfterActiveSessionCompletes(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
firstRes := httptest.NewRecorder()
@@ -397,7 +393,7 @@ func TestQueuedTestPromotesAfterActiveSessionCompletes(t *testing.T) {
}
func TestCancelActiveTestPromotesQueuedSession(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
firstRes := httptest.NewRecorder()
@@ -447,7 +443,7 @@ func TestCancelActiveTestPromotesQueuedSession(t *testing.T) {
}
func TestCancelQueuedTestRemovesItAndUpdatesPositions(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
firstRes := httptest.NewRecorder()
@@ -511,7 +507,7 @@ func TestCancelQueuedTestRemovesItAndUpdatesPositions(t *testing.T) {
}
func TestCreateIperf3SessionStartsOneOffServerAndUsesGlobalLock(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, Iperf3Path: fakeIperf3(t)})
server := New(Options{Store: &memoryStore{}, Iperf3Path: fakeIperf3(t)})
req := httptest.NewRequest(http.MethodPost, "/api/iperf3/sessions", jsonBody(`{"durationSeconds":1,"protocol":"tcp"}`))
req.RemoteAddr = "203.0.113.20:54321"
@@ -546,7 +542,7 @@ func TestCreateIperf3SessionStartsOneOffServerAndUsesGlobalLock(t *testing.T) {
func TestCompleteIperf3SessionPersistsRecordAndReleasesLock(t *testing.T) {
store := &memoryStore{}
server := New(Options{Store: store, Runner: noopRunner, Iperf3Path: fakeIperf3(t), ClientResolver: staticResolver})
server := New(Options{Store: store, Iperf3Path: fakeIperf3(t), ClientResolver: staticResolver})
createReq := httptest.NewRequest(http.MethodPost, "/api/iperf3/sessions", jsonBody(`{"durationSeconds":1,"protocol":"tcp","reverse":true}`))
createReq.RemoteAddr = "203.0.113.20:54321"
@@ -593,7 +589,7 @@ func TestCompleteIperf3SessionPersistsRecordAndReleasesLock(t *testing.T) {
}
func TestBrowserSessionQueuesDuringDownloadAndUploadWindows(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
firstRes := httptest.NewRecorder()
@@ -623,33 +619,10 @@ func TestBrowserSessionQueuesDuringDownloadAndUploadWindows(t *testing.T) {
}
}
func TestSSEStreamsSamplesSummaryAndPersistsRecord(t *testing.T) {
store := &memoryStore{}
start := time.Date(2026, 6, 5, 8, 0, 0, 0, time.UTC)
runner := func(ctx context.Context, request TestRequest, emit func(Event) error) (probe.Summary, error) {
sample := probe.Sample{
At: start,
Kind: "tcp",
Success: true,
Latency: 25 * time.Millisecond,
LatencyMS: 25,
}
if err := emit(Event{Type: "sample", Sample: &sample}); err != nil {
return probe.Summary{}, err
}
return probe.Summary{
Target: "example.com:80",
StartedAt: start,
FinishedAt: start.Add(time.Second),
Samples: 1,
Successes: 1,
Score: 98,
}, nil
}
server := New(Options{Store: store, Runner: runner, ClientResolver: staticResolver})
func TestEventsEndpointIsRemoved(t *testing.T) {
server := New(Options{Store: &memoryStore{}})
createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
createReq.RemoteAddr = "203.0.113.20:54321"
createRes := httptest.NewRecorder()
server.Handler().ServeHTTP(createRes, createReq)
if createRes.Code != http.StatusAccepted {
@@ -659,72 +632,21 @@ func TestSSEStreamsSamplesSummaryAndPersistsRecord(t *testing.T) {
if err := json.Unmarshal(createRes.Body.Bytes(), &created); err != nil {
t.Fatalf("decode create response: %v", err)
}
if strings.Contains(createRes.Body.String(), "eventsUrl") {
t.Fatalf("create response should not advertise an events URL: %s", createRes.Body.String())
}
eventsReq := httptest.NewRequest(http.MethodGet, created.EventsURL, nil)
eventsReq := httptest.NewRequest(http.MethodGet, "/api/tests/"+created.ID+"/events", nil)
eventsRes := httptest.NewRecorder()
server.Handler().ServeHTTP(eventsRes, eventsReq)
if eventsRes.Code != http.StatusOK {
t.Fatalf("events status = %d, want 200; body=%s", eventsRes.Code, eventsRes.Body.String())
}
if contentType := eventsRes.Header().Get("Content-Type"); !strings.HasPrefix(contentType, "text/event-stream") {
t.Fatalf("Content-Type = %q, want text/event-stream", contentType)
}
body := eventsRes.Body.String()
if !strings.Contains(body, "event: sample") || !strings.Contains(body, `"latencyMs":25`) {
t.Fatalf("events body missing sample event: %s", body)
}
if !strings.Contains(body, "event: summary") || !strings.Contains(body, `"score":98`) {
t.Fatalf("events body missing summary event: %s", body)
}
records := store.snapshot()
if len(records) != 1 {
t.Fatalf("persisted records = %d, want 1", len(records))
}
if records[0].ID != created.ID {
t.Fatalf("persisted id = %q, want %q", records[0].ID, created.ID)
}
if records[0].Client.IP != "203.0.113.*" || records[0].Client.Region != "Beijing" || records[0].Client.ISP != "Example Mobile" {
t.Fatalf("persisted client = %#v, want enriched client data", records[0].Client)
}
if len(records[0].Samples) != 1 || records[0].Samples[0].LatencyMS != 25 {
t.Fatalf("persisted samples = %#v, want sample curve", records[0].Samples)
}
}
func TestSessionLockReleasesAfterSSECompletes(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
firstRes := httptest.NewRecorder()
server.Handler().ServeHTTP(firstRes, firstReq)
if firstRes.Code != http.StatusAccepted {
t.Fatalf("first status = %d, want 202; body=%s", firstRes.Code, firstRes.Body.String())
}
var first createTestResponse
if err := json.Unmarshal(firstRes.Body.Bytes(), &first); err != nil {
t.Fatalf("decode first response: %v", err)
}
eventsReq := httptest.NewRequest(http.MethodGet, first.EventsURL, nil)
eventsRes := httptest.NewRecorder()
server.Handler().ServeHTTP(eventsRes, eventsReq)
if eventsRes.Code != http.StatusOK {
t.Fatalf("events status = %d, want 200; body=%s", eventsRes.Code, eventsRes.Body.String())
}
nextReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.org","durationSeconds":1}`))
nextRes := httptest.NewRecorder()
server.Handler().ServeHTTP(nextRes, nextReq)
if nextRes.Code != http.StatusAccepted {
t.Fatalf("next status = %d, want 202 after prior session completes; body=%s", nextRes.Code, nextRes.Body.String())
if eventsRes.Code != http.StatusNotFound {
t.Fatalf("events status = %d, want 404 now that the SSE endpoint is removed; body=%s", eventsRes.Code, eventsRes.Body.String())
}
}
func TestCompleteEndpointPersistsBrowserSpeedRecordAndReleasesLock(t *testing.T) {
store := &memoryStore{}
server := New(Options{Store: store, Runner: noopRunner, ClientResolver: staticResolver})
server := New(Options{Store: store, ClientResolver: staticResolver})
createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
createReq.RemoteAddr = "203.0.113.20:54321"
@@ -782,7 +704,7 @@ func TestCompleteEndpointPersistsBrowserSpeedRecordAndReleasesLock(t *testing.T)
func TestCompleteEndpointAcceptsLargeSamplePayload(t *testing.T) {
store := &memoryStore{}
server := New(Options{Store: store, Runner: noopRunner, ClientResolver: staticResolver})
server := New(Options{Store: store, ClientResolver: staticResolver})
createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
createReq.RemoteAddr = "203.0.113.20:54321"
@@ -810,7 +732,7 @@ func TestCompleteEndpointAcceptsLargeSamplePayload(t *testing.T) {
func TestCompleteEndpointCanRetryWhenPersistFails(t *testing.T) {
store := &flakyStore{fail: true}
server := New(Options{Store: store, Runner: noopRunner, ClientResolver: staticResolver})
server := New(Options{Store: store, ClientResolver: staticResolver})
createReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
createReq.RemoteAddr = "203.0.113.20:54321"
@@ -914,7 +836,7 @@ func TestBrowserSummaryDoesNotTreatShortBurstsAsSustainedMax(t *testing.T) {
}
func TestSSEUnknownSessionReturnsNotFound(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
server := New(Options{Store: &memoryStore{}})
req := httptest.NewRequest(http.MethodGet, "/api/tests/missing/events", nil)
res := httptest.NewRecorder()
@@ -929,7 +851,7 @@ func TestHandlerServesStaticFilesFromFS(t *testing.T) {
staticFS := http.FS(fstest.MapFS{
"index.html": &fstest.MapFile{Data: []byte("<html>dashboard</html>")},
})
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, StaticFS: staticFS})
server := New(Options{Store: &memoryStore{}, StaticFS: staticFS})
req := httptest.NewRequest(http.MethodGet, "/", nil)
res := httptest.NewRecorder()
@@ -949,7 +871,7 @@ func TestDownloadsEndpointServesPackagedLinuxBinary(t *testing.T) {
if err := os.WriteFile(packagePath, []byte("fake-package"), 0644); err != nil {
t.Fatalf("write fake package: %v", err)
}
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, DownloadsDir: downloadsDir})
server := New(Options{Store: &memoryStore{}, DownloadsDir: downloadsDir})
req := httptest.NewRequest(http.MethodGet, "/downloads/netstable-linux-amd64.tar.gz", nil)
res := httptest.NewRecorder()
@@ -967,7 +889,7 @@ func TestDownloadsEndpointServesPackagedLinuxBinary(t *testing.T) {
}
func TestDownloadsEndpointRejectsUnknownBinaryName(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner, DownloadsDir: t.TempDir()})
server := New(Options{Store: &memoryStore{}, DownloadsDir: t.TempDir()})
req := httptest.NewRequest(http.MethodGet, "/downloads/netstable-linux-riscv64", nil)
res := httptest.NewRecorder()
@@ -991,51 +913,6 @@ func TestSubFSHelperReturnsStaticRoot(t *testing.T) {
}
}
func TestDefaultRunnerEmitsSampleEvents(t *testing.T) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("Listen: %v", err)
}
defer listener.Close()
go func() {
for {
conn, err := listener.Accept()
if err != nil {
return
}
_ = conn.Close()
}
}()
target, err := probe.ParseTarget(listener.Addr().String())
if err != nil {
t.Fatalf("ParseTarget: %v", err)
}
var sampleEvents int
summary, err := DefaultRunner(context.Background(), TestRequest{
Target: target,
Duration: 80 * time.Millisecond,
Interval: 20 * time.Millisecond,
Timeout: 100 * time.Millisecond,
}, func(event Event) error {
if event.Type == "sample" && event.Sample != nil {
sampleEvents++
}
return nil
})
if err != nil {
t.Fatalf("DefaultRunner returned error: %v", err)
}
if sampleEvents == 0 {
t.Fatal("no sample events emitted")
}
if summary.Successes == 0 {
t.Fatalf("summary = %#v, want successes", summary)
}
}
func jsonBody(body string) *bytes.Buffer {
return bytes.NewBufferString(body)
}
@@ -1119,10 +996,6 @@ func (s *flakyStore) Append(ctx context.Context, record recordstore.Record) erro
return s.memoryStore.Append(ctx, record)
}
func noopRunner(ctx context.Context, request TestRequest, emit func(Event) error) (probe.Summary, error) {
return probe.Summary{Target: request.Target.Address, StartedAt: time.Now(), FinishedAt: time.Now()}, nil
}
func staticResolver(ctx context.Context, ip string) (recordstore.ClientInfo, error) {
return recordstore.ClientInfo{
IP: ip,
-39
View File
@@ -1,5 +1,4 @@
const state = {
source: null,
samples: [],
records: [],
clientCommand: "",
@@ -114,7 +113,6 @@ async function stopCurrentTest() {
if (!saved) {
await cancelCurrentSession();
}
closeSource();
setServerState("idle", "空闲");
setNotice(saved ? "测试已停止,已保存已完成样本。" : "测试已停止,队列已释放。");
setTestingControls(false);
@@ -194,7 +192,6 @@ async function completeCurrentSessionWithFetch(session, body, options = {}) {
async function startTest(event) {
event.preventDefault();
closeSource();
state.samples = [];
state.currentSession = null;
state.cancelRequested = false;
@@ -422,35 +419,6 @@ function recordSample(sample) {
drawLiveChart();
}
function subscribe(eventsUrl) {
state.source = new EventSource(eventsUrl);
state.source.addEventListener("sample", event => {
const sample = JSON.parse(event.data);
state.samples.push(sample);
appendLog(sample);
updateLiveMetrics();
drawLiveChart();
});
state.source.addEventListener("summary", event => {
const summary = JSON.parse(event.data);
updateSummary(summary);
setServerState("idle", "空闲");
setNotice("测试完成,记录已保存。");
els.startButton.disabled = false;
closeSource();
loadRecords();
});
state.source.addEventListener("error", event => {
if (event.data) {
const payload = JSON.parse(event.data);
setNotice(payload.error || "测试流发生错误", true);
}
setServerState("error", "错误");
els.startButton.disabled = false;
closeSource();
});
}
async function loadRecords() {
try {
const response = await fetch("/api/records?limit=200");
@@ -735,13 +703,6 @@ function sleep(ms, signal) {
});
}
function closeSource() {
if (state.source) {
state.source.close();
state.source = null;
}
}
function value(id) {
return document.getElementById(id).value.trim();
}