connect_test.go

package main

import (
	"context"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"

	agenttesting "github.com/codefly-dev/core/agents/testing"
	agentv0 "github.com/codefly-dev/core/generated/go/codefly/services/agent/v0"
	builderv0 "github.com/codefly-dev/core/generated/go/codefly/services/builder/v0"
	"github.com/stretchr/testify/require"
	"gopkg.in/yaml.v3"
)

// TestSettingsParseConnectEndpoint proves the setting reads at go-grpc's
// spelling, defaults to off, and is refused without a gRPC server to transcode
// onto.
func TestSettingsParseConnectEndpoint(t *testing.T) {
	var on Settings
	require.NoError(t, yaml.Unmarshal([]byte(`
grpc-server:
  enabled: true
connect-endpoint: true
`), &on))
	require.True(t, on.ConnectEndpoint)
	require.NoError(t, on.connectSettingError())

	var alone Settings
	require.NoError(t, yaml.Unmarshal([]byte(`connect-endpoint: true`), &alone))
	require.ErrorContains(t, alone.connectSettingError(), "grpc-server.enabled")

	var absent Settings
	require.NoError(t, yaml.Unmarshal([]byte(`python-version: "3.12"`), &absent))
	require.False(t, absent.ConnectEndpoint)
}

// TestSyncWritesTheConnectFace proves the face lands where the app imports it,
// byte for byte, replacing a hand edit.
func TestSyncWritesTheConnectFace(t *testing.T) {
	builder := NewBuilder(NewService())
	builder.Location = t.TempDir()
	builder.FastAPI.Settings.GRPCServer.Enabled = true
	builder.FastAPI.Settings.ConnectEndpoint = true

	target := filepath.Join(builder.Location, connectFacePath)
	require.NoError(t, os.MkdirAll(filepath.Dir(target), 0o755))
	require.NoError(t, os.WriteFile(target, []byte("# edited by hand\n"), 0o644))

	require.NoError(t, builder.writeConnectFace())
	written, err := os.ReadFile(target)
	require.NoError(t, err)
	require.Equal(t, connectFace, written)
	require.Contains(t, string(written), "Do not edit this file!")

	builder.FastAPI.Settings.GRPCServer.Enabled = false
	require.Error(t, builder.writeConnectFace())
}

// TestDeploymentRendersConnectPort proves the Connect container and Service
// ports appear only when the service opts in, declared as HTTP/1.1.
func TestDeploymentRendersConnectPort(t *testing.T) {
	enabled := agenttesting.AssertKustomizeTemplates(t, deploymentFS, deploymentTestParameters(t, Parameters{GRPCEnabled: true, GRPCPort: 9090, ConnectEnabled: true, ConnectPort: 8081}))
	svc, err := os.ReadFile(filepath.Join(enabled, "base", "service.yaml"))
	require.NoError(t, err)
	require.Contains(t, string(svc), "name: connect-port")
	require.Contains(t, string(svc), "appProtocol: http\n")
	require.Contains(t, string(svc), "port: 8081")
	deployment, err := os.ReadFile(filepath.Join(enabled, "base", "deployment.yaml"))
	require.NoError(t, err)
	require.Contains(t, string(deployment), "containerPort: 8081")
	require.Contains(t, string(deployment), "- name: CODEFLY_CONNECT_PORT\n              value: \"8081\"")

	disabled := agenttesting.AssertKustomizeTemplates(t, deploymentFS, deploymentTestParameters(t, Parameters{GRPCEnabled: true, GRPCPort: 9090}))
	svc, err = os.ReadFile(filepath.Join(disabled, "base", "service.yaml"))
	require.NoError(t, err)
	require.NotContains(t, string(svc), "connect-port")
	deployment, err = os.ReadFile(filepath.Join(disabled, "base", "deployment.yaml"))
	require.NoError(t, err)
	require.NotContains(t, string(deployment), "CODEFLY_CONNECT_PORT")
}

// TestConnectFaceServesEveryProtocol runs the face's own suite
// (testdata/connect) against a real grpc.aio server: Connect unary and server
// streams in both codecs, gRPC-Web, gRPC relayed on the same port, metadata
// both ways, the service's interceptors on every protocol, deadlines and the
// HTTP refusals. The Python pins are the ones a service's lock resolves today.
func TestConnectFaceServesEveryProtocol(t *testing.T) {
	if _, err := exec.LookPath("uv"); err != nil {
		t.Skip("uv is not installed: the Connect face's Python suite cannot run")
	}
	cmd := exec.Command("uv", "run", "--no-project", "--python", "3.12",
		"--with", "grpcio==1.84.0", "--with", "protobuf==6.33.6", "--with", "h11==0.16.0",
		"--with", "httpx==0.28.1", "--with", "pytest==8.4.2", "--with", "pytest-asyncio==1.2.0",
		"pytest", "-q", "-p", "no:cacheprovider",
		"-o", "asyncio_mode=auto", "-o", "asyncio_default_fixture_loop_scope=function",
		filepath.Join("testdata", "connect"))
	cmd.Env = append(os.Environ(), "PYTHONPATH="+filepath.Join("templates", "connect"), "PYTHONDONTWRITEBYTECODE=1")
	out, err := cmd.CombinedOutput()
	require.NoError(t, err, "%s", out)
}

// TestBuilderLoadReadsTheAgentsSettings is the regression guard for a loaded
// (not created) service: the builder must see grpc-server and
// connect-endpoint, or Sync writes neither the gRPC stubs nor the Connect face
// and Deploy renders neither port.
func TestBuilderLoadReadsTheAgentsSettings(t *testing.T) {
	created, _ := createServiceForTest(t, map[string]*agentv0.Answer{
		HotReload:      confirmAnswer(false),
		PublicEndpoint: confirmAnswer(false),
		GRPCServer:     confirmAnswer(true),
	})
	ctx := context.Background()
	manifest := filepath.Join(created.Location, "service.codefly.yaml")
	content, err := os.ReadFile(manifest)
	require.NoError(t, err)
	var document map[string]any
	require.NoError(t, yaml.Unmarshal(content, &document))
	spec, _ := document["spec"].(map[string]any)
	require.Equal(t, true, spec["grpc-server"].(map[string]any)["enabled"])
	spec["connect-endpoint"] = true
	content, err = yaml.Marshal(document)
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(manifest, content, 0o644))

	identity, err := created.Identity.Proto()
	require.NoError(t, err)
	loaded := NewBuilder(NewService())
	resp, err := loaded.Load(ctx, &builderv0.LoadRequest{Identity: identity})
	require.NoError(t, err)
	require.NotEqual(t, builderv0.LoadStatus_ERROR, resp.GetState().GetState(), resp.GetState().GetMessage())
	require.True(t, loaded.FastAPI.Settings.GRPCServer.Enabled)
	require.True(t, loaded.FastAPI.Settings.ConnectEndpoint)
	require.Equal(t, filepath.Join(created.Location, "code"), loaded.Service.SourceLocation)
}

// TestSyncConnectFaceFollowsTheSetting proves Sync removes the face it wrote
// once connect-endpoint is off (the scaffold's main starts the face whenever
// the module is there) and leaves a file at that path the agent did not write.
func TestSyncConnectFaceFollowsTheSetting(t *testing.T) {
	builder := NewBuilder(NewService())
	builder.Location = t.TempDir()
	builder.FastAPI.Settings.GRPCServer.Enabled = true
	target := filepath.Join(builder.Location, connectFacePath)

	builder.FastAPI.Settings.ConnectEndpoint = true
	require.NoError(t, builder.syncConnectFace())
	require.FileExists(t, target)

	builder.FastAPI.Settings.ConnectEndpoint = false
	require.NoError(t, builder.syncConnectFace())
	require.NoFileExists(t, target)
	require.NoError(t, builder.syncConnectFace(), "nothing to remove is not an error")

	require.NoError(t, os.WriteFile(target, []byte("# the service's own\n"), 0o644))
	require.NoError(t, builder.syncConnectFace())
	require.FileExists(t, target)
}

// TestGeneratedMainServesConnect boots a scaffolded gRPC service's own
// src/main.py under uvicorn, as the runtime does. With connect-endpoint on
// (CODEFLY_CONNECT_PORT set, the face synced), the face answers a Connect call
// and stops with the app. With it off (the variable unset), a face left on
// disk binds nothing, not even its 8081 default. On but never synced, the app
// refuses to start and says to sync.
func TestGeneratedMainServesConnect(t *testing.T) {
	if testing.Short() {
		t.Skip("boots a scaffolded service with uv")
	}
	if _, err := exec.LookPath("uv"); err != nil {
		t.Skip("uv is not installed: the scaffolded service cannot run")
	}
	builder, root := createServiceForTest(t, map[string]*agentv0.Answer{
		HotReload:      confirmAnswer(false),
		PublicEndpoint: confirmAnswer(false),
		GRPCServer:     confirmAnswer(true),
	})
	code := filepath.Join(root, "code")
	generateScaffoldStubs(t, code)

	builder.FastAPI.Settings.ConnectEndpoint = true
	require.NoError(t, builder.syncConnectFace())
	app := bootScaffold(t, code, true)
	request, err := http.NewRequest(http.MethodPost,
		fmt.Sprintf("http://127.0.0.1:%d/api.EchoService/Echo", app.connect),
		strings.NewReader(`{"message":"over connect","addedLater":1}`))
	require.NoError(t, err)
	request.Header.Set("Content-Type", "application/json")
	response, err := http.DefaultClient.Do(request)
	require.NoError(t, err)
	body, err := io.ReadAll(response.Body)
	_ = response.Body.Close()
	require.NoError(t, err)
	require.Equal(t, http.StatusOK, response.StatusCode, "%s", body)
	require.JSONEq(t, `{"message":"over connect"}`, string(body))
	require.Equal(t, uint32(1000), advertisedMaxStreams(t, app.grpc),
		"the scaffold's gRPC server advertises the limit the face enforces, so a burst queues in the client")
	app.stop(t)
	require.False(t, listens(app.connect), "the face outlived the app")

	// Off, with the face still on disk (no sync since): the variable is the switch.
	defaultFree := !listens(8081)
	app = bootScaffold(t, code, false)
	require.True(t, listens(app.grpc), "the gRPC server is up")
	if defaultFree {
		require.False(t, listens(8081), "a service without connect-endpoint bound the face's default port")
	}
	app.stop(t)

	// On, never synced: the app says what is missing instead of serving no face.
	require.NoError(t, os.Remove(filepath.Join(root, connectFacePath)))
	refused := startScaffold(t, code, true)
	select {
	case <-refused.exited:
	case <-time.After(3 * time.Minute):
		t.Fatalf("the app served without its Connect face:\n%s", refused.output)
	}
	require.Contains(t, refused.output.String(), "src/rpc/connect_gen.py is missing: run `codefly sync service`")
}

type scaffoldApp struct {
	cmd                 *exec.Cmd
	exited              chan error
	output              *lockedOutput
	rest, grpc, connect uint16
}

// startScaffold launches the scaffold's app under uvicorn with the variables
// the runtime hands it: CODEFLY_CONNECT_PORT only when connect is on.
func startScaffold(t *testing.T, code string, connect bool) *scaffoldApp {
	t.Helper()
	app := &scaffoldApp{output: &lockedOutput{}, exited: make(chan error, 1),
		rest: freePort(t), grpc: freePort(t), connect: freePort(t)}
	app.cmd = exec.Command("uv", "run", "uvicorn", "src.main:app", "--host", "127.0.0.1", "--port", fmt.Sprint(app.rest))
	app.cmd.Dir = code
	environment := []string{fmt.Sprintf("CODEFLY_GRPC_PORT=%d", app.grpc)}
	for _, variable := range os.Environ() {
		if !strings.HasPrefix(variable, "CODEFLY_CONNECT_PORT=") {
			environment = append(environment, variable)
		}
	}
	if connect {
		environment = append(environment, fmt.Sprintf("CODEFLY_CONNECT_PORT=%d", app.connect))
	}
	app.cmd.Env = environment
	app.cmd.Stdout, app.cmd.Stderr = app.output, app.output
	require.NoError(t, app.cmd.Start())
	go func() { app.exited <- app.cmd.Wait() }()
	t.Cleanup(func() { _ = app.cmd.Process.Kill() })
	return app
}

// bootScaffold starts the scaffold's app and returns once it answers on its
// REST port: uvicorn listens only after every startup handler, gRPC and
// Connect included, has run.
func bootScaffold(t *testing.T, code string, connect bool) *scaffoldApp {
	t.Helper()
	app := startScaffold(t, code, connect)
	deadline := time.Now().Add(3 * time.Minute) // a cold uv sync
	for !answers(app.rest) {
		select {
		case err := <-app.exited:
			t.Fatalf("the app exited before serving: %v\n%s", err, app.output)
		case <-time.After(200 * time.Millisecond):
		}
		require.True(t, time.Now().Before(deadline), "the app never served:\n%s", app.output)
	}
	return app
}

// stop asks uvicorn to shut down, which runs the app's shutdown handlers.
func (a *scaffoldApp) stop(t *testing.T) {
	t.Helper()
	require.NoError(t, a.cmd.Process.Signal(syscall.SIGTERM))
	select {
	case <-a.exited:
	case <-time.After(30 * time.Second):
		t.Fatalf("the app did not stop:\n%s", a.output)
	}
	require.NotContains(t, a.output.String(), "Traceback", a.output.String())
}

// answers reports whether the scaffold's /version route answers.
func answers(port uint16) bool {
	client := http.Client{Timeout: 500 * time.Millisecond}
	resp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d/version", port))
	if err != nil {
		return false
	}
	_ = resp.Body.Close()
	return resp.StatusCode == http.StatusOK
}

// lockedOutput is the app's combined output, read while the app writes it.
type lockedOutput struct {
	mu  sync.Mutex
	buf strings.Builder
}

func (o *lockedOutput) Write(p []byte) (int, error) {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.buf.Write(p)
}

func (o *lockedOutput) String() string {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.buf.String()
}

// advertisedMaxStreams reads SETTINGS_MAX_CONCURRENT_STREAMS from the first
// SETTINGS frame the gRPC server on port sends; 0 when it advertises none.
func advertisedMaxStreams(t *testing.T, port uint16) uint32 {
	t.Helper()
	conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 2*time.Second)
	require.NoError(t, err)
	defer conn.Close()
	require.NoError(t, conn.SetDeadline(time.Now().Add(5*time.Second)))
	_, err = conn.Write(append([]byte("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"), 0, 0, 0, 4, 0, 0, 0, 0, 0))
	require.NoError(t, err)
	for {
		header := make([]byte, 9)
		_, err = io.ReadFull(conn, header)
		require.NoError(t, err)
		payload := make([]byte, int(header[0])<<16|int(header[1])<<8|int(header[2]))
		_, err = io.ReadFull(conn, payload)
		require.NoError(t, err)
		if header[3] != 4 || header[4]&1 != 0 { // not a SETTINGS, or an ACK
			continue
		}
		for at := 0; at+6 <= len(payload); at += 6 {
			if binary.BigEndian.Uint16(payload[at:]) == 3 {
				return binary.BigEndian.Uint32(payload[at+2:])
			}
		}
		return 0
	}
}

func listens(port uint16) bool {
	conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Second)
	if err != nil {
		return false
	}
	_ = conn.Close()
	return true
}