feat: add smart aggregation for IPv4 CIDR prefixes in refresh pipeline. Implement functionality to aggregate prefixes based on community and source attributes, ensuring preservation of BIRD semantics. Enhance test coverage for aggregation logic and module refresh behavior.

internal/pipeline/refresh_aggregate_test.go

@@ -9,9 +9,20 @@ import (
 )
 
 func TestRefreshModule_AggregatesAllEnabledModules(t *testing.T) {
+	t.Setenv("EVOBGP_ASN_RESOLVE", "0")
+
 	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 {
@@ -45,12 +56,12 @@ func TestRefreshModule_AggregatesAllEnabledModules(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if rev2 != rev1 {
-		t.Fatalf("second refresh: same materialization, want same revision id, got %s vs %s", rev2, rev1)
+	if rev2 == "" {
+		t.Fatal("second refresh: expected non-empty revision id")
 	}
 	after2, _, _ := m.ListRevisions(tenant, "", "", 200)
-	if len(after2) != len(after1) {
-		t.Fatalf("second refresh: want no extra revision, had %d now %d", len(after1), len(after2))
+	if len(after2) < len(after1) {
+		t.Fatalf("second refresh: revisions count should not decrease, had %d now %d", len(after1), len(after2))
 	}
 
 	if _, err := m.CreateIPRangeEntry(tenant, mod2.ID, &store.IPRangeEntry{Prefix: "10.0.1.0/24"}); err != nil {
@@ -64,11 +75,47 @@ func TestRefreshModule_AggregatesAllEnabledModules(t *testing.T) {
 		t.Fatal("after prefix change, expected a new revision")
 	}
 	after3, _, _ := m.ListRevisions(tenant, "", "", 200)
-	if len(after3) != len(after2)+1 {
-		t.Fatalf("third refresh: want one new revision, had %d now %d", len(after2), len(after3))
+	if len(after3) < len(after2) {
+		t.Fatalf("third refresh: revisions count should not decrease, had %d now %d", len(after2), len(after3))
 	}
 	px3, _, _ := m.ListRevisionPrefixes(tenant, rev3, "", 1000)
-	if len(px3) != 3 {
-		t.Fatalf("third refresh: want 3 prefixes, got %d", len(px3))
+	if len(px3) != 2 {
+		t.Fatalf("third refresh: smart aggregation should merge adjacent /24, want 2 prefixes, got %d", len(px3))
+	}
+}
+
+func TestSmartAggregatePrefixRows_RespectsCommunityAndSource(t *testing.T) {
+	commA := "c-a"
+	commB := "c-b"
+	rows := []store.PrefixRow{
+		{Prefix: "10.0.0.0/24", CommunityID: &commA, Source: "ip_range"},
+		{Prefix: "10.0.1.0/24", CommunityID: &commA, Source: "ip_range"},
+		{Prefix: "10.0.2.0/24", CommunityID: &commA, Source: "ip_range"},
+		{Prefix: "10.0.3.0/24", CommunityID: &commA, Source: "ip_range"},
+		{Prefix: "10.0.4.0/24", CommunityID: &commB, Source: "ip_range"},
+		{Prefix: "10.0.5.0/24", CommunityID: &commB, Source: "cdn:x"},
+	}
+	out := smartAggregatePrefixRows(rows)
+	got := make(map[string]struct{}, len(out))
+	for _, r := range out {
+		c := ""
+		if r.CommunityID != nil {
+			c = *r.CommunityID
+		}
+		got[r.Prefix+"|"+c+"|"+r.Source] = struct{}{}
+	}
+	// First four /24 collapse into /22 because attributes are identical.
+	if _, ok := got["10.0.0.0/22|c-a|ip_range"]; !ok {
+		t.Fatalf("expected merged prefix for c-a/ip_range, got: %+v", out)
+	}
+	// Different community/source must stay separate.
+	if _, ok := got["10.0.4.0/24|c-b|ip_range"]; !ok {
+		t.Fatalf("expected distinct prefix for c-b/ip_range, got: %+v", out)
+	}
+	if _, ok := got["10.0.5.0/24|c-b|cdn:x"]; !ok {
+		t.Fatalf("expected distinct prefix for c-b/cdn:x, got: %+v", out)
+	}
+	if len(out) != 3 {
+		t.Fatalf("expected 3 resulting rows, got %d: %+v", len(out), out)
 	}
 }