package httpapi import ( "context" "errors" "net/http" "net/http/httptest" "strings" "sync/atomic" "testing" "time" "evobgp/internal/nodedispatch" "evobgp/internal/store" ) type telemetryHTTPStore struct { store.Backend discoveryWrites int snapshotErr error refreshCalls int } func (s *telemetryHTTPStore) UpsertPeerDiscovery(tenant string, in *store.PeerDiscoveryUpsert) (*store.BGPPeerDiscovery, error) { s.discoveryWrites++ return s.Backend.UpsertPeerDiscovery(tenant, in) } func (s *telemetryHTTPStore) ListTelemetry(ctx context.Context, tenant string) ([]store.TelemetrySnapshot, error) { if s.snapshotErr != nil { return nil, s.snapshotErr } return s.Backend.ListTelemetry(ctx, tenant) } func (s *telemetryHTTPStore) RequestTelemetryRefresh(ctx context.Context, tenant string, cooldown time.Duration) (bool, error) { s.refreshCalls++ return s.Backend.RequestTelemetryRefresh(ctx, tenant, cooldown) } func telemetryRequest(method, path, tenant string) *http.Request { r := httptest.NewRequest(method, path, nil) return r.WithContext(context.WithValue(r.Context(), authCtxKey, Auth{TenantID: tenant, Role: "viewer"})) } func TestTelemetryGETsAndRefreshNeverPollAgents(t *testing.T) { var calls atomic.Int64 agent := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls.Add(1) _, _ = w.Write([]byte(`{"ok":true,"sessions":[]}`)) })) defer agent.Close() t.Setenv("EVOBGP_NODE_DISPATCH_INSECURE_TLS", "1") mem := store.NewMemory() mem.SeedDemo() tenant, _, _, _, _ := mem.DemoIDs() sp, err := mem.CreateSpeaker(tenant, &store.Speaker{Role: "replica", MetaJSON: store.SpeakerMetaJSON(store.SpeakerMeta{AgentDomain: strings.TrimPrefix(agent.URL, "https://"), AgentSecret: "test"})}) if err != nil { t.Fatal(err) } // Positive control: the legacy adapter reaches the fake configured agent. if result := nodedispatch.FetchAgentHealth(context.Background(), sp, nodedispatch.Options{HTTPClient: agent.Client()}); !result.OK { t.Fatal(result) } calls.Store(0) st := &telemetryHTTPStore{Backend: mem} server := &Server{store: st} claims, err := mem.ClaimTelemetry(context.Background(), "collector", 1, time.Minute, "") if err != nil || len(claims) != 1 { t.Fatalf("claim: %v %v", claims, err) } for i := 0; i < 20; i++ { for _, tc := range []struct { path string handler http.HandlerFunc }{ {"/v1/peers", server.handleListPeers}, {"/v1/peers?live=1", server.handleListPeers}, {"/v1/speakers?live=1", server.handleListSpeakers}, {"/v1/peers/discovered?status=all", server.handleListPeerDiscoveries}, } { w := httptest.NewRecorder() tc.handler(w, telemetryRequest(http.MethodGet, tc.path, tenant)) if w.Code != http.StatusOK { t.Fatalf("%s: %d %s", tc.path, w.Code, w.Body.String()) } } w := httptest.NewRecorder() server.handleTelemetryRefresh(w, telemetryRequest(http.MethodPost, "/v1/telemetry/refresh", tenant)) if w.Code != http.StatusAccepted { t.Fatalf("refresh: %d %s", w.Code, w.Body.String()) } want := `"requested":false` if i == 0 { want = `"requested":true` } if !strings.Contains(w.Body.String(), want) { t.Fatalf("refresh coalescing: %s", w.Body.String()) } } if calls.Load() != 0 || st.discoveryWrites != 0 { t.Fatalf("GET/refresh caused IO: agent=%d discovery=%d", calls.Load(), st.discoveryWrites) } } func TestTelemetryReadFailureAndRefreshPermission(t *testing.T) { mem := store.NewMemory() mem.SeedDemo() tenant, _, _, _, _ := mem.DemoIDs() st := &telemetryHTTPStore{Backend: mem, snapshotErr: errors.New("private database failure")} server := &Server{store: st} for _, handler := range []http.HandlerFunc{server.handleListPeers, server.handleListSpeakers} { w := httptest.NewRecorder() handler(w, telemetryRequest(http.MethodGet, "/v1/resource?live=1", tenant)) if w.Code != http.StatusServiceUnavailable || strings.Contains(w.Body.String(), "private") { t.Fatalf("snapshot failure: %d %s", w.Code, w.Body.String()) } } r := httptest.NewRequest(http.MethodPost, "/v1/telemetry/refresh", nil) w := httptest.NewRecorder() server.handleTelemetryRefresh(w, r) if w.Code != http.StatusUnauthorized || st.refreshCalls != 0 { t.Fatal("unauthenticated refresh admitted") } r = r.WithContext(context.WithValue(r.Context(), authCtxKey, Auth{TenantID: tenant, Kind: AuthKindJWT})) w = httptest.NewRecorder() server.handleTelemetryRefresh(w, r) if w.Code != http.StatusForbidden || st.refreshCalls != 0 { t.Fatal("unprivileged refresh admitted") } }