package scheduler import ( "context" "encoding/json" "fmt" "io" "net/http" "strings" "testing" "time" "evobgp/internal/jobs" "evobgp/internal/pipeline" "evobgp/internal/store" ) type schedulerStore struct { store.Backend tenants []string modules map[string][]*store.Module enumerations int reads []string } func (s *schedulerStore) ListTenantIDs() ([]string, error) { s.enumerations++; return s.tenants, nil } func (s *schedulerStore) ListModules(tenant string) []*store.Module { s.reads = append(s.reads, tenant) return s.modules[tenant] } type schedulerRoundTrip func(*http.Request) (*http.Response, error) func (f schedulerRoundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } func schedulerFixture(t *testing.T) (*schedulerStore, time.Time) { t.Helper() now := time.Date(2026, 10, 26, 12, 0, 0, 0, time.UTC) st := &schedulerStore{Backend: store.NewMemory(), tenants: []string{"a", "b", store.SystemJobTenantID}, modules: map[string][]*store.Module{}} for _, tenant := range st.tenants { for i := 0; i < 1000; i++ { mod := &store.Module{ID: fmt.Sprintf("%s-%d", tenant, i), TenantID: tenant, Type: "DOMAINS", Enabled: true, RefreshIntervalSec: 60} if pipeline.ModuleDueForScheduler(mod, now) { st.modules[tenant] = []*store.Module{mod} break } } if len(st.modules[tenant]) == 0 { t.Fatal("fixture has no due module") } } return st, now } func TestSchedulerHTTPExplicitTenantScope(t *testing.T) { st, now := schedulerFixture(t) requests := 0 client := &http.Client{Transport: schedulerRoundTrip(func(r *http.Request) (*http.Response, error) { requests++ var body struct { ModuleIDs []string `json:"module_ids"` } if err := json.NewDecoder(r.Body).Decode(&body); err != nil { t.Fatal(err) } if len(body.ModuleIDs) != 1 || body.ModuleIDs[0] != st.modules["a"][0].ID { t.Fatalf("cross-tenant payload: %+v", body) } if r.Header.Get("Authorization") != "Bearer scoped" { t.Fatal("missing scoped bearer") } if _, ok := r.Context().Deadline(); !ok { t.Fatal("request has no deadline") } return &http.Response{StatusCode: http.StatusAccepted, Body: io.NopCloser(strings.NewReader("{}")), Header: make(http.Header)}, nil })} deps := &Deps{Store: st, APIBase: "https://cp.test", APIToken: "scoped", APITenantID: "a", HTTP: client} tickAt(context.Background(), deps, now) if requests != 1 || st.enumerations != 0 || len(st.reads) != 1 || st.reads[0] != "a" { t.Fatalf("scope requests=%d enumerations=%d reads=%v", requests, st.enumerations, st.reads) } deps.APITenantID = "" tickAt(context.Background(), deps, now) if requests != 1 || len(st.reads) != 1 { t.Fatal("unscoped HTTP scheduler performed work") } } func TestSchedulerTrustedQueueSeparatesTenantsAndHonorsCancellation(t *testing.T) { st, now := schedulerFixture(t) reg := jobs.NewRegistry(nil) reg.UseDurableQueue(st) deps := &Deps{Store: st, Jobs: reg} ctx, cancel := context.WithCancel(context.Background()) cancel() tickAt(ctx, deps, now) if st.enumerations != 0 { t.Fatal("cancelled scheduler accessed store") } tickAt(context.Background(), deps, now) tickAt(context.Background(), deps, now) for _, tenant := range []string{"a", "b"} { rows, _, _, err := st.ListJobs(context.Background(), tenant, "", "", "", 10) if err != nil || len(rows) != 1 || rows[0].TenantID != tenant { t.Fatalf("queue isolation/idempotency: %s %+v %v", tenant, rows, err) } raw, _ := json.Marshal(rows[0].Meta) if !strings.Contains(string(raw), st.modules[tenant][0].ID) { t.Fatalf("wrong tenant module: %s", raw) } } for _, read := range st.reads { if read == store.SystemJobTenantID { t.Fatal("scheduler read system tenant") } } }