Merge pull request #111 from DIMO-Network/ex-3369-move-telemetry-api-…

…to-use-fetch-api-for-vinvcs

Ex 3369 move telemetry api to use fetch api for vinvcs

e2e/auth_server_test.go

@@ -10,6 +10,7 @@ import (
 	"net/http/httptest"
 	"testing"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/go-jose/go-jose/v4"
 )
 
@@ -22,7 +23,7 @@ type mockAuthServer struct {
 	ManufacturerContractAddress string
 }
 
-func setupAuthServer(t *testing.T, vehicleContractAddress, manufacturerContractAddress string) *mockAuthServer {
+func setupAuthServer(t *testing.T, vehicleContractAddress, manufacturerContractAddress common.Address) *mockAuthServer {
 	t.Helper()
 
 	// Generate RSA key
@@ -73,8 +74,8 @@ func setupAuthServer(t *testing.T, vehicleContractAddress, manufacturerContractA
 		signer:                      sig,
 		jwks:                        jwk,
 		defaultClaims:               defaultClaims,
-		VehicleContractAddress:      vehicleContractAddress,
-		ManufacturerContractAddress: manufacturerContractAddress,
+		VehicleContractAddress:      vehicleContractAddress.String(),
+		ManufacturerContractAddress: manufacturerContractAddress.String(),
 	}
 
 	// Create test server with only JWKS endpoint

e2e/clickhouse_container_test.go

@@ -5,16 +5,14 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/DIMO-Network/clickhouse-infra/pkg/connect"
 	chconfig "github.com/DIMO-Network/clickhouse-infra/pkg/connect/config"
 	"github.com/DIMO-Network/clickhouse-infra/pkg/container"
 	sigmigrations "github.com/DIMO-Network/model-garage/pkg/migrations"
 	"github.com/DIMO-Network/model-garage/pkg/vss"
-	indexmigrations "github.com/DIMO-Network/nameindexer/pkg/clickhouse/migrations"
 	"github.com/stretchr/testify/require"
 )
 
-func setupClickhouseContainer(t *testing.T, indexDB string) *container.Container {
+func setupClickhouseContainer(t *testing.T) *container.Container {
 	t.Helper()
 	ctx := context.Background()
 
@@ -32,19 +30,6 @@ func setupClickhouseContainer(t *testing.T, indexDB string) *container.Container
 	if err != nil {
 		t.Fatalf("Failed to run migrations: %v", err)
 	}
-	// Create index database
-	_, err = db.Exec("CREATE DATABASE IF NOT EXISTS " + indexDB)
-	require.NoError(t, err, "Failed to create index database")
-	chConfig := chContainer.Config()
-	chConfig.Database = indexDB
-	fileDB := connect.GetClickhouseDB(&chConfig)
-	if err != nil {
-		t.Fatalf("Failed to get clickhouse connection: %v", err)
-	}
-	err = indexmigrations.RunGoose(ctx, []string{"up", "-v"}, fileDB)
-	if err != nil {
-		t.Fatalf("Failed to run migrations: %v", err)
-	}
 
 	return chContainer
 }

e2e/fetchapi_test.go

@@ -0,0 +1,154 @@
+// Package rpc provides the gRPC server implementation for the index repo service.
+package e2e_test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	pb "github.com/DIMO-Network/fetch-api/pkg/grpc"
+	"github.com/DIMO-Network/model-garage/pkg/cloudevent"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+// mockFetchServer wraps the gRPC server and contains test configuration
+type mockFetchServer struct {
+	grpcServer       *grpc.Server
+	listener         net.Listener
+	port             int
+	mutex            sync.Mutex
+	cloudeventReturn cloudevent.CloudEvent[json.RawMessage]
+	pb.UnimplementedFetchServiceServer
+	t *testing.T
+}
+
+// NewTestFetchAPI creates and starts a gRPC server on a random available port
+func NewTestFetchAPI(t *testing.T) *mockFetchServer {
+	// Find an available port
+	listener, err := net.Listen("tcp", ":0")
+	require.NoError(t, err)
+
+	// Create the gRPC server
+	grpcServer := grpc.NewServer()
+	testServer := &mockFetchServer{
+		grpcServer: grpcServer,
+		t:          t,
+		listener:   listener,
+		port:       listener.Addr().(*net.TCPAddr).Port,
+	}
+
+	pb.RegisterFetchServiceServer(grpcServer, testServer)
+
+	// Start the server
+	go func() {
+		if err := grpcServer.Serve(listener); err != nil {
+			t.Logf("server stopped: %v", err)
+		}
+	}()
+
+	// Wait a moment for the server to start
+	time.Sleep(100 * time.Millisecond)
+
+	return testServer
+
+}
+
+// Stop gracefully stops the test server
+func (ts *mockFetchServer) Close() {
+	ts.grpcServer.GracefulStop()
+
+	if ts.listener != nil {
+		ts.listener.Close()
+	}
+}
+
+func (ts *mockFetchServer) SetCloudEventReturn(ce cloudevent.CloudEvent[json.RawMessage]) {
+	ts.mutex.Lock()
+	ts.cloudeventReturn = ce
+	ts.mutex.Unlock()
+}
+
+// GetAddress returns the full address of the server
+func (ts *mockFetchServer) URL() string {
+	return fmt.Sprintf("localhost:%d", ts.port)
+}
+
+// GetLatestIndex translates the gRPC call to the indexrepo type and returns the latest index for the given options.
+func (s *mockFetchServer) GetLatestIndex(ctx context.Context, req *pb.GetLatestIndexRequest) (*pb.GetLatestIndexResponse, error) {
+	return nil, nil
+}
+
+// ListIndex translates the pb call to the indexrepo type and fetches index keys for the given options.
+func (s *mockFetchServer) ListIndex(ctx context.Context, req *pb.ListIndexesRequest) (*pb.ListIndexesResponse, error) {
+	return nil, nil
+}
+
+// ListCloudEvents translates the pb call to the indexrepo type and fetches data for the given options.
+func (s *mockFetchServer) ListCloudEvents(ctx context.Context, req *pb.ListCloudEventsRequest) (*pb.ListCloudEventsResponse, error) {
+	return nil, nil
+}
+
+// GetLatestCloudEvent translates the pb call to the indexrepo type and fetches the latest data for the given options.
+func (s *mockFetchServer) GetLatestCloudEvent(ctx context.Context, req *pb.GetLatestCloudEventRequest) (*pb.GetLatestCloudEventResponse, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	return &pb.GetLatestCloudEventResponse{
+		CloudEvent: cloudEventToProto(s.cloudeventReturn),
+	}, nil
+
+}
+
+// ListCloudEventsFromIndex translates the pb call to the indexrepo type and fetches data for the given index keys.
+func (s *mockFetchServer) ListCloudEventsFromIndex(ctx context.Context, req *pb.ListCloudEventsFromKeysRequest) (*pb.ListCloudEventsFromKeysResponse, error) {
+	return nil, nil
+}
+
+func cloudEventHeaderToProto(event *cloudevent.CloudEventHeader) *pb.CloudEventHeader {
+	if event == nil {
+		return nil
+	}
+	extras := make(map[string][]byte)
+	for k, v := range event.Extras {
+		v, err := json.Marshal(v)
+		if err != nil {
+			// Skip the extra if it can't be marshaled
+			continue
+		}
+		extras[k] = v
+	}
+	return &pb.CloudEventHeader{
+		Id:              event.ID,
+		Source:          event.Source,
+		Producer:        event.Producer,
+		Subject:         event.Subject,
+		SpecVersion:     event.SpecVersion,
+		Time:            timestamppb.New(event.Time),
+		Type:            event.Type,
+		DataContentType: event.DataContentType,
+		DataSchema:      event.DataSchema,
+		DataVersion:     event.DataVersion,
+		Extras:          extras,
+	}
+}
+
+func cloudEventToProto(event cloudevent.CloudEvent[json.RawMessage]) *pb.CloudEvent {
+	extras := make(map[string][]byte)
+	for k, v := range event.Extras {
+		v, err := json.Marshal(v)
+		if err != nil {
+			// Skip the extra if it can't be marshaled
+			continue
+		}
+		extras[k] = v
+	}
+	return &pb.CloudEvent{
+		Header: cloudEventHeaderToProto(&event.CloudEventHeader),
+		Data:   event.Data,
+	}
+}