diff --git a/AGENTS.md b/AGENTS.md index 43a16bd..ed395ea 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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` diff --git a/cmd/netstable/main.go b/cmd/netstable/main.go index d6d5211..a7667cc 100644 --- a/cmd/netstable/main.go +++ b/cmd/netstable/main.go @@ -72,7 +72,6 @@ func runServer(args []string) { options := appweb.Options{ Store: store, - Runner: appweb.DefaultRunner, ClientResolver: resolver, TrustProxyHeaders: *trustProxyHeaders, BandwidthLimitMbps: *bandwidthLimitMbps, diff --git a/internal/client/client_test.go b/internal/client/client_test.go index cabace3..877750c 100644 --- a/internal/client/client_test.go +++ b/internal/client/client_test.go @@ -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") diff --git a/internal/probe/probe.go b/internal/probe/probe.go index 6cf9b6f..be99791 100644 --- a/internal/probe/probe.go +++ b/internal/probe/probe.go @@ -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) diff --git a/internal/probe/probe_test.go b/internal/probe/probe_test.go index 2650217..ed6b4d6 100644 --- a/internal/probe/probe_test.go +++ b/internal/probe/probe_test.go @@ -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 } diff --git a/internal/web/runner.go b/internal/web/runner.go deleted file mode 100644 index 6c361a5..0000000 --- a/internal/web/runner.go +++ /dev/null @@ -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}) - }) -} diff --git a/internal/web/server.go b/internal/web/server.go index 1fe9b9f..090eda1 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -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 { diff --git a/internal/web/server_test.go b/internal/web/server_test.go index a4ab2ea..2716aba 100644 --- a/internal/web/server_test.go +++ b/internal/web/server_test.go @@ -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("dashboard")}, }) - 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, diff --git a/web/static/app.js b/web/static/app.js index 090b660..f3a47fb 100644 --- a/web/static/app.js +++ b/web/static/app.js @@ -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(); }