EvoBGP/internal/jobs/worker_test.go

package jobs

import (
	"testing"
	"time"

	"evobgp/internal/store"
)

func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) {
	t.Setenv("EVOBGP_ASN_RESOLVE", "0")
	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "") // skip bird binary path in deploy_apply
	t.Setenv("EVOBGP_JOB_MAX_CONCURRENT", "8")

	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, modIP, _, _ := m.DemoIDs()
	for _, mod := range m.ListModules(tenant) {
		if mod == nil || mod.ID == modIP {
			continue
		}
		disabled := false
		if _, err := m.UpdateModule(tenant, mod.ID, &store.ModulePatch{Enabled: &disabled}); err != nil {
			t.Fatal(err)
		}
	}

	mod2, err := m.CreateModule(tenant, &store.Module{Type: "IP_RANGES", Name: "extra-ip", Enabled: true, Priority: 30})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := m.CreateIPRangeEntry(tenant, modIP, &store.IPRangeEntry{Prefix: "10.0.0.0/24"}); err != nil {
		t.Fatal(err)
	}
	if _, err := m.CreateIPRangeEntry(tenant, mod2.ID, &store.IPRangeEntry{Prefix: "192.168.0.0/24"}); err != nil {
		t.Fatal(err)
	}

	// Hold workers until both jobs are enqueued so inflightRefresh=2 before either
	// finishModuleRefreshSuccess. Otherwise a fast ingest can finalize+deploy before
	// the second Enqueue — sequential refreshes correctly produce two deploy_apply jobs.
	start := make(chan struct{})
	wk := &Worker{Store: m}
	reg := NewRegistry(func(j *Job) {
		<-start
		wk.Process(j)
	})
	wk.Registry = reg

	mid1 := modIP
	mid2 := mod2.ID
	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid1, map[string]any{"module_id": modIP}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid2, map[string]any{"module_id": mod2.ID}); err != nil {
		t.Fatal(err)
	}
	close(start)

	waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, 2)

	deployJobs, _, _ := reg.List(tenant, "", KindDeployApply, "", 100)
	if len(deployJobs) != 1 {
		t.Fatalf("want exactly one deploy_apply job, got %d", len(deployJobs))
	}

	var deferred, withDeployID int
	refreshJobs, _, _ := reg.List(tenant, "", KindModuleRefresh, "", 100)
	for _, j := range refreshJobs {
		if j == nil {
			continue
		}
		snap := j.Snapshot()
		if st, _ := snap["status"].(string); st != StatusSucceeded {
			continue
		}
		meta, _ := snap["meta"].(map[string]any)
		if v, ok := meta["deploy_apply_deferred"].(bool); ok && v {
			deferred++
		}
		if _, ok := meta["deploy_apply_job_id"].(string); ok {
			withDeployID++
		}
	}
	if deferred != 1 || withDeployID != 1 {
		t.Fatalf("want one deferred and one with deploy_apply_job_id, got deferred=%d deploy_meta=%d", deferred, withDeployID)
	}
}

func waitSucceededModuleRefreshCount(t *testing.T, reg *Registry, tenant string, want int) {
	t.Helper()
	waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, want)
}

func waitSucceededJobsByKindCount(t *testing.T, reg *Registry, tenant, kind string, want int) {
	t.Helper()
	deadline := time.Now().Add(30 * time.Second)
	for time.Now().Before(deadline) {
		jobs, _, _ := reg.List(tenant, StatusSucceeded, kind, "", 100)
		if len(jobs) >= want {
			return
		}
		all, _, _ := reg.List(tenant, "", kind, "", 100)
		for _, j := range all {
			if j == nil {
				continue
			}
			snap := j.Snapshot()
			if st, _ := snap["status"].(string); st != StatusFailed {
				continue
			}
			t.Fatalf("%s job failed: %#v", kind, snap["error"])
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("timeout waiting for %s jobs", kind)
}

func TestPeerReconcile_RendersFromLatestRevisionAndQueuesDeploy(t *testing.T) {
	t.Setenv("EVOBGP_ASN_RESOLVE", "0")
	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "") // skip bird binary path in deploy_apply

	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, modIP, _, _ := m.DemoIDs()
	for _, mod := range m.ListModules(tenant) {
		if mod == nil || mod.ID == modIP {
			continue
		}
		disabled := false
		if _, err := m.UpdateModule(tenant, mod.ID, &store.ModulePatch{Enabled: &disabled}); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := m.CreateIPRangeEntry(tenant, modIP, &store.IPRangeEntry{Prefix: "10.10.0.0/24"}); err != nil {
		t.Fatal(err)
	}
	peer, err := m.CreatePeer(tenant, &store.BGPPeer{
		Name:      "peer-a",
		Neighbor:  "192.0.2.2",
		RemoteASN: 64512,
		Enabled:   true,
	})
	if err != nil {
		t.Fatal(err)
	}

	wk := &Worker{Store: m}
	reg := NewRegistry(wk.Process)
	wk.Registry = reg

	// Create baseline revision from enabled IP module.
	mid := modIP
	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid, map[string]any{"module_id": modIP}); err != nil {
		t.Fatal(err)
	}
	waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, 1)
	beforeRevs, _, _ := m.ListRevisions(tenant, "", "", 200)

	enabled := false
	if _, err := m.UpdatePeer(tenant, peer.ID, &store.PeerPatch{Enabled: &enabled}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := reg.Enqueue(tenant, KindPeerReconcile, nil, nil, map[string]any{"trigger": "peer_patch"}); err != nil {
		t.Fatal(err)
	}
	waitSucceededJobsByKindCount(t, reg, tenant, KindPeerReconcile, 1)

	afterRevs, _, _ := m.ListRevisions(tenant, "", "", 200)
	if len(afterRevs) <= len(beforeRevs) {
		t.Fatalf("expected new revision after peer_reconcile, before=%d after=%d", len(beforeRevs), len(afterRevs))
	}
	peerJobs, _, _ := reg.List(tenant, StatusSucceeded, KindPeerReconcile, "", 10)
	if len(peerJobs) == 0 {
		t.Fatal("expected succeeded peer_reconcile job")
	}
	meta := peerJobs[0].Snapshot()["meta"].(map[string]any)
	if _, ok := meta["revision_id"].(string); !ok {
		t.Fatalf("expected revision_id in peer_reconcile meta, got %#v", meta)
	}
	if title, ok := meta["job_title"].(string); !ok || title != "Обновление BGP пиров" {
		t.Fatalf("expected job_title in peer_reconcile meta, got %#v", meta["job_title"])
	}
	if _, ok := meta["deploy_apply_job_id"].(string); !ok {
		t.Fatalf("expected deploy_apply_job_id in peer_reconcile meta, got %#v", meta)
	}
}