alihan/kubernetes-mcp-server
1
0
Fork 0
mirror of https://github.com/containers/kubernetes-mcp-server.git synced 2025-10-23 01:22:57 +03:00
Code Issues Packages Projects Releases Wiki Activity

test(mcp): refactor events toolset tests (#328)

Browse Source 
Signed-off-by: Marc Nuri <marc@marcnuri.com>
This commit is contained in:
Marc Nuri
2025-09-17 11:48:35 +02:00
committed by GitHub
parent f496c643e7
commit d6936f42d3
3 changed files with 126 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
pkg/mcp/common_test.go
View File

@@ -20,6 +20,7 @@ import (
"github.com/mark3labs/mcp-go/server"
"github.com/pkg/errors"
"github.com/spf13/afero"
"github.com/stretchr/testify/suite"
"golang.org/x/sync/errgroup"
corev1 "k8s.io/api/core/v1"
rbacv1 "k8s.io/api/rbac/v1"
@@ -418,3 +419,32 @@ func createTestData(ctx context.Context) {
_, _ = kubernetesAdmin.CoreV1().ConfigMaps("default").
Create(ctx, &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: "a-configmap-to-delete"}}, metav1.CreateOptions{})
}
type BaseMcpSuite struct {
suite.Suite
*test.McpClient
mcpServer *Server
Cfg *config.StaticConfig
}
func (s *BaseMcpSuite) SetupTest() {
s.Cfg = config.Default()
s.Cfg.KubeConfig = filepath.Join(s.T().TempDir(), "config")
s.Require().NoError(os.WriteFile(s.Cfg.KubeConfig, envTest.KubeConfig, 0600), "Expected to write kubeconfig")
}
func (s *BaseMcpSuite) TearDownTest() {
if s.McpClient != nil {
s.McpClient.Close()
}
if s.mcpServer != nil {
s.mcpServer.Close()
}
}
func (s *BaseMcpSuite) InitMcpClient() {
var err error
s.mcpServer, err = NewServer(Configuration{StaticConfig: s.Cfg})
s.Require().NoError(err, "Expected no error creating MCP server")
s.McpClient = test.NewMcpClient(s.T(), s.mcpServer.ServeHTTP(nil))
}

33
pkg/mcp/configuration_test.go
View File

@@ -10,42 +10,22 @@ import (
"sigs.k8s.io/yaml"
"github.com/containers/kubernetes-mcp-server/internal/test"
"github.com/containers/kubernetes-mcp-server/pkg/config"
"github.com/containers/kubernetes-mcp-server/pkg/kubernetes"
)
type ConfigurationSuite struct {
suite.Suite
*test.McpClient
mcpServer *Server
Cfg *config.StaticConfig
BaseMcpSuite
}
func (s *ConfigurationSuite) SetupTest() {
s.Cfg = config.Default()
}
func (s *ConfigurationSuite) TearDownTest() {
if s.McpClient != nil {
s.McpClient.Close()
}
if s.mcpServer != nil {
s.mcpServer.Close()
}
}
func (s *ConfigurationSuite) InitMcpClient() {
var err error
s.mcpServer, err = NewServer(Configuration{StaticConfig: s.Cfg})
s.Require().NoError(err, "Expected no error creating MCP server")
s.McpClient = test.NewMcpClient(s.T(), s.mcpServer.ServeHTTP(nil))
}
func (s *ConfigurationSuite) TestConfigurationView() {
// Out of cluster requires kubeconfig
s.BaseMcpSuite.SetupTest()
// Use mock server for predictable kubeconfig content
mockServer := test.NewMockServer()
s.T().Cleanup(mockServer.Close)
s.Cfg.KubeConfig = mockServer.KubeconfigFile(s.T())
}
func (s *ConfigurationSuite) TestConfigurationView() {
s.InitMcpClient()
s.Run("configuration_view", func() {
toolResult, err := s.CallTool("configuration_view", map[string]interface{}{})
@@ -108,6 +88,7 @@ func (s *ConfigurationSuite) TestConfigurationView() {
}
func (s *ConfigurationSuite) TestConfigurationViewInCluster() {
s.Cfg.KubeConfig = "" // Force in-cluster
kubernetes.InClusterConfig = func() (*rest.Config, error) {
return &rest.Config{
Host: "https://kubernetes.default.svc",

117
pkg/mcp/events_test.go
View File

@@ -3,32 +3,34 @@ package mcp
import (
"testing"
"github.com/BurntSushi/toml"
"github.com/mark3labs/mcp-go/mcp"
"github.com/stretchr/testify/suite"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/containers/kubernetes-mcp-server/internal/test"
"github.com/containers/kubernetes-mcp-server/pkg/config"
"k8s.io/client-go/kubernetes"
)
func TestEventsList(t *testing.T) {
testCase(t, func(c *mcpContext) {
c.withEnvTest()
toolResult, err := c.callTool("events_list", map[string]interface{}{})
t.Run("events_list with no events returns OK", func(t *testing.T) {
if err != nil {
t.Fatalf("call tool failed %v", err)
}
if toolResult.IsError {
t.Fatalf("call tool failed")
}
if toolResult.Content[0].(mcp.TextContent).Text != "No events found" {
t.Fatalf("unexpected result %v", toolResult.Content[0].(mcp.TextContent).Text)
type EventsSuite struct {
BaseMcpSuite
}
func (s *EventsSuite) TestEventsList() {
s.InitMcpClient()
s.Run("events_list (no events)", func() {
toolResult, err := s.CallTool("events_list", map[string]interface{}{})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
client := c.newKubernetesClient()
s.Run("returns no events message", func() {
s.Equal("No events found", toolResult.Content[0].(mcp.TextContent).Text)
})
})
s.Run("events_list (with events)", func() {
client := kubernetes.NewForConfigOrDie(envTestRestConfig)
for _, ns := range []string{"default", "ns-1"} {
_, _ = client.CoreV1().Events(ns).Create(c.ctx, &v1.Event{
_, _ = client.CoreV1().Events(ns).Create(s.T().Context(), &v1.Event{
ObjectMeta: metav1.ObjectMeta{
Name: "an-event-in-" + ns,
},
@@ -42,15 +44,14 @@ func TestEventsList(t *testing.T) {
Message: "The event message",
}, metav1.CreateOptions{})
}
toolResult, err = c.callTool("events_list", map[string]interface{}{})
t.Run("events_list with events returns all OK", func(t *testing.T) {
if err != nil {
t.Fatalf("call tool failed %v", err)
}
if toolResult.IsError {
t.Fatalf("call tool failed")
}
if toolResult.Content[0].(mcp.TextContent).Text != "The following events (YAML format) were found:\n"+
s.Run("events_list()", func() {
toolResult, err := s.CallTool("events_list", map[string]interface{}{})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
s.Run("returns all events", func() {
s.Equalf("The following events (YAML format) were found:\n"+
"- InvolvedObject:\n"+
" Kind: Pod\n"+
" Name: a-pod\n"+
@@ -68,21 +69,22 @@ func TestEventsList(t *testing.T) {
" Namespace: ns-1\n"+
" Reason: \"\"\n"+
" Timestamp: 0001-01-01 00:00:00 +0000 UTC\n"+
" Type: Normal\n" {
t.Fatalf("unexpected result %v", toolResult.Content[0].(mcp.TextContent).Text)
}
" Type: Normal\n",
toolResult.Content[0].(mcp.TextContent).Text,
"unexpected result %v", toolResult.Content[0].(mcp.TextContent).Text)
})
toolResult, err = c.callTool("events_list", map[string]interface{}{
})
s.Run("events_list(namespace=ns-1)", func() {
toolResult, err := s.CallTool("events_list", map[string]interface{}{
"namespace": "ns-1",
})
t.Run("events_list in namespace with events returns from namespace OK", func(t *testing.T) {
if err != nil {
t.Fatalf("call tool failed %v", err)
}
if toolResult.IsError {
t.Fatalf("call tool failed")
}
if toolResult.Content[0].(mcp.TextContent).Text != "The following events (YAML format) were found:\n"+
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
s.Run("returns events from namespace", func() {
s.Equalf("The following events (YAML format) were found:\n"+
"- InvolvedObject:\n"+
" Kind: Pod\n"+
" Name: a-pod\n"+
@@ -91,30 +93,33 @@ func TestEventsList(t *testing.T) {
" Namespace: ns-1\n"+
" Reason: \"\"\n"+
" Timestamp: 0001-01-01 00:00:00 +0000 UTC\n"+
" Type: Normal\n" {
t.Fatalf("unexpected result %v", toolResult.Content[0].(mcp.TextContent).Text)
}
" Type: Normal\n",
toolResult.Content[0].(mcp.TextContent).Text,
"unexpected result %v", toolResult.Content[0].(mcp.TextContent).Text)
})
})
})
}
func TestEventsListDenied(t *testing.T) {
deniedResourcesServer := test.Must(config.ReadToml([]byte(`
func (s *EventsSuite) TestEventsListDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [ { version = "v1", kind = "Event" } ]
`)))
testCaseWithContext(t, &mcpContext{staticConfig: deniedResourcesServer}, func(c *mcpContext) {
c.withEnvTest()
eventList, _ := c.callTool("events_list", map[string]interface{}{})
t.Run("events_list has error", func(t *testing.T) {
if !eventList.IsError {
t.Fatalf("call tool should fail")
}
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("events_list (denied)", func() {
eventList, err := s.CallTool("events_list", map[string]interface{}{})
s.Run("events_list has error", func() {
s.Truef(eventList.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
t.Run("events_list describes denial", func(t *testing.T) {
s.Run("events_list describes denial", func() {
expectedMessage := "failed to list events in all namespaces: resource not allowed: /v1, Kind=Event"
if eventList.Content[0].(mcp.TextContent).Text != expectedMessage {
t.Fatalf("expected descriptive error '%s', got %v", expectedMessage, eventList.Content[0].(mcp.TextContent).Text)
}
s.Equalf(expectedMessage, eventList.Content[0].(mcp.TextContent).Text,
"expected descriptive error '%s', got %v", expectedMessage, eventList.Content[0].(mcp.TextContent).Text)
})
})
}
func TestEvents(t *testing.T) {
suite.Run(t, new(EventsSuite))
}