test(19.7-09): named tests for every framework and admin SPA fix of plans 06 and 07

- pier: TestSessionReserve (A-WR-03), TestLockScope (A-WR-07),
  TestScopeWidening (A-WR-04), TestStatelessFanOut (A-WR-05, A-WR-06),
  TestPersonalTokenBinding (A-WR-09), TestDispatchHardening (A-WR-01,
  A-WR-02, A-WR-08, A-IN-04, A-IN-05, A-IN-08), TestCatalogPathFields
  (A-WR-02, A-IN-07), TestPromptWriteCeiling (C-IN-01),
  TestTokenScopedRevoke (B-WR-04, C-IN-02), TestResolveClientAddress
  (B-IN-04); test seam WrapStore
- pact TestFrontendTokenScopes, towel TestClientIP, cabana
  TestTokenScopeNameParity
- cabana TestAccountContributionURL (B-WR-03), TestAccountContributionsIsolation
  (B-WR-06, B-IN-03), TestRecordActionPromptLocalized (B-IN-02), and the
  Postgres TestAccountTokenRevokeEvent (B-WR-04) and
  TestResolveBackendTokenRemote (B-IN-04)
- vitest: hotkeysEditing.test.ts (B-WR-01, B-WR-02) and the extended
  AccountContributions.test.ts (B-WR-03, B-WR-06, B-IN-01)
- check-phase19.7.sh requires every name (GO_TESTS, GO_TAGGED_TESTS,
  DOCKER_TESTS, ADMIN_FILES) and scans the new files (PHASE_TEST_FILES)
This commit is contained in:
2026-10-23 18:36:16 +02:00
parent fda46466c7
commit ccf77b9103
20 changed files with 2010 additions and 1 deletions

View File

@@ -0,0 +1,65 @@
package pier_test
import (
"net/http"
"strings"
"testing"
"git.golem15.com/golem15/summercms/modules/pact"
)
type optionalPathInput struct {
ID string `json:"id,omitempty" path:"id"`
}
type requiredPathInput struct {
ID string `json:"id" path:"id"`
}
type pageFields struct {
Title string `json:"title"`
}
type embeddedInput struct {
pageFields
ID string `json:"id" path:"id"`
}
// TestCatalogPathFields pins plan 06's boot rules for tool inputs (the boot
// half of A-WR-02, A-IN-07): a path field marked omitempty fails boot and
// the error names the field, the same field without it compiles, and an
// embedded struct (whose promoted fields would be body fields) fails boot
// on a GET tool and compiles on a POST tool.
func TestCatalogPathFields(t *testing.T) {
route := func(method string) testRoute { return guarded(method, "/fields/{id}") }
tool := func(method string, input any) pact.MCPTool {
return pact.MCPTool{Name: "f", Description: "Fields.", Method: method, Path: "/fields/{id}", Input: input}
}
mount := func(method string, input any) error {
_, err := catalogMount(t, routePlugin{id: "test.fields", routes: []testRoute{route(method)}, tools: []pact.MCPTool{tool(method, input)}})
return err
}
t.Run("an omitempty path field fails boot naming the field", func(t *testing.T) {
err := mount(http.MethodGet, optionalPathInput{})
if err == nil || !strings.Contains(err.Error(), `path field "id" must be required`) {
t.Fatalf("err = %v", err)
}
})
t.Run("a required path field compiles", func(t *testing.T) {
if err := mount(http.MethodGet, requiredPathInput{}); err != nil {
t.Fatal(err)
}
})
t.Run("an embedded struct fails boot on a GET tool", func(t *testing.T) {
err := mount(http.MethodGet, embeddedInput{})
if err == nil || !strings.Contains(err.Error(), `embedded field "pageFields"`) || !strings.Contains(err.Error(), "GET") {
t.Fatalf("err = %v", err)
}
})
t.Run("an embedded struct compiles on a POST tool", func(t *testing.T) {
if err := mount(http.MethodPost, embeddedInput{}); err != nil {
t.Fatal(err)
}
})
}

View File

@@ -0,0 +1,207 @@
package pier_test
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/url"
"strings"
"sync/atomic"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/bouncer"
"git.golem15.com/golem15/summercms/modules/pact"
"github.com/modelcontextprotocol/go-sdk/mcp"
)
// dispatchGuard is a fakeGuard whose MCPDispatch answer is given per
// route method.
type dispatchGuard struct {
*fakeGuard
byMethod func(pact.MCPSubject, string) (pact.MCPDispatch, error)
}
func (g dispatchGuard) MCPDispatch(_ context.Context, s pact.MCPSubject, method string) (pact.MCPDispatch, error) {
return g.byMethod(s, method)
}
type oneSegmentInput struct {
ID string `json:"id" path:"id"`
}
type manySegmentsInput struct {
Rest string `json:"rest" path:"rest"`
}
type numberInput struct {
ID int64 `json:"id" path:"id"`
N int64 `json:"n,omitempty" query:"n"`
}
// TestDispatchHardening pins plan 06's dispatch hardening (A-WR-01, the
// runtime half of A-WR-02, A-WR-08, A-IN-04, A-IN-05, A-IN-08): path values
// the mux never delivers (., .., empty segments, a slash in a single
// segment) are refused before the route; a missing path value is refused;
// a provider error answers a fixed text and its message reaches only the
// log; a large integer keeps its digits in the path and the query; the
// inner request carries the outer Host and forwarding headers but never the
// outer credentials; and a provider whose answer contradicts the tool's
// guard is refused.
func TestDispatchHardening(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
var hits atomic.Int64
seen := func(w http.ResponseWriter, r *http.Request) {
hits.Add(1)
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"id": r.PathValue("id"),
"rest": r.PathValue("rest"),
"n": r.URL.Query().Get("n"),
"host": r.Host,
"proto": r.Header.Get("X-Forwarded-Proto"),
"fhost": r.Header.Get("X-Forwarded-Host"),
"auth": r.Header.Get("Authorization"),
"cookie": r.Header.Get("Cookie"),
})
}
hard := routePlugin{id: "test.hard",
routes: []testRoute{
{method: http.MethodPost, path: "/hard/one/{id}", mw: []string{"acme.auth"}, h: seen},
{method: http.MethodGet, path: "/hard/many/{rest...}", mw: []string{"acme.auth"}, h: seen},
{method: http.MethodGet, path: "/hard/num/{id}", mw: []string{"acme.auth"}, h: seen},
},
tools: []pact.MCPTool{
{Name: "one", Method: "POST", Path: "/hard/one/{id}", Input: oneSegmentInput{}},
{Name: "many", Method: "GET", Path: "/hard/many/{rest...}", Input: manySegmentsInput{}},
{Name: "num", Method: "GET", Path: "/hard/num/{id}", Input: numberInput{}},
}}
// One frontend guard provider (an OAuth grant acts under exactly one);
// its dispatch answer depends on the route's method: PUT fails like a
// database error, PATCH names the backend guard, DELETE answers a
// backend principal.
pgxLike := "ERROR: relation \"backend_users\" does not exist (SQLSTATE 42P01)"
guard := dispatchGuard{fakeGuard: &fakeGuard{}}
guard.byMethod = func(s pact.MCPSubject, method string) (pact.MCPDispatch, error) {
switch method {
case http.MethodPut:
return pact.MCPDispatch{}, errors.New(pgxLike)
case http.MethodPatch:
return pact.MCPDispatch{Principal: &bouncer.Principal{ID: s.UserID}, Guards: []string{"backend"}}, nil
case http.MethodDelete:
return pact.MCPDispatch{Principal: &bouncer.Principal{ID: s.UserID, Backend: true}, Guards: []string{"acme"}}, nil
}
return pact.MCPDispatch{Principal: &bouncer.Principal{ID: s.UserID}, Guards: []string{"acme"}}, nil
}
hard.routes = append(hard.routes,
testRoute{method: http.MethodPut, path: "/hard/broken", mw: []string{"acme.auth"}, h: seen},
testRoute{method: http.MethodPatch, path: "/hard/claims-backend", mw: []string{"acme.auth"}, h: seen},
testRoute{method: http.MethodDelete, path: "/hard/backend-principal", mw: []string{"acme.auth"}, h: seen},
)
hard.tools = append(hard.tools,
pact.MCPTool{Name: "broken", Method: "PUT", Path: "/hard/broken", Input: struct{}{}},
pact.MCPTool{Name: "claims_backend", Method: "PATCH", Path: "/hard/claims-backend", Input: struct{}{}},
pact.MCPTool{Name: "backend_principal", Method: "DELETE", Path: "/hard/backend-principal", Input: struct{}{}},
)
app := newApp(t, appSpec{guards: []pact.MCPGuard{guard}})
in := startInstance(t, app, hard)
t.Cleanup(in.pier().StopReaper)
outer, err := url.Parse(in.srv.URL)
if err != nil {
t.Fatal(err)
}
logs := captureLogs(t)
cs := sdkClient(t, ctx, in.url(), "oas_at_test", clientOpts{headers: map[string]string{
"Cookie": "summer_session=outer",
"X-Forwarded-Proto": "https",
"X-Forwarded-Host": "site.example",
}})
call := func(name string, args map[string]any) *mcp.CallToolResult {
t.Helper()
res, err := cs.CallTool(ctx, &mcp.CallToolParams{Name: name, Arguments: args})
if err != nil {
t.Fatalf("%s: %v", name, err)
}
return res
}
refused := func(t *testing.T, name string, args map[string]any, want string) {
t.Helper()
before := hits.Load()
res := call(name, args)
if !res.IsError || !strings.Contains(resultText(t, res), want) {
t.Fatalf("%s %v = %+v, want a tool error with %q", name, args, res, want)
}
if hits.Load() != before {
t.Fatalf("%s %v reached the route", name, args)
}
}
t.Run("path values the mux never delivers are refused", func(t *testing.T) {
for _, v := range []string{"..", "."} {
refused(t, "hard_one", map[string]any{"id": v}, "must not contain empty, . or .. segments")
}
refused(t, "hard_one", map[string]any{"id": ""}, "one non-empty path segment")
refused(t, "hard_one", map[string]any{"id": "a/b"}, "one non-empty path segment")
for _, v := range []string{"a/../b", "a//b", "./a", "a/.."} {
refused(t, "hard_many", map[string]any{"rest": v}, "must not contain empty, . or .. segments")
}
if res := call("hard_many", map[string]any{"rest": "a/b"}); res.IsError || structured(t, res)["rest"] != "a/b" {
t.Fatalf("a/b on a {rest...} route = %+v", res)
}
})
t.Run("a missing path value is refused", func(t *testing.T) {
refused(t, "hard_one", map[string]any{}, "invalid arguments")
})
t.Run("a provider error answers the fixed text and is only logged", func(t *testing.T) {
before := hits.Load()
res := call("hard_broken", nil)
if !res.IsError || resultText(t, res) != "pier: the tool could not be run for this caller" {
t.Fatalf("provider error = %+v %q", res, resultText(t, res))
}
if strings.Contains(resultText(t, res), "SQLSTATE") {
t.Fatal("the provider's message reached the agent")
}
if hits.Load() != before {
t.Fatal("a failed dispatch reached the route")
}
if !strings.Contains(logs.String(), "42P01") || !strings.Contains(logs.String(), "guard dispatch failed") {
t.Fatalf("the provider's message is not in the log: %s", logs.String())
}
})
t.Run("a large integer keeps its digits in the path and the query", func(t *testing.T) {
res := cs.CallTool
raw := json.RawMessage(`{"id":9007199254740993,"n":9007199254740993}`)
out, err := res(ctx, &mcp.CallToolParams{Name: "hard_num", Arguments: raw})
if err != nil {
t.Fatal(err)
}
got := structured(t, out)
if out.IsError || got["id"] != "9007199254740993" || got["n"] != "9007199254740993" {
t.Fatalf("large integer = %+v", got)
}
})
t.Run("the inner request has the outer Host and forwarding headers, never credentials", func(t *testing.T) {
got := structured(t, call("hard_num", map[string]any{"id": 5}))
if got["host"] != outer.Host || got["proto"] != "https" || got["fhost"] != "site.example" {
t.Fatalf("inner request host %v proto %v forwarded host %v, want %s https site.example", got["host"], got["proto"], got["fhost"], outer.Host)
}
if got["auth"] != "" || got["cookie"] != "" {
t.Fatalf("outer credentials reached the route: auth %q cookie %q", got["auth"], got["cookie"])
}
})
t.Run("a dispatch contradicting the tool's guard is refused", func(t *testing.T) {
for _, name := range []string{"hard_claims_backend", "hard_backend_principal"} {
refused(t, name, nil, "pier: the tool could not be run for this caller")
}
if !strings.Contains(logs.String(), "contradicts the tool's guard") {
t.Fatalf("no contradiction line: %s", logs.String())
}
})
}

View File

@@ -85,3 +85,7 @@ func (p *Pier) SessionDone(sid string) (<-chan struct{}, bool) {
}()
return done, true
}
// WrapStore replaces the session store with wrap(current store), so a test
// can count calls or hide the store's optional capabilities.
func (p *Pier) WrapStore(wrap func(Store) Store) { p.store = wrap(p.store) }

View File

@@ -0,0 +1,161 @@
package pier_test
import (
"bytes"
"context"
"io"
"net/http"
"sync/atomic"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/pact"
"github.com/modelcontextprotocol/go-sdk/mcp"
)
// blockingGuard is the acme guard whose MCPAllows blocks for user 8 while
// block is set, until release closes; entered is signalled once per block.
type blockingGuard struct {
acmeGuard
block atomic.Bool
entered chan struct{}
release chan struct{}
}
func (g *blockingGuard) MCPAllows(ctx context.Context, s pact.MCPSubject, _ pact.MCPRequirements, _ []string) (bool, error) {
if s.UserID == 8 && g.block.Load() {
select {
case g.entered <- struct{}{}:
default:
}
select {
case <-g.release:
case <-ctx.Done():
}
}
return true, nil
}
// rawStatus sends one request with a bearer, an optional session id and a
// protocol version, and answers the status (safe from any goroutine).
func rawStatus(ctx context.Context, method, url, token, sid, version, body string) (int, error) {
var rd io.Reader
if body != "" {
rd = bytes.NewBufferString(body)
}
req, err := http.NewRequestWithContext(ctx, method, url, rd)
if err != nil {
return 0, 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", version)
if sid != "" {
req.Header.Set("Mcp-Session-Id", sid)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return 0, err
}
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<16))
_ = resp.Body.Close()
return resp.StatusCode, nil
}
// TestLockScope pins plan 06's lock scope (A-WR-07): while one client's
// initialize is blocked in a guard check (as a database query would block
// it), another client's call on an existing session, a DELETE and a
// stateless request each finish promptly, because no router-wide lock is
// held across guard I/O.
func TestLockScope(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
guard := &blockingGuard{entered: make(chan struct{}, 1), release: make(chan struct{})}
in := startInstance(t, newApp(t, appSpec{
mcp: "enabled: false\nsessions:\n store: memory\n",
guards: []pact.MCPGuard{guard},
}), acmePlugin{tools: []pact.MCPTool{echoTool(), showTool()}})
t.Cleanup(in.pier().StopReaper)
active := openSession(t, ctx, in, "oas_at_test")
doomed := openSession(t, ctx, in, "oas_at_test")
guard.block.Store(true)
blocked := make(chan int, 1)
go func() {
status, _ := initStatus(ctx, in.url(), "oas_at_other", initializeBody)
blocked <- status
}()
select {
case <-guard.entered:
case <-time.After(10 * time.Second):
t.Fatal("the blocked initialize never reached the guard")
}
within := func(name string, fn func() error) {
t.Helper()
done := make(chan error, 1)
go func() { done <- fn() }()
select {
case err := <-done:
if err != nil {
t.Fatalf("%s: %v", name, err)
}
case <-time.After(3 * time.Second):
t.Fatalf("%s waited for the blocked initialize", name)
}
}
statusIs := func(want int, method, sid, version, body string) func() error {
return func() error {
got, err := rawStatus(ctx, method, in.url(), "oas_at_test", sid, version, body)
if err != nil {
return err
}
if got != want {
return &statusError{got: got, want: want}
}
return nil
}
}
t.Run("a call on an existing session", func(t *testing.T) {
within("tools/call", statusIs(http.StatusOK, http.MethodPost, active, "2025-11-25", callBody))
})
t.Run("a DELETE", func(t *testing.T) {
within("DELETE", statusIs(http.StatusNoContent, http.MethodDelete, doomed, "2025-11-25", ""))
})
t.Run("a stateless request", func(t *testing.T) {
within("stateless tools/list", func() error {
client := mcp.NewClient(&mcp.Implementation{Name: "lock-test", Version: "1"}, nil)
transport := &mcp.StreamableClientTransport{Endpoint: in.url(), HTTPClient: &http.Client{Transport: headerTransport{token: "oas_at_test"}}}
cs, err := client.Connect(ctx, transport, nil)
if err != nil {
return err
}
defer cs.Close()
_, err = cs.ListTools(ctx, nil)
return err
})
})
select {
case status := <-blocked:
t.Fatalf("the blocked initialize finished early (%d)", status)
default:
}
close(guard.release)
select {
case status := <-blocked:
if status != http.StatusOK {
t.Fatalf("the released initialize = %d", status)
}
case <-time.After(10 * time.Second):
t.Fatal("the released initialize never finished")
}
}
type statusError struct{ got, want int }
func (e *statusError) Error() string {
return "status " + http.StatusText(e.got) + ", want " + http.StatusText(e.want)
}

View File

@@ -0,0 +1,65 @@
package pier_test
import (
"context"
"net/http"
"strings"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/pact"
)
// TestPersonalTokenBinding pins plan 06's personal-token binding rule
// (A-WR-09): a resolver answering a subject without a TokenID is refused
// with 401 at /mcp (it would put every user of the resolver on one
// binding), and two users' personal tokens get distinct bindings, so one
// cannot use or DELETE the other's session.
func TestPersonalTokenBinding(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
personal := fakePersonalResolver{tokens: map[string]pact.MCPSubject{
"usr_no_token_id": {Guard: "acme", UserID: 31, Scopes: []string{"read", "write"}},
"usr_alice": {Guard: "acme", UserID: 32, TokenID: 501, Scopes: []string{"read", "write"}},
"usr_bob": {Guard: "acme", UserID: 33, TokenID: 502, Scopes: []string{"read", "write"}},
}}
in := startInstance(t, newApp(t, appSpec{
mcp: "enabled: false\nallow_personal_tokens: true\nsessions:\n store: memory\n",
guards: []pact.MCPGuard{&acmeGuard{}},
personal: personal,
}), acmePlugin{tools: []pact.MCPTool{echoTool()}})
t.Cleanup(in.pier().StopReaper)
t.Run("a subject without a TokenID is 401", func(t *testing.T) {
if r := rpc(t, ctx, in.url(), "usr_no_token_id", "", initializeBody); r.status != http.StatusUnauthorized {
t.Fatalf("initialize with a TokenID-less subject = %d %s", r.status, r.body)
}
if _, err := in.pier().Resolve(ctx, "usr_no_token_id"); err == nil {
t.Fatal("Resolve accepted a subject without a TokenID")
}
})
t.Run("two users get distinct bindings", func(t *testing.T) {
alice, err := in.pier().Resolve(ctx, "usr_alice")
if err != nil {
t.Fatal(err)
}
bob, err := in.pier().Resolve(ctx, "usr_bob")
if err != nil {
t.Fatal(err)
}
if alice.Binding != "pat:user:501" || bob.Binding != "pat:user:502" {
t.Fatalf("bindings = %q, %q", alice.Binding, bob.Binding)
}
sid := openSession(t, ctx, in, "usr_alice")
if r := rpc(t, ctx, in.url(), "usr_bob", sid, callBody); r.status != http.StatusForbidden || !strings.Contains(r.body, "session_binding") {
t.Fatalf("bob on alice's session = %d %s", r.status, r.body)
}
if code := deleteSession(t, ctx, in.url(), "usr_bob", sid); code != http.StatusForbidden {
t.Fatalf("bob's DELETE of alice's session = %d", code)
}
if r := rpc(t, ctx, in.url(), "usr_alice", sid, callBody); r.status != http.StatusOK || !strings.Contains(r.body, `\"principal\":32`) {
t.Fatalf("alice's session after bob's attempts = %d %s", r.status, r.body)
}
})
}

View File

@@ -0,0 +1,63 @@
package pier_test
import (
"context"
"slices"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/pact"
"github.com/modelcontextprotocol/go-sdk/mcp"
)
// promptNames lists the prompt names a session sees, sorted.
func promptNames(t *testing.T, ctx context.Context, cs *mcp.ClientSession) []string {
t.Helper()
list, err := cs.ListPrompts(ctx, nil)
if err != nil {
t.Fatalf("prompts/list: %v", err)
}
var out []string
for _, p := range list.Prompts {
out = append(out, p.Name)
}
slices.Sort(out)
return out
}
// TestPromptWriteCeiling pins plan 06's prompt ceiling (C-IN-01, P-28): a
// prompt declaring Writes is listed for a read and write grant and hidden
// from a read-only grant (and cannot be fetched by it), while a prompt
// without Writes is listed for both.
func TestPromptWriteCeiling(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
draft := summarizePrompt()
draft.Name = "draft"
draft.Writes = true
in := startInstance(t, newApp(t, appSpec{guards: []pact.MCPGuard{&acmeGuard{}}}),
acmePlugin{tools: []pact.MCPTool{echoTool()}, prompts: []pact.MCPPrompt{summarizePrompt(), draft}})
t.Cleanup(in.pier().StopReaper)
for _, c := range []struct {
name string
token string
want []string
}{
{"read and write grant", "oas_at_test", []string{"acme_draft", "acme_summarize"}},
{"read-only grant", "oas_at_read", []string{"acme_summarize"}},
} {
t.Run(c.name, func(t *testing.T) {
for _, version := range []string{"", "newest"} {
cs := sdkClient(t, ctx, in.url(), c.token, clientOpts{version: version})
if got := promptNames(t, ctx, cs); !slices.Equal(got, c.want) {
t.Fatalf("version %q: prompts = %v, want %v", version, got, c.want)
}
_, err := cs.GetPrompt(ctx, &mcp.GetPromptParams{Name: "acme_draft", Arguments: map[string]string{"topic": "x"}})
if allowed := slices.Contains(c.want, "acme_draft"); (err == nil) != allowed {
t.Fatalf("version %q: get acme_draft err = %v, allowed %v", version, err, allowed)
}
}
})
}
}

View File

@@ -0,0 +1,194 @@
package pier_test
import (
"bytes"
"context"
"io"
"net/http"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/pact"
"git.golem15.com/golem15/summercms/modules/pier"
)
// initStatus posts one initialize without a session and answers the status
// (safe to call from any goroutine).
func initStatus(ctx context.Context, url, token, body string) (int, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBufferString(body))
if err != nil {
return 0, 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")
resp, err := http.DefaultClient.Do(req)
if err != nil {
return 0, err
}
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<16))
_ = resp.Body.Close()
return resp.StatusCode, nil
}
// burst sends n concurrent initializes for token, spread over the urls, and
// counts the answers by status.
func burst(t *testing.T, ctx context.Context, n int, token string, urls ...string) map[int]int {
t.Helper()
var mu sync.Mutex
got := map[int]int{}
var wg sync.WaitGroup
start := make(chan struct{})
for i := 0; i < n; i++ {
wg.Add(1)
go func(url string) {
defer wg.Done()
<-start
status, err := initStatus(ctx, url, token, initializeBody)
if err != nil {
status = -1
}
mu.Lock()
got[status]++
mu.Unlock()
}(urls[i%len(urls)])
}
close(start)
wg.Wait()
return got
}
// countingStore hides the shipped store's Reserve, Release and Update (it
// embeds only pier.Store) and counts Count and Put.
type countingStore struct {
pier.Store
counts atomic.Int64
puts atomic.Int64
}
func (s *countingStore) Count(ctx context.Context, binding string) (int, error) {
s.counts.Add(1)
return s.Store.Count(ctx, binding)
}
func (s *countingStore) Put(ctx context.Context, id string, rec pier.Record, ttl time.Duration) error {
s.puts.Add(1)
return s.Store.Put(ctx, id, rec, ttl)
}
// TestSessionReserve pins the atomic per-binding reservation of plan 06
// (A-WR-03, P-25): concurrent initializes of one binding open exactly
// mcp.sessions.max_per_binding sessions on the memory store and on a Redis
// store shared by two instances, the rest answer 429; an initialize that
// fails releases its slot; and a store without the reservation falls back
// to Count then Put.
func TestSessionReserve(t *testing.T) {
plugin := acmePlugin{tools: []pact.MCPTool{echoTool()}}
const cap = 3
t.Run("memory store opens exactly max_per_binding of 50 concurrent initializes", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
in := startInstance(t, newApp(t, appSpec{
mcp: "enabled: false\nsessions:\n store: memory\n max_per_binding: 3\n",
guards: []pact.MCPGuard{&acmeGuard{}},
}), plugin)
t.Cleanup(in.pier().StopReaper)
got := burst(t, ctx, 50, "oas_at_test", in.url())
if got[http.StatusOK] != cap || got[http.StatusTooManyRequests] != 50-cap {
t.Fatalf("answers = %v, want %d OK and %d 429", got, cap, 50-cap)
}
if n := in.pier().LocalSessions(); n != cap {
t.Fatalf("local sessions = %d, want %d", n, cap)
}
})
t.Run("redis store opens exactly max_per_binding across two instances", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
a, b, _ := twoInstances(t, appSpec{
mcp: "enabled: false\nsessions:\n store: redis\n max_per_binding: 3\n",
guards: []pact.MCPGuard{&acmeGuard{}},
}, plugin)
t.Cleanup(a.pier().StopReaper)
t.Cleanup(b.pier().StopReaper)
got := burst(t, ctx, 50, "oas_at_test", a.url(), b.url())
if got[http.StatusOK] != cap || got[http.StatusTooManyRequests] != 50-cap {
t.Fatalf("answers = %v, want %d OK and %d 429", got, cap, 50-cap)
}
if n := a.pier().LocalSessions() + b.pier().LocalSessions(); n != cap {
t.Fatalf("local sessions on both instances = %d, want %d", n, cap)
}
})
t.Run("a failed initialize releases its slot", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
in := startInstance(t, newApp(t, appSpec{
mcp: "enabled: false\nsessions:\n store: memory\n max_per_binding: 1\n",
guards: []pact.MCPGuard{&acmeGuard{}},
}), plugin)
t.Cleanup(in.pier().StopReaper)
bad := `{"jsonrpc":"2.0","id":1,"method":"initialize","params":"not an object"}`
// Each one reserves a slot (it gets a session id) and fails in the
// SDK's initialize; with max_per_binding 1, a kept slot would make
// the next one 429.
// The SDK streams the error answer before serveNew releases the
// slot, so each attempt waits for the release (the local copy gone).
released := func(i int) {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for in.pier().LocalSessions() != 0 {
if time.Now().After(deadline) {
t.Fatalf("failed initialize %d kept its local session", i)
}
time.Sleep(5 * time.Millisecond)
}
}
for i := 0; i < 3; i++ {
r := rpc(t, ctx, in.url(), "oas_at_test", "", bad)
if r.status == http.StatusTooManyRequests {
t.Fatalf("failed initialize %d hit the cap: an earlier one kept its slot", i)
}
if r.session == "" || !strings.Contains(r.body, `"error"`) {
t.Fatalf("failed initialize %d = %d session %q %s, want a reserved id and an error", i, r.status, r.session, r.body)
}
released(i)
}
openSession(t, ctx, in, "oas_at_test")
if third := rpc(t, ctx, in.url(), "oas_at_test", "", initializeBody); third.status != http.StatusTooManyRequests {
t.Fatalf("second session over max_per_binding 1 = %d %s", third.status, third.body)
}
})
t.Run("a store without the reservation falls back to Count and Put", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
in := startInstance(t, newApp(t, appSpec{
mcp: "enabled: false\nsessions:\n store: memory\n max_per_binding: 2\n",
guards: []pact.MCPGuard{&acmeGuard{}},
}), plugin)
p := in.pier()
p.StopReaper()
store := &countingStore{}
p.WrapStore(func(s pier.Store) pier.Store {
store.Store = s
return store
})
first := openSession(t, ctx, in, "oas_at_test")
openSession(t, ctx, in, "oas_at_test")
if over := rpc(t, ctx, in.url(), "oas_at_test", "", initializeBody); over.status != http.StatusTooManyRequests {
t.Fatalf("third session = %d %s", over.status, over.body)
}
if store.counts.Load() < 3 || store.puts.Load() < 2 {
t.Fatalf("Count called %d times, Put %d times: the fallback did not run", store.counts.Load(), store.puts.Load())
}
if call := rpc(t, ctx, in.url(), "oas_at_test", first, callBody); call.status != http.StatusOK {
t.Fatalf("call on a fallback session = %d %s", call.status, call.body)
}
})
}

View File

@@ -0,0 +1,173 @@
package pier_test
import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"strings"
"sync"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/pact"
"github.com/modelcontextprotocol/go-sdk/mcp"
)
// listenWatch is a client transport that adds the bearer and closes ended
// when the response body of the client's subscriptions/listen request ends
// (the server closed the stream).
type listenWatch struct {
token string
once sync.Once
ended chan struct{}
}
func (l *listenWatch) RoundTrip(r *http.Request) (*http.Response, error) {
r = r.Clone(r.Context())
r.Header.Set("Authorization", "Bearer "+l.token)
listen := false
if r.Body != nil {
raw, err := io.ReadAll(r.Body)
if err != nil {
return nil, err
}
_ = r.Body.Close()
r.Body = io.NopCloser(bytes.NewReader(raw))
listen = strings.Contains(string(raw), `"subscriptions/listen"`)
}
resp, err := http.DefaultTransport.RoundTrip(r)
if err != nil || !listen {
return resp, err
}
resp.Body = &watchedBody{ReadCloser: resp.Body, done: func() { l.once.Do(func() { close(l.ended) }) }}
return resp, nil
}
type watchedBody struct {
io.ReadCloser
done func()
}
func (b *watchedBody) Read(p []byte) (int, error) {
n, err := b.ReadCloser.Read(p)
if err != nil {
b.done()
}
return n, err
}
// listenClient connects a newest-protocol client with a list_changed
// handler, so it opens a subscriptions/listen stream, and answers the
// channel closed when that stream ends.
func listenClient(t *testing.T, ctx context.Context, url, token string) (*mcp.ClientSession, <-chan struct{}) {
t.Helper()
watch := &listenWatch{token: token, ended: make(chan struct{})}
client := mcp.NewClient(&mcp.Implementation{Name: "listen-test", Version: "1"}, &mcp.ClientOptions{
ToolListChangedHandler: func(context.Context, *mcp.ToolListChangedRequest) {},
})
cs, err := client.Connect(ctx, &mcp.StreamableClientTransport{Endpoint: url, HTTPClient: &http.Client{Transport: watch}}, nil)
if err != nil {
t.Fatalf("connect with %s: %v", token, err)
}
t.Cleanup(func() { _ = cs.Close() })
if cs.ID() != "" {
t.Fatalf("the newest client got a session %q", cs.ID())
}
return cs, watch.ended
}
func waitEnded(t *testing.T, ended <-chan struct{}, within time.Duration, what string) {
t.Helper()
select {
case <-ended:
case <-time.After(within):
t.Fatalf("%s: the listen stream is still open after %s", what, within)
}
}
// TestStatelessFanOut pins plan 06's stateless lifecycle (A-WR-05,
// A-WR-06): an instance that served only stateless requests receives a
// revocation published by the other instance through Redis and evicts its
// entry; a subscriptions/listen stream ends within the re-check interval
// once its bearer stops resolving; and an LRU eviction at capacity closes
// the evicted server's listen stream.
func TestStatelessFanOut(t *testing.T) {
plugin := acmePlugin{tools: []pact.MCPTool{echoTool(), showTool()}}
t.Run("a stateless-only instance applies a revocation from the other instance", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
a, b, _ := twoInstances(t, appSpec{guards: []pact.MCPGuard{&acmeGuard{}}}, plugin)
t.Cleanup(a.pier().StopReaper)
t.Cleanup(b.pier().StopReaper)
cs := sdkClient(t, ctx, b.url(), "oas_at_test", clientOpts{version: "newest"})
listNames(t, ctx, cs)
if n := b.pier().StatelessServers(); n != 1 {
t.Fatalf("B stateless servers = %d, want 1", n)
}
fireSubjectChanged(t, a.app, "acme", 7, pact.MCPSubjectRevoked)
deadline := time.Now().Add(10 * time.Second)
for b.pier().StatelessServers() != 0 {
if time.Now().After(deadline) {
t.Fatal("B never evicted its stateless server: the revocation did not reach it")
}
time.Sleep(20 * time.Millisecond)
}
})
t.Run("a listen stream ends once its bearer stops resolving", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
verifier := newFakeVerifier(map[string]pact.MCPGrant{"oas_at_listen": grant(12, "c7", 12, "read", "write")})
in := startInstance(t, newApp(t, appSpec{guards: []pact.MCPGuard{&acmeGuard{}}, verifier: verifier}), plugin)
t.Cleanup(in.pier().StopReaper)
in.pier().SetRecheck(50 * time.Millisecond)
logs := captureLogs(t)
_, ended := listenClient(t, ctx, in.url(), "oas_at_listen")
select {
case <-ended:
t.Fatal("the listen stream ended while the bearer resolved")
case <-time.After(300 * time.Millisecond):
}
verifier.revoke("oas_at_listen")
waitEnded(t, ended, 3*time.Second, "revoked bearer")
if !strings.Contains(logs.String(), "step=stream_closed") {
t.Fatalf("no stream_closed line: %s", logs.String())
}
})
t.Run("an LRU eviction closes the evicted server's listen stream", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
const capacity = 16 // max_per_binding 1 times the LRU factor
extra := map[string]pact.MCPGrant{"oas_at_first": grant(20, "c20", 20, "read")}
for i := 0; i < capacity; i++ {
extra[fmt.Sprintf("oas_at_lru_%d", i)] = grant(uint(100+i), fmt.Sprintf("lru%d", i), uint(100+i), "read")
}
in := startInstance(t, newApp(t, appSpec{
mcp: "enabled: false\nsessions:\n max_per_binding: 1\n",
guards: []pact.MCPGuard{&acmeGuard{}},
verifier: newFakeVerifier(extra),
}), plugin)
t.Cleanup(in.pier().StopReaper)
_, ended := listenClient(t, ctx, in.url(), "oas_at_first")
for i := 0; i < capacity-1; i++ {
listNames(t, ctx, sdkClient(t, ctx, in.url(), fmt.Sprintf("oas_at_lru_%d", i), clientOpts{version: "newest"}))
}
select {
case <-ended:
t.Fatal("the listen stream ended below capacity")
case <-time.After(200 * time.Millisecond):
}
if n := in.pier().StatelessServers(); n != capacity {
t.Fatalf("stateless servers at capacity = %d, want %d", n, capacity)
}
listNames(t, ctx, sdkClient(t, ctx, in.url(), fmt.Sprintf("oas_at_lru_%d", capacity-1), clientOpts{version: "newest"}))
waitEnded(t, ended, 5*time.Second, "LRU eviction")
if n := in.pier().StatelessServers(); n != capacity {
t.Fatalf("stateless servers after the eviction = %d, want %d", n, capacity)
}
})
}

View File

@@ -0,0 +1,174 @@
package pier_test
import (
"context"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/pact"
"git.golem15.com/golem15/summercms/modules/party"
"git.golem15.com/golem15/summercms/modules/pier"
"git.golem15.com/golem15/summercms/modules/surf"
"git.golem15.com/golem15/summercms/modules/towel"
)
// TestTokenScopedRevoke pins plan 07's narrowed revocation (B-WR-04, the
// framework half of C-IN-02, P-31) across two instances on one Redis
// store: a revoked event fired on one instance with a TokenID closes, on
// the other, only the session opened with that personal token (another
// token's session and an OAuth session of the same user, even one whose
// grant carries the same token id, stay); with a ClientID only that
// client's OAuth sessions close; with neither every session of the user
// closes.
func TestTokenScopedRevoke(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
personal := fakePersonalResolver{tokens: map[string]pact.MCPSubject{
"usr_first": {Guard: "acme", UserID: 40, TokenID: 601, Scopes: []string{"read", "write"}},
"usr_second": {Guard: "acme", UserID: 40, TokenID: 602, Scopes: []string{"read", "write"}},
}}
verifier := newFakeVerifier(map[string]pact.MCPGrant{
// The same token id as usr_first: a TokenID event must not reach it.
"oas_at_u40_ca": grant(40, "ca", 601, "read", "write"),
"oas_at_u40_cb": grant(40, "cb", 702, "read", "write"),
})
a, b, _ := twoInstances(t, appSpec{
mcp: "enabled: false\nallow_personal_tokens: true\nsessions:\n store: redis\n",
guards: []pact.MCPGuard{&acmeGuard{}},
verifier: verifier,
personal: personal,
}, acmePlugin{tools: []pact.MCPTool{echoTool()}})
t.Cleanup(a.pier().StopReaper)
t.Cleanup(b.pier().StopReaper)
sessions := map[string]string{}
for _, tok := range []string{"usr_first", "usr_second", "oas_at_u40_ca", "oas_at_u40_cb"} {
sessions[tok] = openSession(t, ctx, b, tok)
}
status := func(tok string) int {
return rpc(t, ctx, b.url(), tok, sessions[tok], callBody).status
}
waitClosed := func(tok string) {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for status(tok) != http.StatusNotFound {
if time.Now().After(deadline) {
t.Fatalf("the session of %s survived the event", tok)
}
time.Sleep(20 * time.Millisecond)
}
}
alive := func(toks ...string) {
t.Helper()
for _, tok := range toks {
if got := status(tok); got != http.StatusOK {
t.Fatalf("the session of %s = %d after an event that does not name it", tok, got)
}
}
}
fire := func(ev *pact.MCPSubjectChanged) {
t.Helper()
ev.Guard, ev.UserID, ev.Kind = "acme", 40, pact.MCPSubjectRevoked
if err := a.app.Events.Fire(context.Background(), ev); err != nil {
t.Fatal(err)
}
}
t.Run("a TokenID closes only that personal token's session", func(t *testing.T) {
fire(&pact.MCPSubjectChanged{TokenID: 601})
waitClosed("usr_first")
alive("usr_second", "oas_at_u40_ca", "oas_at_u40_cb")
})
t.Run("a ClientID closes only that client's OAuth sessions", func(t *testing.T) {
fire(&pact.MCPSubjectChanged{ClientID: "ca"})
waitClosed("oas_at_u40_ca")
alive("usr_second", "oas_at_u40_cb")
})
t.Run("neither closes every session of the user", func(t *testing.T) {
fire(&pact.MCPSubjectChanged{})
waitClosed("usr_second")
waitClosed("oas_at_u40_cb")
})
}
// addressVerifier is the fake verifier that records the client address it
// finds on the resolve context.
type addressVerifier struct {
*fakeVerifier
mu sync.Mutex
seen []string
}
func (v *addressVerifier) MCPVerifyToken(ctx context.Context, raw string) (pact.MCPGrant, error) {
ip, _ := towel.ClientIP(ctx)
v.mu.Lock()
v.seen = append(v.seen, ip)
v.mu.Unlock()
return v.fakeVerifier.MCPVerifyToken(ctx, raw)
}
func (v *addressVerifier) last() string {
v.mu.Lock()
defer v.mu.Unlock()
if len(v.seen) == 0 {
return ""
}
return v.seen[len(v.seen)-1]
}
// proxyHost is a test host that also answers the client address the way
// surf does behind a trusted proxy (here: the X-Forwarded-For header).
type proxyHost struct {
*testHost
}
func (proxyHost) ClientIP(r *http.Request) string { return r.Header.Get("X-Forwarded-For") }
// TestResolveClientAddress pins plan 07's client address on the resolve
// context (B-IN-04, P-32): a token resolver sees towel.ClientIP equal to
// the host part of RemoteAddr, or, through a host that implements
// ClientIP (surf's trusted-proxy answer), to that host's answer.
func TestResolveClientAddress(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
plugins := []party.Plugin{acmePlugin{tools: []pact.MCPTool{echoTool()}}}
t.Run("RemoteAddr host", func(t *testing.T) {
verifier := &addressVerifier{fakeVerifier: newFakeVerifier(nil)}
in := startInstance(t, newApp(t, appSpec{guards: []pact.MCPGuard{&acmeGuard{}}, verifier: verifier}), plugins...)
t.Cleanup(in.pier().StopReaper)
openSession(t, ctx, in, "oas_at_test")
if got := verifier.last(); got != "127.0.0.1" {
t.Fatalf("resolver saw client address %q, want 127.0.0.1", got)
}
})
t.Run("a host answering ClientIP", func(t *testing.T) {
verifier := &addressVerifier{fakeVerifier: newFakeVerifier(nil)}
app := newApp(t, appSpec{guards: []pact.MCPGuard{&acmeGuard{}}, verifier: verifier})
router, err := surf.BuildRouter(app, plugins)
if err != nil {
t.Fatal(err)
}
if err := app.Config.Set("mcp.enabled", true); err != nil {
t.Fatal(err)
}
host := proxyHost{testHost: &testHost{r: router, handlers: map[string]http.Handler{}}}
if err := pier.Mount(host, app, plugins); err != nil {
t.Fatal(err)
}
if p, ok := app.Lookup[*pier.Pier](); ok {
t.Cleanup(p.StopReaper)
}
srv := httptest.NewServer(host)
t.Cleanup(srv.Close)
cs := sdkClient(t, ctx, srv.URL+"/mcp", "oas_at_test", clientOpts{headers: map[string]string{"X-Forwarded-For": "203.0.113.7"}})
listNames(t, ctx, cs)
if got := verifier.last(); got != "203.0.113.7" {
t.Fatalf("resolver saw client address %q, want the host's answer 203.0.113.7", got)
}
})
}

View File

@@ -0,0 +1,93 @@
package pier_test
import (
"context"
"net/http"
"slices"
"testing"
"time"
"git.golem15.com/golem15/summercms/modules/pact"
)
// setGrant replaces (or adds) the grant a fake verifier answers for raw.
func (v *fakeVerifier) setGrant(raw string, g pact.MCPGrant) {
v.mu.Lock()
v.grants[raw] = g
v.mu.Unlock()
}
// TestScopeWidening pins plan 06's re-filtering of a returning binding
// (A-WR-04, P-26): a binding first seen with read and then with read write
// lists the write tool on its next request, on the stateless and on the
// stateful path; a backend subject added to the binding (the grant gains
// backend, the user is linked) adds the backend tools without any fan-out
// event; and a narrowed grant hides the tools again.
func TestScopeWidening(t *testing.T) {
links := newFakeLinks()
links.set(11, 42)
verifier := newFakeVerifier(nil)
front := &fakeGuard{}
back := &fakeGuard{name: "backend", tokenMW: "backend"}
acme := routePlugin{id: "test.acme",
routes: []testRoute{guarded(http.MethodPost, "/acme/echo/{id}"), guarded(http.MethodGet, "/acme/items/{id}")},
tools: []pact.MCPTool{
{Name: "echo", Method: "POST", Path: "/acme/echo/{id}", Input: echoInput{}},
{Name: "show", Method: "GET", Path: "/acme/items/{id}", Input: showInput{}},
}}
office := routePlugin{id: "test.office", guard: "backend", middleware: "backend",
routes: []testRoute{{method: http.MethodGet, path: "/office/posts", mw: []string{"backend"}, h: noContent}},
tools: []pact.MCPTool{{Name: "posts", Method: "GET", Path: "/office/posts", Input: struct{}{}}}}
in := startInstance(t, newApp(t, appSpec{
mcp: "enabled: false\nsessions:\n store: memory\n",
guards: []pact.MCPGuard{front, back},
verifier: verifier,
links: links,
}), acme, office)
t.Cleanup(in.pier().StopReaper)
readOnly := []string{"acme_show"}
readWrite := []string{"acme_echo", "acme_show"}
withBackend := []string{"acme_echo", "acme_show", "office_posts"}
for _, path := range []struct {
name string
version string
token string
client string
}{
{"stateless", "newest", "oas_at_wide_stateless", "c5"},
{"stateful", "", "oas_at_wide_stateful", "c6"},
} {
t.Run(path.name, func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
verifier.setGrant(path.token, grant(11, path.client, 11, "read"))
cs := sdkClient(t, ctx, in.url(), path.token, clientOpts{version: path.version})
if (cs.ID() == "") != (path.name == "stateless") {
t.Fatalf("%s client session id %q", path.name, cs.ID())
}
if got := listNames(t, ctx, cs); !slices.Equal(got, readOnly) {
t.Fatalf("read grant lists %v", got)
}
verifier.setGrant(path.token, grant(11, path.client, 11, "read", "write"))
if got := listNames(t, ctx, cs); !slices.Equal(got, readWrite) {
t.Fatalf("after widening to read write: %v", got)
}
// The same binding returns with the backend scope; the user is
// linked, so a backend subject joins the identity. No event is
// fired: the request itself re-filters the cached server.
verifier.setGrant(path.token, grant(11, path.client, 11, "read", "write", "backend"))
if got := listNames(t, ctx, cs); !slices.Equal(got, withBackend) {
t.Fatalf("after adding the backend subject: %v", got)
}
verifier.setGrant(path.token, grant(11, path.client, 11, "read"))
if got := listNames(t, ctx, cs); !slices.Equal(got, readOnly) {
t.Fatalf("after narrowing to read: %v", got)
}
})
}
}