gitea-mcp/internal/tools/strict_args_test.go

package tools_test

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"

	"git.d-ma.be/mathias/gitea-mcp/internal/allowlist"
	"git.d-ma.be/mathias/gitea-mcp/internal/gitea"
	"git.d-ma.be/mathias/gitea-mcp/internal/tools"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// countingServer fails nothing by itself; it records how many requests reached
// it so a test can assert that a rejected call never touched Gitea.
func countingServer(t *testing.T) (*httptest.Server, *atomic.Int32) {
	t.Helper()
	var n atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		n.Add(1)
		w.WriteHeader(http.StatusTeapot)
	}))
	t.Cleanup(srv.Close)
	return srv, &n
}

// #65 (1): encoding/json matches keys case-insensitively and the last one wins,
// so a client-side check reading `branch` could see a different value from the
// one the server acts on. Any two keys equal under strings.EqualFold, and any
// key that differs from a schema field only in case, are rejected before a
// request is made.
func TestParseArgsRejectsCaseVariantKeys(t *testing.T) {
	cases := map[string]string{
		"issue repro: branch + Branch": `{"owner":"probe","repo":"t","name":"t","path":"case8.txt","content":"x","message":"m","branch":"feat/y8","Branch":"main"}`,
		"exact duplicate key":          `{"owner":"probe","repo":"t","path":"p","content":"x","message":"m","branch":"feat/y8","branch":"main"}`,
		"case-only schema mismatch":    `{"owner":"probe","repo":"t","path":"p","content":"x","message":"m","BRANCH":"main"}`,
		"nested object duplicate":      `{"owner":"probe","repo":"t","path":"p","content":"x","message":"m","branch":"feat/y8","extra":{"a":1,"A":2}}`,
	}
	for name, raw := range cases {
		t.Run(name, func(t *testing.T) {
			srv, n := countingServer(t)
			tool := tools.NewFileWriteBranch(gitea.NewClient(srv.URL, "tok"), allowlist.New([]string{"probe"}))
			_, err := tool.Call(context.Background(), json.RawMessage(raw))
			require.Error(t, err)
			assert.ErrorIs(t, err, gitea.ErrValidation)
			assert.Zero(t, n.Load(), "no request may reach gitea")
		})
	}
}

// #65 (2): `refs/heads/main` as a branch makes Gitea create
// refs/heads/refs/heads/main, after which every API write to main 500s.
// Branch names that can create such refs are rejected before any request.
func TestBranchNameValidationRejectsBeforeRequest(t *testing.T) {
	bad := []string{
		"refs/heads/main", "refs/tags/v1", "a..b", "feat x", "feat\tx", "feat/",
		"/feat", "-feat", ".feat", "feat//x", "feat/.x", "feat.lock", "feat.", "HEAD", "feat@{1}", "",
	}
	for _, b := range bad {
		t.Run("file_write_branch "+b, func(t *testing.T) {
			srv, n := countingServer(t)
			tool := tools.NewFileWriteBranch(gitea.NewClient(srv.URL, "tok"), allowlist.New([]string{"probe"}))
			args, _ := json.Marshal(map[string]string{
				"owner": "probe", "repo": "t", "path": "p", "content": "x", "message": "m", "branch": b,
			})
			_, err := tool.Call(context.Background(), args)
			require.Error(t, err)
			assert.ErrorIs(t, err, gitea.ErrValidation)
			assert.Zero(t, n.Load(), "no request may reach gitea")
		})
		if b == "" {
			continue // tbd_ship derives a branch when empty
		}
		t.Run("tbd_ship "+b, func(t *testing.T) {
			srv, n := countingServer(t)
			tool := tools.NewTBDShip(gitea.NewClient(srv.URL, "tok"), allowlist.New([]string{"probe"}))
			args, _ := json.Marshal(map[string]string{
				"owner": "probe", "repo": "t", "path": "p", "content": "x", "message": "m", "branch": b,
			})
			_, err := tool.Call(context.Background(), args)
			require.Error(t, err)
			assert.ErrorIs(t, err, gitea.ErrValidation)
			assert.Zero(t, n.Load(), "no request may reach gitea")
		})
	}
}

// tbd_ship's base is the branch-protection probe key: `refs/heads/main` would
// read as unprotected, so it is held to the same rule as the head branch.
func TestTBDShipRejectsRefsPrefixedBase(t *testing.T) {
	srv, n := countingServer(t)
	tool := tools.NewTBDShip(gitea.NewClient(srv.URL, "tok"), allowlist.New([]string{"probe"}))
	_, err := tool.Call(context.Background(), json.RawMessage(
		`{"owner":"probe","repo":"t","path":"p","content":"x","message":"m","base":"refs/heads/main"}`))
	require.Error(t, err)
	assert.ErrorIs(t, err, gitea.ErrValidation)
	assert.Zero(t, n.Load(), "no request may reach gitea")
}

func TestBranchNameValidationAcceptsOrdinaryNames(t *testing.T) {
	for _, b := range []string{"main", "feat/y8", "tbd/fix-abc-1234abcd", "hold/mail-chat-infra-21", "v1.2_x"} {
		t.Run(b, func(t *testing.T) {
			assert.NoError(t, tools.ValidateBranchName(b))
		})
	}
}