aeon: mock gRPC server

Mock server Implement some base methods for integration tests.

Part of #1050

test/integration/aeon/conftest.py

@@ -0,0 +1,40 @@
+import os
+from pathlib import Path
+from signal import SIGQUIT
+from subprocess import PIPE, STDOUT, Popen, run
+import pytest
+
+from utils import wait_for_lines_in_output
+
+
+@pytest.fixture(scope="session")
+def mock_aeon(tmp_path_factory) -> Path:
+    server_dir = Path(__file__).parent / "server"
+    exec = tmp_path_factory.mktemp("aeon_mock") / "aeon"
+    result = run(f"go build -C {server_dir} -o {exec}".split())
+    assert result.returncode == 0, "Failed build mock aeon server"
+    return exec
+
+
+@pytest.fixture(params=[50052, "@aeon_unix_socket", "AEON"])
+def aeon_plain(mock_aeon, tmp_path, request):
+    cmd = [mock_aeon]
+    param = request.param
+    if isinstance(param, int):
+        cmd.append(f"-port={param}")
+    elif isinstance(param, str):
+        if param[0] != "@":
+            param = tmp_path / param
+        cmd.append(f"-unix={param}")
+
+    aeon = Popen(
+        cmd,
+        env=dict(os.environ, GRPC_GO_LOG_SEVERITY_LEVEL="info"),
+        stderr=STDOUT,
+        stdout=PIPE,
+        text=True,
+    )
+    print(wait_for_lines_in_output(aeon.stdout, ["ListenSocket created"]))
+    yield param
+    aeon.send_signal(SIGQUIT)
+    assert aeon.wait(5) == 0, "Mock aeon server didn't stopped properly"

test/integration/aeon/server/.gitignore

@@ -0,0 +1,2 @@
+*.proto
+aeon

test/integration/aeon/server/README.md

@@ -0,0 +1,15 @@
+# Mock `aeon` server
+
+## Update and generate from `.proto` file
+
+1. cd `test/integration/aeon/server`;
+2. Get a new [aeon_router.proto](https://github.com/tarantool/aeon/blob/master/proto/aeon_router.proto);
+3. Run: `protoc --go_out=. --go-grpc_out=. aeon_router.proto`;
+4. Directory `pb` will be created with files:
+```
+pb
+├── aeon_router_grpc.pb.go
+└── aeon_router.pb.go
+```
+5. Commit the new changes in the created files;
+6. Update the `service/server.go` file to comply with the new `*.pb.go` requirements.

test/integration/aeon/server/go.mod

@@ -0,0 +1,15 @@
+module mock/server/aeon
+
+go 1.22.8
+
+require (
+	google.golang.org/grpc v1.68.1
+	google.golang.org/protobuf v1.35.2
+)
+
+require (
+	golang.org/x/net v0.29.0 // indirect
+	golang.org/x/sys v0.25.0 // indirect
+	golang.org/x/text v0.18.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
+)

test/integration/aeon/server/go.sum

@@ -0,0 +1,16 @@
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
+golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
+golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
+golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
+golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU=
+google.golang.org/grpc v1.68.1 h1:oI5oTa11+ng8r8XMMN7jAOmWfPZWbYpCFaMUTACxkM0=
+google.golang.org/grpc v1.68.1/go.mod h1:+q1XYFJjShcqn0QZHvCyeR4CXPA+llXIeUIfIe00waw=
+google.golang.org/protobuf v1.35.2 h1:8Ar7bF+apOIoThw1EdZl0p1oWvMqTHmpA2fRTyZO8io=
+google.golang.org/protobuf v1.35.2/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=

test/integration/aeon/server/mock.go

@@ -0,0 +1,148 @@
+package main
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"flag"
+	"fmt"
+	"log"
+	"mock/server/aeon/pb"
+	"mock/server/aeon/service"
+	"net"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"syscall"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+var args = struct {
+	is_ssl      *bool
+	ca_file     *string
+	cert_file   *string
+	key_file    *string
+	port        *int
+	unix_socket *string
+}{
+	is_ssl:      flag.Bool("ssl", false, "Connection uses SSL if set, (default plain TCP)"),
+	ca_file:     flag.String("ca_file", "", "The CA file"),
+	cert_file:   flag.String("cert_file", "", "The TLS cert file"),
+	key_file:    flag.String("key_file", "", "The TLS key file"),
+	port:        flag.Int("port", 50051, "The server port"),
+	unix_socket: flag.String("unix", "", "The Unix socket name"),
+}
+
+func getCertificate() tls.Certificate {
+	if *args.cert_file == "" || *args.key_file == "" {
+		log.Fatalln("Both 'key_file' and 'cert_file' required")
+	}
+	cert, err := os.ReadFile(*args.cert_file)
+	if err != nil {
+		log.Fatalf("Failed to read Cert file: %v", err)
+	}
+	key, err := os.ReadFile(*args.key_file)
+	if err != nil {
+		log.Fatalf("Failed to read Key file: %v", err)
+	}
+
+	tls_cert, err := tls.X509KeyPair(cert, key)
+	if err != nil {
+		log.Fatalln("Failed create key pair")
+	}
+	return tls_cert
+}
+
+func getTlsConfig() *tls.Config {
+	if *args.ca_file == "" {
+		return &tls.Config{
+			Certificates: []tls.Certificate{getCertificate()},
+			ClientAuth:   tls.NoClientCert,
+		}
+	}
+
+	ca, err := os.ReadFile(*args.ca_file)
+	if err != nil {
+		log.Fatalf("Failed to read CA file: %v", err)
+	}
+	certPool := x509.NewCertPool()
+	if !certPool.AppendCertsFromPEM(ca) {
+		log.Fatalln("Failed to append CA data")
+	}
+	return &tls.Config{
+		Certificates: []tls.Certificate{getCertificate()},
+		ClientAuth:   tls.RequireAndVerifyClientCert,
+		ClientCAs:    certPool,
+	}
+}
+
+func getServerOpts() []grpc.ServerOption {
+	if !*args.is_ssl {
+		return []grpc.ServerOption{}
+	}
+	creds := credentials.NewTLS(getTlsConfig())
+	return []grpc.ServerOption{grpc.Creds(creds)}
+}
+
+func getListener() net.Listener {
+	var protocol string
+	var address string
+
+	if *args.unix_socket != "" {
+		protocol = "unix"
+		address = *args.unix_socket
+		if strings.HasPrefix(address, "@") {
+			address = "\x00" + address[1:]
+		}
+	} else {
+		protocol = "tcp"
+		address = fmt.Sprintf("localhost:%d", *args.port)
+	}
+	lis, err := net.Listen(protocol, address)
+	if err != nil {
+		log.Fatalf("Failed to listen: %v", err)
+	}
+	return lis
+}
+
+func exit() {
+	log.Println("Exit func")
+	syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
+	os.Exit(0)
+}
+
+func main() {
+	log.Println("Start aeon mock server:", os.Args)
+
+	flag.Parse()
+
+	srv := grpc.NewServer(getServerOpts()...)
+	pb.RegisterAeonRouterServiceServer(srv, &service.Server{})
+
+	// Run gRPC server.
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		if err := srv.Serve(getListener()); err != nil {
+			log.Fatalf("Failed to serve: %v", err)
+		}
+		wg.Done()
+	}()
+
+	// Shutdown on signals.
+	exit_sig := make(chan os.Signal, 1)
+	signal.Notify(exit_sig,
+		syscall.SIGTERM,
+		syscall.SIGINT,
+		syscall.SIGQUIT,
+		syscall.SIGHUP,
+	)
+	s := <-exit_sig
+	log.Println("Got terminate signal:", s)
+
+	srv.GracefulStop()
+	wg.Wait()
+	log.Println("Exit aeon mock server.")
+}