Handle create/delete ref in Etcd notifier

Closes-Bug: #1794707
Change-Id: I4a6fd45f219a988f7529bd9af8755b5254e04186

pkg/db/etcd/event.go

@@ -85,7 +85,7 @@ func ParseEvent(oper int32, key string, newValue []byte) (*services.Event, error
 		Data:      data,
 	})
 	if err != nil {
-		return nil, errors.Wrapf(err, "failed to create event from data: %v, reason: %v", data)
+		return nil, errors.Wrapf(err, "failed to create event from data: %v", data)
 	}
 	return event, nil
 }

pkg/db/etcd/service_test.go

@@ -0,0 +1,193 @@
+package etcd_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+	"github.com/gogo/protobuf/types"
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/Juniper/contrail/pkg/db/etcd"
+	"github.com/Juniper/contrail/pkg/models"
+	"github.com/Juniper/contrail/pkg/services"
+	"github.com/Juniper/contrail/pkg/testutil"
+	"github.com/Juniper/contrail/pkg/testutil/integration/etcd"
+)
+
+func TestEtcdNotifierService(t *testing.T) {
+	tests := []struct {
+		name     string
+		ops      func(*testing.T, context.Context, *etcd.NotifierService)
+		watchers []watcher
+	}{
+		{
+			name: "create and update virtual network",
+			ops: func(t *testing.T, ctx context.Context, sv *etcd.NotifierService) {
+				_, err := sv.CreateVirtualNetwork(ctx, &services.CreateVirtualNetworkRequest{
+					VirtualNetwork: &models.VirtualNetwork{
+						UUID: "vn-blue",
+						Name: "vn_blue",
+					},
+				})
+				assert.NoError(t, err, "create virtual network failed")
+
+				_, err = sv.UpdateVirtualNetwork(ctx, &services.UpdateVirtualNetworkRequest{
+					VirtualNetwork: &models.VirtualNetwork{
+						UUID: "vn-blue",
+						Name: "vn_bluuee",
+					},
+					FieldMask: types.FieldMask{Paths: []string{"name"}},
+				})
+				assert.NoError(t, err, "update virtual network failed")
+			},
+			watchers: []watcher{
+				{
+					key: "/test/virtual_network/vn-blue",
+					expectedValues: []map[string]interface{}{
+						{
+							"name": "vn_blue",
+						},
+						{
+							"name": "vn_bluuee",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "create and delete reference from virtual network to logical router",
+			ops: func(t *testing.T, ctx context.Context, sv *etcd.NotifierService) {
+				_, err := sv.CreateVirtualNetwork(ctx, &services.CreateVirtualNetworkRequest{
+					VirtualNetwork: &models.VirtualNetwork{
+						UUID: "vn-blue",
+						Name: "vn_blue",
+					},
+				})
+				assert.NoError(t, err, "create virtual network failed")
+
+				_, err = sv.CreateLogicalRouter(ctx, &services.CreateLogicalRouterRequest{
+					LogicalRouter: &models.LogicalRouter{
+						UUID: "lr-blue",
+						Name: "lr_blue",
+					},
+				})
+				assert.NoError(t, err, "create logical router failed")
+
+				_, err = sv.CreateVirtualNetworkLogicalRouterRef(ctx,
+					&services.CreateVirtualNetworkLogicalRouterRefRequest{
+						ID: "vn-blue",
+						VirtualNetworkLogicalRouterRef: &models.VirtualNetworkLogicalRouterRef{
+							UUID: "lr-blue",
+						}})
+				assert.NoError(t, err, "create vn-lr reference failed")
+
+				_, err = sv.DeleteVirtualNetworkLogicalRouterRef(ctx,
+					&services.DeleteVirtualNetworkLogicalRouterRefRequest{
+						ID: "vn-blue",
+						VirtualNetworkLogicalRouterRef: &models.VirtualNetworkLogicalRouterRef{
+							UUID: "lr-blue",
+						}})
+				assert.NoError(t, err, "delete vn-lr reference failed")
+			},
+			watchers: []watcher{
+				{
+					key: "/test/virtual_network/vn-blue",
+					expectedValues: []map[string]interface{}{
+						{
+							"name": "vn_blue",
+						},
+						{
+							"name": "vn_blue",
+							"logical_router_refs": []interface{}{
+								map[string]interface{}{
+									"uuid": "lr-blue",
+								},
+							},
+						},
+						{
+							"name": "vn_blue",
+						},
+					},
+				},
+				{
+					key: "/test/logical_router/lr-blue",
+					expectedValues: []map[string]interface{}{
+						{
+							"name": "lr_blue",
+						},
+						{
+							"name": "lr_blue",
+							"virtual_network_backrefs": []interface{}{
+								map[string]interface{}{
+									"uuid": "vn-blue",
+								},
+							},
+						},
+						{
+							"name": "lr_blue",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	etcdPath := "test"
+	viper.Set("etcd.endpoints", integrationetcd.Endpoint)
+	viper.Set("etcd.path", etcdPath)
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ec := integrationetcd.NewEtcdClient(t)
+			defer ec.Close(t)
+
+			// Clean the database
+			ec.DeleteKey(t, etcdPath, clientv3.WithPrefix())
+
+			check := startWatchers(t, tt.watchers)
+			sv, err := etcd.NewNotifierService(etcdPath, models.JSONCodec)
+			require.NoError(t, err)
+
+			tt.ops(t, context.Background(), sv)
+
+			check(t)
+		})
+	}
+}
+
+type watcher struct {
+	key            string
+	expectedValues []map[string]interface{}
+}
+
+func startWatchers(t *testing.T, watchers []watcher) func(t *testing.T) {
+	checks := []func(t *testing.T){}
+
+	ec := integrationetcd.NewEtcdClient(t)
+	for _, w := range watchers {
+		w := w
+
+		collect := ec.WatchKeyN(w.key, len(w.expectedValues), 2*time.Second)
+		checks = append(checks, func(t *testing.T) {
+			collected := collect()
+			assert.Equal(t, len(w.expectedValues), len(collected))
+			for i, e := range w.expectedValues {
+				var data interface{}
+				err := json.Unmarshal([]byte(collected[i]), &data)
+				assert.NoError(t, err)
+				testutil.AssertEqual(t, e, data, "different data under key %s\n", w.key)
+			}
+		})
+	}
+
+	return func(t *testing.T) {
+		for _, c := range checks {
+			c(t)
+		}
+	}
+}

pkg/models/serialized_object.go

@@ -0,0 +1,40 @@
+package models
+
+import (
+	"github.com/gogo/protobuf/proto"
+)
+
+// SerializedObject represents serialized data and its object form.
+type SerializedObject struct {
+	data  []byte
+	obj   proto.Message
+	codec Codec
+}
+
+// NewSerializedObject constructs serialized object.
+func NewSerializedObject(d []byte, o proto.Message, c Codec) *SerializedObject {
+	return &SerializedObject{
+		data:  d,
+		obj:   o,
+		codec: c,
+	}
+}
+
+// GetData returns serialized data from object.
+func (s *SerializedObject) GetData() []byte {
+	if s != nil {
+		return s.data
+	}
+	return nil
+}
+
+// Map could be used to apply action on serialized object.
+func (s *SerializedObject) Map(f func()) error {
+	err := s.codec.Decode(s.data, s.obj)
+	if err != nil {
+		return err
+	}
+	f()
+	s.data, err = s.codec.Encode(s.obj)
+	return err
+}

pkg/models/virtual_network_test.go

@@ -229,8 +229,8 @@ func TestVirtualNetworkAddNetworkIpamRef(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			tt.virtualNetwork.AddNetworkIpamRef(tt.toAdd)
-			assert.Equal(t, tt.virtualNetwork.NetworkIpamRefs, tt.expected.NetworkIpamRefs)
-			assert.Equal(t, tt.virtualNetwork, tt.expected)
+			assert.Equal(t, tt.expected.NetworkIpamRefs, tt.virtualNetwork.NetworkIpamRefs)
+			assert.Equal(t, tt.expected, tt.virtualNetwork)
 		})
 	}
 }