EvoBGP/internal/pipeline/opt_collect_test.go

package pipeline

import (
	"context"
	"fmt"
	"net/http"
	"net/http/httptest"
	"sync"
	"testing"
	"time"

	"evobgp/internal/store"
)

func TestModuleDueForScheduler_JitterSpreads(t *testing.T) {
	dueTicks := map[int64]int{}
	for i := 0; i < 100; i++ {
		mod := &store.Module{
			ID:                 fmt.Sprintf("module-%d", i),
			Enabled:            true,
			Type:               "AS_PREFIXES",
			RefreshIntervalSec: 3600,
		}
		for tick := int64(0); tick < 7200; tick += SchedulerTickSec {
			if ModuleDueForScheduler(mod, time.Unix(tick, 0)) {
				dueTicks[tick]++
			}
		}
	}
	if len(dueTicks) < 10 {
		t.Fatalf("expected due events across many ticks, got %d buckets", len(dueTicks))
	}
	for tick, n := range dueTicks {
		if n == 100 {
			t.Fatalf("all 100 modules due on tick %d", tick)
		}
	}
}

func TestCollectASPrefixRows_KeepsSamePrefixDifferentCommunity(t *testing.T) {
	st := store.NewMemory()
	st.SeedDemo()
	tenant, _, _, _, _ := st.DemoIDs()
	mod, err := st.CreateModule(tenant, &store.Module{Type: "AS_PREFIXES", Name: "as-dedup", Enabled: true})
	if err != nil {
		t.Fatal(err)
	}
	c1, c2 := "comm-a", "comm-b"
	e1, err := st.CreateASEntry(tenant, mod.ID, &store.ASEntry{ASN: 64500, CommunityID: &c1})
	if err != nil {
		t.Fatal(err)
	}
	e2, err := st.CreateASEntry(tenant, mod.ID, &store.ASEntry{ASN: 64501, CommunityID: &c2})
	if err != nil {
		t.Fatal(err)
	}
	if err := st.SetASNPrefixCache(64500, "a", []string{"192.0.2.0/24"}); err != nil {
		t.Fatal(err)
	}
	if err := st.SetASNPrefixCache(64501, "b", []string{"192.0.2.0/24"}); err != nil {
		t.Fatal(err)
	}
	rows, err := collectASPrefixRows(context.Background(), st, http.DefaultClient, tenant, mod, []*store.ASEntry{e1, e2}, nil)
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 2 {
		t.Fatalf("expected 2 rows (same prefix, different community), got %d: %+v", len(rows), rows)
	}
}

func TestCollectCDNPrefixRows_PartialSkipKeepsPrior(t *testing.T) {
	t.Setenv("EVOBGP_CDN_PARTIAL_OK", "1")
	t.Setenv("EVOBGP_STALE_ON_UPSTREAM_ERROR", "0")
	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "1")

	good := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte("198.51.100.0/24\n"))
	}))
	defer good.Close()
	bad := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		http.Error(w, "down", http.StatusServiceUnavailable)
	}))
	defer bad.Close()

	st := store.NewMemory()
	st.SeedDemo()
	tenant, _, _, _, _ := st.DemoIDs()
	mod, err := st.CreateModule(tenant, &store.Module{Type: "CDN_CIDRS", Name: "cdn-partial", Enabled: true})
	if err != nil {
		t.Fatal(err)
	}
	prior := []store.PrefixRow{
		{Prefix: "203.0.113.0/24", Source: "cdn:bad"},
		{Prefix: "1.2.3.0/24", Source: "cdn:good"},
	}
	sources := []*store.CDNSource{
		{ID: "good", URL: good.URL, SourceKind: "plain"},
		{ID: "bad", URL: bad.URL, SourceKind: "plain"},
	}
	rows, err := collectCDNPrefixRows(context.Background(), st, good.Client(), tenant, mod, sources, prior)
	if err != nil {
		t.Fatalf("partial skip should succeed: %v", err)
	}
	got := map[string]string{}
	for _, r := range rows {
		got[r.Source] = r.Prefix
	}
	if got["cdn:bad"] != "203.0.113.0/24" {
		t.Fatalf("skipped source lost prior row: %+v", rows)
	}
	if got["cdn:good"] != "198.51.100.0/24" {
		t.Fatalf("fetched source missing new row: %+v", rows)
	}
}

func TestMergeCDNSource_ParallelNoLostUpdate(t *testing.T) {
	st := store.NewMemory()
	st.SeedDemo()
	tenant, _, _, _, _ := st.DemoIDs()
	mod, err := st.CreateModule(tenant, &store.Module{Type: "CDN_CIDRS", Name: "cdn-lock", Enabled: true})
	if err != nil {
		t.Fatal(err)
	}
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		_ = mergeCDNSourceIntoModuleSnapshot(st, tenant, mod, "s1", []store.PrefixRow{
			{Prefix: "1.0.0.0/24", Source: "cdn:s1"},
		})
	}()
	go func() {
		defer wg.Done()
		_ = mergeCDNSourceIntoModuleSnapshot(st, tenant, mod, "s2", []store.PrefixRow{
			{Prefix: "2.0.0.0/24", Source: "cdn:s2"},
		})
	}()
	wg.Wait()
	snap, ok, err := st.GetModulePrefixSnapshot(tenant, mod.ID)
	if err != nil || !ok || snap == nil {
		t.Fatalf("snapshot missing: ok=%v err=%v", ok, err)
	}
	got := map[string]bool{}
	for _, r := range snap.Prefixes {
		got[r.Prefix] = true
	}
	if !got["1.0.0.0/24"] || !got["2.0.0.0/24"] {
		t.Fatalf("lost parallel merge update: %+v", snap.Prefixes)
	}
}

func TestResolveDomainIPsCached_UsesTTL(t *testing.T) {
	t.Setenv("EVOBGP_DOMAIN_CACHE_TTL_SEC", "300")
	st := store.NewMemory()
	if err := st.SetDomainResolveCache("cached.test", []string{"192.0.2.9"}); err != nil {
		t.Fatal(err)
	}
	addrs, err := resolveDomainIPsCached(context.Background(), st, nil, nil, "", "cached.test")
	if err != nil {
		t.Fatal(err)
	}
	if len(addrs) != 1 || addrs[0].String() != "192.0.2.9" {
		t.Fatalf("expected cached addr, got %v", addrs)
	}
}

func TestInputHashInvalidatedOnEntryCRUD(t *testing.T) {
	st := store.NewMemory()
	st.SeedDemo()
	tenant, _, _, _, _ := st.DemoIDs()
	mod, err := st.CreateModule(tenant, &store.Module{Type: "IP_RANGES", Name: "ip-hash", Enabled: true})
	if err != nil {
		t.Fatal(err)
	}
	if err := st.SetModuleInputHash(tenant, mod.ID, "pre"); err != nil {
		t.Fatal(err)
	}
	if _, err := st.CreateIPRangeEntry(tenant, mod.ID, &store.IPRangeEntry{Prefix: "10.0.0.0/8"}); err != nil {
		t.Fatal(err)
	}
	got, err := st.GetModule(tenant, mod.ID)
	if err != nil {
		t.Fatal(err)
	}
	if got.InputHash != "" {
		t.Fatalf("expected hash cleared after CRUD, got %q", got.InputHash)
	}
}