EvoBGP/internal/scheduler/run_test.go

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")
		}
	}
}