update the source client

Signed-off-by: myan <[email protected]>

test/e2e/pkg/sourceclient_test.go

@@ -32,7 +32,19 @@ var _ = Describe("gRPC Source ManifestWork Client Test", func() {
 		BeforeEach(func() {
 			watchedWorks = []*workv1.ManifestWork{}
 
-			watcher, err := workClient.ManifestWorks(consumer_name).Watch(ctx, metav1.ListOptions{})
+			grpcOptions = grpc.NewGRPCOptions()
+			grpcOptions.URL = grpcServerAddress
+
+			workClient, err := work.NewClientHolderBuilder(grpcOptions).
+				WithClientID(fmt.Sprintf("%s-watcher", sourceID)).
+				WithSourceID(sourceID).
+				WithCodecs(codec.NewManifestBundleCodec()).
+				WithWorkClientWatcherStore(grpcsource.NewRESTFullAPIWatcherStore(apiClient, sourceID)).
+				WithResyncEnabled(false).
+				NewSourceClientHolder(ctx)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			watcher, err := workClient.ManifestWorks(consumer.Name).Watch(ctx, metav1.ListOptions{})
 			Expect(err).ShouldNot(HaveOccurred())
 
 			go func() {
@@ -72,36 +84,36 @@ var _ = Describe("gRPC Source ManifestWork Client Test", func() {
 
 			By("create a work")
 			workName := "work-" + rand.String(5)
-			_, err = workClient.ManifestWorks(consumer_name).Create(ctx, NewManifestWork(workName), metav1.CreateOptions{})
+			_, err = workClient.ManifestWorks(consumer.Name).Create(ctx, NewManifestWork(workName), metav1.CreateOptions{})
 			Expect(err).ShouldNot(HaveOccurred())
 
 			// wait for few seconds to ensure the creation is finished
 			<-time.After(5 * time.Second)
 
 			By("list the works")
-			works, err := workClient.ManifestWorks(consumer_name).List(ctx, metav1.ListOptions{})
+			works, err := workClient.ManifestWorks(consumer.Name).List(ctx, metav1.ListOptions{})
 			Expect(err).ShouldNot(HaveOccurred())
 			Expect(len(works.Items) == 1).To(BeTrue())
 
 			// wait for few seconds to ensure the work status is updated by agent
 			<-time.After(5 * time.Second)
 
 			By("update a work")
-			work, err := workClient.ManifestWorks(consumer_name).Get(ctx, workName, metav1.GetOptions{})
+			work, err := workClient.ManifestWorks(consumer.Name).Get(ctx, workName, metav1.GetOptions{})
 			Expect(err).ShouldNot(HaveOccurred())
 
 			newWork := work.DeepCopy()
 			newWork.Spec.Workload.Manifests = []workv1.Manifest{NewManifest(workName)}
 			patchData, err := ToWorkPatch(work, newWork)
 			Expect(err).ShouldNot(HaveOccurred())
-			_, err = workClient.ManifestWorks(consumer_name).Patch(ctx, workName, types.MergePatchType, patchData, metav1.PatchOptions{})
+			_, err = workClient.ManifestWorks(consumer.Name).Patch(ctx, workName, types.MergePatchType, patchData, metav1.PatchOptions{})
 			Expect(err).ShouldNot(HaveOccurred())
 
 			// wait for few seconds to ensure the work status is updated by agent
 			<-time.After(5 * time.Second)
 
 			By("delete a work")
-			err = workClient.ManifestWorks(consumer_name).Delete(ctx, workName, metav1.DeleteOptions{})
+			err = workClient.ManifestWorks(consumer.Name).Delete(ctx, workName, metav1.DeleteOptions{})
 			Expect(err).ShouldNot(HaveOccurred())
 
 			Eventually(func() error {
@@ -123,7 +135,6 @@ var _ = Describe("gRPC Source ManifestWork Client Test", func() {
 
 				return nil
 			}, 30*time.Second, 1*time.Second).ShouldNot(HaveOccurred())
-
 		})
 	})
 })