summercms/modules/pier/sessions_smoke_test.go

package pier_test

import (
	"bufio"
	"bytes"
	"context"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/pact"
	"git.golem15.com/golem15/summercms/modules/party"
	"github.com/alicebob/miniredis/v2"
	"github.com/modelcontextprotocol/go-sdk/mcp"
)

// Smoke tests of phase 19.7 plan 02 Task 3: Redis sessions shared by two
// pier instances on one miniredis (rehydration, binding, DELETE, idle TTL,
// the per-binding cap) and the pub/sub fan-out of subject events.

// instance is one mounted pier with its app and test server.
type instance struct {
	app  *backpack.App
	host *testHost
	srv  *httptest.Server
}

// newInstance mounts pier on a fresh app sharing the miniredis and guard.
func newInstance(t *testing.T, mr *miniredis.Miniredis, mcpYAML string, guard *acmeGuard) *instance {
	t.Helper()
	files := map[string]string{"mcp.yaml": mcpYAML}
	if mr != nil {
		files["redis.yaml"] = "url: redis://" + mr.Addr() + "/0\n"
	}
	app := pierAppWith(t, files, guard)
	host, err := mountPier(t, app, []party.Plugin{acmePlugin{tools: []pact.MCPTool{echoTool()}}})
	if err != nil {
		t.Fatal(err)
	}
	srv := httptest.NewServer(host)
	t.Cleanup(srv.Close)
	return &instance{app: app, host: host, srv: srv}
}

// rpcReply is one raw JSON-RPC exchange: the status, the session header
// and the first message (an SSE data line or the JSON body).
type rpcReply struct {
	status  int
	session string
	body    string
	header  http.Header
}

// rpc posts one JSON-RPC message with a bearer and an optional session id.
func rpc(t *testing.T, ctx context.Context, url, token, session, body string) rpcReply {
	t.Helper()
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBufferString(body))
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Authorization", "Bearer "+token)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Accept", "application/json, text/event-stream")
	req.Header.Set("Mcp-Protocol-Version", "2025-11-25")
	if session != "" {
		req.Header.Set("Mcp-Session-Id", session)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	reply := rpcReply{status: resp.StatusCode, session: resp.Header.Get("Mcp-Session-Id"), header: resp.Header}
	if strings.HasPrefix(resp.Header.Get("Content-Type"), "text/event-stream") {
		scanner := bufio.NewScanner(resp.Body)
		scanner.Buffer(make([]byte, 1<<20), 1<<20)
		for scanner.Scan() {
			line := scanner.Text()
			if strings.HasPrefix(line, "data: ") {
				reply.body = strings.TrimPrefix(line, "data: ")
				break
			}
		}
		return reply
	}
	raw, _ := io.ReadAll(resp.Body)
	reply.body = strings.TrimSpace(string(raw))
	return reply
}

const initializeBody = `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"raw","version":"1"}}}`
const initializedBody = `{"jsonrpc":"2.0","method":"notifications/initialized"}`
const callBody = `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"acme_echo","arguments":{"id":9,"text":"hey"}}}`

// openSession initialises a session on an instance and completes the
// handshake, returning the session id.
func openSession(t *testing.T, ctx context.Context, in *instance, token string) string {
	t.Helper()
	init := rpc(t, ctx, in.srv.URL+"/mcp", token, "", initializeBody)
	if init.status != http.StatusOK || init.session == "" || !strings.Contains(init.body, `"protocolVersion":"2025-11-25"`) {
		t.Fatalf("initialize = %d session %q body %s", init.status, init.session, init.body)
	}
	if done := rpc(t, ctx, in.srv.URL+"/mcp", token, init.session, initializedBody); done.status != http.StatusAccepted {
		t.Fatalf("initialized = %d %s", done.status, done.body)
	}
	return init.session
}

func TestPierSessionsSmoke(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	mr := miniredis.RunT(t)
	cfg := "enabled: false\nsessions:\n  store: redis\n  idle_ttl: 30m\n  max_per_binding: 2\n"
	a := newInstance(t, mr, cfg, &acmeGuard{})
	b := newInstance(t, mr, cfg, &acmeGuard{})

	// Initialise on A, call on B: B rehydrates the session from Redis.
	sid := openSession(t, ctx, a, "oas_at_test")
	call := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_test", sid, callBody)
	if call.status != http.StatusOK || !strings.Contains(call.body, `\"principal\":7`) || strings.Contains(call.body, `"isError":true`) {
		t.Fatalf("call on B = %d %s", call.status, call.body)
	}
	keys := mr.Keys()
	if !containsPrefix(keys, "acme:mcp:s:") || !containsPrefix(keys, "acme:mcp:b:oauth:c1:7") {
		t.Fatalf("redis keys = %v, want acme:mcp:s:<id> and acme:mcp:b:oauth:c1:7", keys)
	}

	// Another binding presenting the same id is refused.
	other := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_other", sid, callBody)
	if other.status != http.StatusForbidden || !strings.Contains(other.body, "session_binding") {
		t.Fatalf("other binding = %d %s", other.status, other.body)
	}

	// DELETE on B ends the session everywhere: A answers 404 next.
	del, err := http.NewRequestWithContext(ctx, http.MethodDelete, b.srv.URL+"/mcp", nil)
	if err != nil {
		t.Fatal(err)
	}
	del.Header.Set("Authorization", "Bearer oas_at_test")
	del.Header.Set("Mcp-Session-Id", sid)
	resp, err := http.DefaultClient.Do(del)
	if err != nil {
		t.Fatal(err)
	}
	_ = resp.Body.Close()
	if resp.StatusCode != http.StatusNoContent {
		t.Fatalf("DELETE on B = %d", resp.StatusCode)
	}
	if gone := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", sid, callBody); gone.status != http.StatusNotFound {
		t.Fatalf("call on A after DELETE = %d %s", gone.status, gone.body)
	}

	// An idle session expires: after the TTL the id is unknown.
	expiring := openSession(t, ctx, a, "oas_at_test")
	mr.FastForward(31 * time.Minute)
	if expired := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", expiring, callBody); expired.status != http.StatusNotFound {
		t.Fatalf("call after the idle TTL = %d %s", expired.status, expired.body)
	}

	// max_per_binding + 1 initialisations answer 429.
	openSession(t, ctx, a, "oas_at_test")
	openSession(t, ctx, b, "oas_at_test")
	third := rpc(t, ctx, a.srv.URL+"/mcp", "oas_at_test", "", initializeBody)
	if third.status != http.StatusTooManyRequests || !strings.Contains(third.body, "too_many_sessions") {
		t.Fatalf("third session = %d %s", third.status, third.body)
	}
}

func containsPrefix(keys []string, prefix string) bool {
	for _, k := range keys {
		if strings.HasPrefix(k, prefix) {
			return true
		}
	}
	return false
}

func TestPierFanOutSmoke(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	mr := miniredis.RunT(t)
	guard := &acmeGuard{}
	cfg := "enabled: false\nsessions:\n  store: redis\n"
	a := newInstance(t, mr, cfg, guard)
	b := newInstance(t, mr, cfg, guard)

	changed := make(chan struct{}, 4)
	client := mcp.NewClient(&mcp.Implementation{Name: "acme-test", Version: "1"}, &mcp.ClientOptions{
		ToolListChangedHandler: func(context.Context, *mcp.ToolListChangedRequest) {
			select {
			case changed <- struct{}{}:
			default:
			}
		},
	})
	transport := &mcp.StreamableClientTransport{
		Endpoint:   b.srv.URL + "/mcp",
		HTTPClient: &http.Client{Transport: bearerTransport{token: "oas_at_test"}},
	}
	cs, err := client.Connect(ctx, transport, &mcp.ClientSessionOptions{ProtocolVersion: "2025-11-25"})
	if err != nil {
		t.Fatalf("connect: %v", err)
	}
	defer cs.Close()
	list, err := cs.ListTools(ctx, nil)
	if err != nil || len(list.Tools) != 1 {
		t.Fatalf("tools/list = %+v, %v", list, err)
	}

	// The guard now denies the tool; a permissions event fired on A reaches
	// the session on B as notifications/tools/list_changed.
	guard.deny.Store(true)
	if err := a.app.Events.Fire(ctx, &pact.MCPSubjectChanged{Guard: "acme", UserID: 7, Kind: pact.MCPSubjectPermissions}); err != nil {
		t.Fatal(err)
	}
	select {
	case <-changed:
	case <-time.After(10 * time.Second):
		t.Fatal("no tools/list_changed within 10 s")
	}
	list, err = cs.ListTools(ctx, nil)
	if err != nil || len(list.Tools) != 0 {
		t.Fatalf("tools/list after the permission change = %+v, %v", list, err)
	}

	// A revocation event ends the session: the next call is refused.
	if err := a.app.Events.Fire(ctx, &pact.MCPSubjectChanged{Guard: "acme", UserID: 7, Kind: pact.MCPSubjectRevoked}); err != nil {
		t.Fatal(err)
	}
	deadline := time.Now().Add(10 * time.Second)
	for {
		if _, err := cs.ListTools(ctx, nil); err != nil {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the session survived the revocation event")
		}
		time.Sleep(50 * time.Millisecond)
	}
	if raw := rpc(t, ctx, b.srv.URL+"/mcp", "oas_at_test", cs.ID(), callBody); raw.status != http.StatusNotFound {
		t.Fatalf("call after revocation = %d %s", raw.status, raw.body)
	}
}