import latest gcp and llama tests fixes from pgwatch_rpc_server

Author: 0xgouda

rpc/cmd/gcp_pubsub_receiver/pubsub_receiver_test.go

@@ -55,16 +55,17 @@ func TestPubsubReceiver(t *testing.T) {
 	a.NotNil(psr)
 
 	t.Run("Test Pub/Sub Receiver UpdateMeasurements()", func(t *testing.T) {
+		// To read the published message from the Pub/Sub server.
+		sub, err := CreateSubscription(psr)
+		a.NoError(err)
+
 		msg := testutils.GetTestMeasurementEnvelope()
 		reply, err := psr.UpdateMeasurements(context.Background(), msg)
+		psr.publisher.Flush()
 
 		a.NoError(err)
 		a.Equal(reply.GetLogmsg(), "Message published.")
 
-		// Try read the published message from the Pub/Sub server.
-		sub, err := CreateSubscription(psr)
-		a.NoError(err)
-
 		ctx, cancel := context.WithCancel(context.Background())
 		err = sub.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
 			var recvd_msg map[string]any

rpc/cmd/llama_receiver/llama_receiver_test.go

@@ -141,7 +141,7 @@ func TestLLamaReceiver(t *testing.T) {
 	})
 
 	t.Run("Update Measurements Multiple", func(t *testing.T) {
-		for range 10 {
+		for range 3 {
 			_, err := recv.UpdateMeasurements(ctx, msg)
 			assert.NoError(t, err, "error encountered while updating measurements")
 		}
@@ -151,7 +151,7 @@ func TestLLamaReceiver(t *testing.T) {
 		err := conn.QueryRow(recv.Ctx, "SELECT COUNT(*) FROM insights;").Scan(&newInsightsCount)
 
 		assert.NoError(t, err)
-		assert.Greater(t, newInsightsCount, 1, "No new entries inserted in insights table")
+		assert.GreaterOrEqual(t, newInsightsCount, 1, "No new entries inserted in insights table")
 	})
 
 	t.Run("LLama SyncMetricHandler", func(t *testing.T) {