Skip to content

test(pods): update PodsRun tests to use testify and improve readability #433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 3, 2025
Merged

test(pods): update PodsRun tests to use testify and improve readability #433

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
215 changes: 73 additions & 142 deletions pkg/mcp/pods_run_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,180 +4,107 @@ import (
"strings"
"testing"

"github.com/containers/kubernetes-mcp-server/internal/test"
"github.com/containers/kubernetes-mcp-server/pkg/config"
"github.com/BurntSushi/toml"
"github.com/mark3labs/mcp-go/mcp"
"github.com/stretchr/testify/suite"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"sigs.k8s.io/yaml"
)

func TestPodsRun(t *testing.T) {
testCase(t, func(c *mcpContext) {
c.withEnvTest()
t.Run("pods_run with nil image returns error", func(t *testing.T) {
toolResult, _ := c.callTool("pods_run", map[string]interface{}{})
if toolResult.IsError != true {
t.Errorf("call tool should fail")
return
}
if toolResult.Content[0].(mcp.TextContent).Text != "failed to run pod, missing argument image" {
t.Errorf("invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
return
}
})
podsRunNilNamespace, err := c.callTool("pods_run", map[string]interface{}{"image": "nginx"})
t.Run("pods_run with image and nil namespace runs pod", func(t *testing.T) {
if err != nil {
t.Errorf("call tool failed %v", err)
return
}
if podsRunNilNamespace.IsError {
t.Errorf("call tool failed")
return
}
type PodsRunSuite struct {
BaseMcpSuite
}

func (s *PodsRunSuite) TestPodsRun() {
s.InitMcpClient()
s.Run("pods_run with nil image returns error", func() {
toolResult, _ := s.CallTool("pods_run", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to run pod, missing argument image", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_run(image=nginx, namespace=nil), uses configured namespace", func() {
podsRunNilNamespace, err := s.CallTool("pods_run", map[string]interface{}{"image": "nginx"})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsRunNilNamespace.IsError, "call tool failed")
})
var decodedNilNamespace []unstructured.Unstructured
err = yaml.Unmarshal([]byte(podsRunNilNamespace.Content[0].(mcp.TextContent).Text), &decodedNilNamespace)
t.Run("pods_run with image and nil namespace has yaml content", func(t *testing.T) {
if err != nil {
t.Errorf("invalid tool result content %v", err)
return
}
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
t.Run("pods_run with image and nil namespace returns 1 item (Pod)", func(t *testing.T) {
if len(decodedNilNamespace) != 1 {
t.Errorf("invalid pods count, expected 1, got %v", len(decodedNilNamespace))
return
}
if decodedNilNamespace[0].GetKind() != "Pod" {
t.Errorf("invalid pod kind, expected Pod, got %v", decodedNilNamespace[0].GetKind())
return
}
s.Run("returns 1 item (Pod)", func() {
s.Lenf(decodedNilNamespace, 1, "invalid pods count, expected 1, got %v", len(decodedNilNamespace))
s.Equalf("Pod", decodedNilNamespace[0].GetKind(), "invalid pod kind, expected Pod, got %v", decodedNilNamespace[0].GetKind())
})
t.Run("pods_run with image and nil namespace returns pod in default", func(t *testing.T) {
if decodedNilNamespace[0].GetNamespace() != "default" {
t.Errorf("invalid pod namespace, expected default, got %v", decodedNilNamespace[0].GetNamespace())
return
}
s.Run("returns pod in default", func() {
s.Equalf("default", decodedNilNamespace[0].GetNamespace(), "invalid pod namespace, expected default, got %v", decodedNilNamespace[0].GetNamespace())
})
t.Run("pods_run with image and nil namespace returns pod with random name", func(t *testing.T) {
if !strings.HasPrefix(decodedNilNamespace[0].GetName(), "kubernetes-mcp-server-run-") {
t.Errorf("invalid pod name, expected random, got %v", decodedNilNamespace[0].GetName())
return
}
s.Run("returns pod with random name", func() {
s.Truef(strings.HasPrefix(decodedNilNamespace[0].GetName(), "kubernetes-mcp-server-run-"),
"invalid pod name, expected random, got %v", decodedNilNamespace[0].GetName())
})
t.Run("pods_run with image and nil namespace returns pod with labels", func(t *testing.T) {
s.Run("returns pod with labels", func() {
labels := decodedNilNamespace[0].Object["metadata"].(map[string]interface{})["labels"].(map[string]interface{})
if labels["app.kubernetes.io/name"] == "" {
t.Errorf("invalid labels, expected app.kubernetes.io/name, got %v", labels)
return
}
if labels["app.kubernetes.io/component"] == "" {
t.Errorf("invalid labels, expected app.kubernetes.io/component, got %v", labels)
return
}
if labels["app.kubernetes.io/managed-by"] != "kubernetes-mcp-server" {
t.Errorf("invalid labels, expected app.kubernetes.io/managed-by, got %v", labels)
return
}
if labels["app.kubernetes.io/part-of"] != "kubernetes-mcp-server-run-sandbox" {
t.Errorf("invalid labels, expected app.kubernetes.io/part-of, got %v", labels)
return
}
s.NotEqualf("", labels["app.kubernetes.io/name"], "invalid labels, expected app.kubernetes.io/name, got %v", labels)
s.NotEqualf("", labels["app.kubernetes.io/component"], "invalid labels, expected app.kubernetes.io/component, got %v", labels)
s.Equalf("kubernetes-mcp-server", labels["app.kubernetes.io/managed-by"], "invalid labels, expected app.kubernetes.io/managed-by, got %v", labels)
s.Equalf("kubernetes-mcp-server-run-sandbox", labels["app.kubernetes.io/part-of"], "invalid labels, expected app.kubernetes.io/part-of, got %v", labels)
})
t.Run("pods_run with image and nil namespace returns pod with nginx container", func(t *testing.T) {
s.Run("returns pod with nginx container", func() {
containers := decodedNilNamespace[0].Object["spec"].(map[string]interface{})["containers"].([]interface{})
if containers[0].(map[string]interface{})["image"] != "nginx" {
t.Errorf("invalid container name, expected nginx, got %v", containers[0].(map[string]interface{})["image"])
return
}
s.Equalf("nginx", containers[0].(map[string]interface{})["image"], "invalid container name, expected nginx, got %v", containers[0].(map[string]interface{})["image"])
})

podsRunNamespaceAndPort, err := c.callTool("pods_run", map[string]interface{}{"image": "nginx", "port": 80})
t.Run("pods_run with image, namespace, and port runs pod", func(t *testing.T) {
if err != nil {
t.Errorf("call tool failed %v", err)
return
}
if podsRunNamespaceAndPort.IsError {
t.Errorf("call tool failed")
return
}
})
s.Run("pods_run(image=nginx, namespace=nil, port=80)", func() {
podsRunNamespaceAndPort, err := s.CallTool("pods_run", map[string]interface{}{"image": "nginx", "port": 80})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsRunNamespaceAndPort.IsError, "call tool failed")
})
var decodedNamespaceAndPort []unstructured.Unstructured
err = yaml.Unmarshal([]byte(podsRunNamespaceAndPort.Content[0].(mcp.TextContent).Text), &decodedNamespaceAndPort)
t.Run("pods_run with image, namespace, and port has yaml content", func(t *testing.T) {
if err != nil {
t.Errorf("invalid tool result content %v", err)
return
}
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
t.Run("pods_run with image, namespace, and port returns 2 items (Pod + Service)", func(t *testing.T) {
if len(decodedNamespaceAndPort) != 2 {
t.Errorf("invalid pods count, expected 2, got %v", len(decodedNamespaceAndPort))
return
}
if decodedNamespaceAndPort[0].GetKind() != "Pod" {
t.Errorf("invalid pod kind, expected Pod, got %v", decodedNamespaceAndPort[0].GetKind())
return
}
if decodedNamespaceAndPort[1].GetKind() != "Service" {
t.Errorf("invalid service kind, expected Service, got %v", decodedNamespaceAndPort[1].GetKind())
return
}
s.Run("returns 2 items (Pod + Service)", func() {
s.Lenf(decodedNamespaceAndPort, 2, "invalid pods count, expected 2, got %v", len(decodedNamespaceAndPort))
s.Equalf("Pod", decodedNamespaceAndPort[0].GetKind(), "invalid pod kind, expected Pod, got %v", decodedNamespaceAndPort[0].GetKind())
s.Equalf("Service", decodedNamespaceAndPort[1].GetKind(), "invalid service kind, expected Service, got %v", decodedNamespaceAndPort[1].GetKind())
})
t.Run("pods_run with image, namespace, and port returns pod with port", func(t *testing.T) {
s.Run("returns pod with port", func() {
containers := decodedNamespaceAndPort[0].Object["spec"].(map[string]interface{})["containers"].([]interface{})
ports := containers[0].(map[string]interface{})["ports"].([]interface{})
if ports[0].(map[string]interface{})["containerPort"] != int64(80) {
t.Errorf("invalid container port, expected 80, got %v", ports[0].(map[string]interface{})["containerPort"])
return
}
s.Equalf(int64(80), ports[0].(map[string]interface{})["containerPort"], "invalid container port, expected 80, got %v", ports[0].(map[string]interface{})["containerPort"])
})
t.Run("pods_run with image, namespace, and port returns service with port and selector", func(t *testing.T) {
s.Run("returns service with port and selector", func() {
ports := decodedNamespaceAndPort[1].Object["spec"].(map[string]interface{})["ports"].([]interface{})
if ports[0].(map[string]interface{})["port"] != int64(80) {
t.Errorf("invalid service port, expected 80, got %v", ports[0].(map[string]interface{})["port"])
return
}
if ports[0].(map[string]interface{})["targetPort"] != int64(80) {
t.Errorf("invalid service target port, expected 80, got %v", ports[0].(map[string]interface{})["targetPort"])
return
}
s.Equalf(int64(80), ports[0].(map[string]interface{})["port"], "invalid service port, expected 80, got %v", ports[0].(map[string]interface{})["port"])
s.Equalf(int64(80), ports[0].(map[string]interface{})["targetPort"], "invalid service target port, expected 80, got %v", ports[0].(map[string]interface{})["targetPort"])
selector := decodedNamespaceAndPort[1].Object["spec"].(map[string]interface{})["selector"].(map[string]interface{})
if selector["app.kubernetes.io/name"] == "" {
t.Errorf("invalid service selector, expected app.kubernetes.io/name, got %v", selector)
return
}
if selector["app.kubernetes.io/managed-by"] != "kubernetes-mcp-server" {
t.Errorf("invalid service selector, expected app.kubernetes.io/managed-by, got %v", selector)
return
}
if selector["app.kubernetes.io/part-of"] != "kubernetes-mcp-server-run-sandbox" {
t.Errorf("invalid service selector, expected app.kubernetes.io/part-of, got %v", selector)
return
}
s.NotEqualf("", selector["app.kubernetes.io/name"], "invalid service selector, expected app.kubernetes.io/name, got %v", selector)
s.Equalf("kubernetes-mcp-server", selector["app.kubernetes.io/managed-by"], "invalid service selector, expected app.kubernetes.io/managed-by, got %v", selector)
s.Equalf("kubernetes-mcp-server-run-sandbox", selector["app.kubernetes.io/part-of"], "invalid service selector, expected app.kubernetes.io/part-of, got %v", selector)
})
})
}

func TestPodsRunDenied(t *testing.T) {
deniedResourcesServer := test.Must(config.ReadToml([]byte(`
func (s *PodsRunSuite) TestPodsRunDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [ { version = "v1", kind = "Pod" } ]
`)))
testCaseWithContext(t, &mcpContext{staticConfig: deniedResourcesServer}, func(c *mcpContext) {
c.withEnvTest()
podsRun, _ := c.callTool("pods_run", map[string]interface{}{"image": "nginx"})
t.Run("pods_run has error", func(t *testing.T) {
if !podsRun.IsError {
t.Fatalf("call tool should fail")
}
})
t.Run("pods_run describes denial", func(t *testing.T) {
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("pods_run (denied)", func() {
podsRun, err := s.CallTool("pods_run", map[string]interface{}{"image": "nginx"})
s.Run("has error", func() {
s.Truef(podsRun.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
expectedMessage := "failed to run pod in namespace : resource not allowed: /v1, Kind=Pod"
if podsRun.Content[0].(mcp.TextContent).Text != expectedMessage {
t.Fatalf("expected descriptive error '%s', got %v", expectedMessage, podsRun.Content[0].(mcp.TextContent).Text)
}
s.Equalf(expectedMessage, podsRun.Content[0].(mcp.TextContent).Text,
"expected descriptive error '%s', got %v", expectedMessage, podsRun.Content[0].(mcp.TextContent).Text)
})
})
}
Expand Down Expand Up @@ -216,3 +143,7 @@ func TestPodsRunInOpenShift(t *testing.T) {
})
})
}

func TestPodsRun(t *testing.T) {
suite.Run(t, new(PodsRunSuite))
}
Loading