Function TestWatchUpdateCurrentDeliversInitBeforeLiveEvents

pkg/server/server_test.go:126–241

(t *testing.T)

}

func TestWatchUpdateCurrentDeliversInitBeforeLiveEvents(t *testing.T) {
	ctx := context.Background()
	s1 := runNewServer(t, 1, "1.1.1.1", 10179)
	defer s1.StopBgp(context.Background(), &api.StopBgpRequest{})
	s2 := runNewServer(t, 1, "2.2.2.2", 20179)
	defer s2.StopBgp(context.Background(), &api.StopBgpRequest{})

	established := newPeerStateWaiter(s1, api.PeerState_SESSION_STATE_ESTABLISHED)
	err := peerServers(t, ctx, []*BgpServer{s1, s2}, []oc.AfiSafiType{oc.AFI_SAFI_TYPE_IPV4_UNICAST})
	require.NoError(t, err)
	established.Wait(t, 10*time.Second)

	panh, _ := bgp.NewPathAttributeNextHop(netip.MustParseAddr("10.0.0.1"))
	attrs := []bgp.PathAttributeInterface{
		bgp.NewPathAttributeOrigin(0),
		panh,
	}
	nlri, _ := bgp.NewIPAddrPrefix(netip.MustParsePrefix("10.10.0.0/24"))
	path, _ := apiutil.NewPath(bgp.RF_IPv4_UC, nlri, false, attrs, time.Now())
	_, err = s2.AddPath(apiutil.AddPathRequest{Paths: []*apiutil.Path{mustApi2apiutilPath(path)}})
	require.NoError(t, err)

	require.Eventually(t, func() bool {
		found := false
		_ = s1.mgmtOperation(func() error {
			for _, peer := range s1.neighborMap {
				if peer.adjRibIn.Count([]bgp.Family{bgp.RF_IPv4_UC}) > 0 {
					found = true
					break
				}
			}
			return nil
		}, true)
		return found
	}, 10*time.Second, 100*time.Millisecond)

	enterFilter := make(chan struct{})
	releaseFilter := make(chan struct{})
	var filterBlocked atomic.Bool
	var watcher *watcher
	done := make(chan struct{})

	go func() {
		defer close(done)
		watcher = s1.watch(func(o *watchOptions) {
			o.preUpdate = true
			o.initUpdate = true
			o.preUpdateFilter = func(w watchEvent) bool {
				if filterBlocked.CompareAndSwap(false, true) {
					close(enterFilter)
					<-releaseFilter
				}
				return true
			}
		})
	}()

	select {