This is page 8 of 8. Use http://codebase.md/manusa/kubernetes-mcp-server?page={x} to view the full context.
# Directory Structure
```
├── .github
│ ├── dependabot.yml
│ └── workflows
│ ├── build.yaml
│ ├── gevals.yaml
│ ├── release-helm.yaml
│ ├── release-image.yml
│ └── release.yaml
├── .gitignore
├── AGENTS.md
├── build
│ ├── gevals.mk
│ ├── helm.mk
│ ├── keycloak.mk
│ ├── kind.mk
│ ├── node.mk
│ ├── python.mk
│ └── tools.mk
├── charts
│ └── kubernetes-mcp-server
│ ├── .helmignore
│ ├── Chart.yaml
│ ├── README.md
│ ├── README.md.gotmpl
│ ├── templates
│ │ ├── _helpers.tpl
│ │ ├── configmap.yaml
│ │ ├── deployment.yaml
│ │ ├── ingress.yaml
│ │ ├── service.yaml
│ │ └── serviceaccount.yaml
│ └── values.yaml
├── CLAUDE.md
├── cmd
│ └── kubernetes-mcp-server
│ ├── main_test.go
│ └── main.go
├── dev
│ └── config
│ ├── cert-manager
│ │ └── selfsigned-issuer.yaml
│ ├── ingress
│ │ └── nginx-ingress.yaml
│ ├── keycloak
│ │ ├── client-scopes
│ │ │ ├── groups.json
│ │ │ ├── mcp-openshift.json
│ │ │ └── mcp-server.json
│ │ ├── clients
│ │ │ ├── mcp-client.json
│ │ │ ├── mcp-server-update.json
│ │ │ ├── mcp-server.json
│ │ │ └── openshift.json
│ │ ├── deployment.yaml
│ │ ├── ingress.yaml
│ │ ├── mappers
│ │ │ ├── groups-membership.json
│ │ │ ├── mcp-server-audience.json
│ │ │ ├── openshift-audience.json
│ │ │ └── username.json
│ │ ├── rbac.yaml
│ │ ├── realm
│ │ │ ├── realm-create.json
│ │ │ └── realm-events-config.json
│ │ └── users
│ │ └── mcp.json
│ └── kind
│ └── cluster.yaml
├── Dockerfile
├── docs
│ ├── GETTING_STARTED_CLAUDE_CODE.md
│ ├── GETTING_STARTED_KUBERNETES.md
│ ├── images
│ │ ├── keycloak-login-page.png
│ │ ├── keycloak-mcp-inspector-connect.png
│ │ ├── keycloak-mcp-inspector-results.png
│ │ ├── kubernetes-mcp-server-github-copilot.jpg
│ │ └── vibe-coding.jpg
│ ├── KEYCLOAK_OIDC_SETUP.md
│ ├── KIALI.md
│ ├── PROMPTS.md
│ └── README.md
├── evals
│ ├── claude-code
│ │ ├── agent.yaml
│ │ ├── eval-inline.yaml
│ │ └── eval.yaml
│ ├── mcp-config.yaml
│ ├── openai-agent
│ │ ├── agent.yaml
│ │ ├── eval-inline.yaml
│ │ └── eval.yaml
│ ├── README.md
│ └── tasks
│ ├── create-canary-deployment
│ │ ├── artifacts
│ │ │ ├── deployment-v1.yaml
│ │ │ └── service.yaml
│ │ ├── cleanup.sh
│ │ ├── create-canary-deployment.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── create-network-policy
│ │ ├── artifacts
│ │ │ └── desired-policy.yaml
│ │ ├── cleanup.sh
│ │ ├── create-network-policy.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── create-pod
│ │ ├── cleanup.sh
│ │ ├── create-pod.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── create-pod-mount-configmaps
│ │ ├── cleanup.sh
│ │ ├── create-pod-mount-configmaps.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── create-pod-resources-limits
│ │ ├── cleanup.sh
│ │ ├── create-pod-resources-limits.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── create-simple-rbac
│ │ ├── cleanup.sh
│ │ ├── create-simple-rbac.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── debug-app-logs
│ │ ├── artifacts
│ │ │ ├── calc-app-pod.yaml
│ │ │ └── calc-app.py
│ │ ├── cleanup.sh
│ │ ├── debug-app-logs.yaml
│ │ └── setup.sh
│ ├── deployment-traffic-switch
│ │ ├── artifacts
│ │ │ ├── blue-deployment.yaml
│ │ │ ├── green-deployment.yaml
│ │ │ └── service.yaml
│ │ ├── cleanup.sh
│ │ ├── deployment-traffix-switch.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── fix-crashloop
│ │ ├── cleanup.sh
│ │ ├── fix-crashloop.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── fix-image-pull
│ │ ├── cleanup.sh
│ │ ├── fix-image-pull.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── fix-pending-pod
│ │ ├── artifacts
│ │ │ ├── homepage-pod.yaml
│ │ │ └── homepage-pvc.yaml
│ │ ├── cleanup.sh
│ │ ├── fix-pending-pod.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── fix-probes
│ │ ├── cleanup.sh
│ │ ├── fix-probes.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── fix-rbac-wrong-resource
│ │ ├── cleanup.sh
│ │ ├── fix-rbac-wrong-resource.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── fix-service-routing
│ │ ├── cleanup.sh
│ │ ├── fix-service-routing.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── fix-service-with-no-endpoints
│ │ ├── artifacts
│ │ │ ├── deployment.yaml
│ │ │ └── service.yaml
│ │ ├── cleanup.sh
│ │ ├── fix-service-with-no-endpoints.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── horizontal-pod-autoscaler
│ │ ├── cleanup.sh
│ │ ├── horizontal-pod-autoscaler.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── list-images-for-pods
│ │ ├── artifacts
│ │ │ └── manifest.yaml
│ │ ├── cleanup.sh
│ │ ├── list-images-for-pods.yaml
│ │ └── setup.sh
│ ├── multi-container-pod-communication
│ │ ├── cleanup.sh
│ │ ├── multi-container-pod-communication.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── resize-pvc
│ │ ├── artifacts
│ │ │ ├── storage-pod.yaml
│ │ │ └── storage-pvc.yaml
│ │ ├── cleanup.sh
│ │ ├── resize-pvc.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── rolling-update-deployment
│ │ ├── cleanup.sh
│ │ ├── rolling-update-deployment.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── scale-deployment
│ │ ├── cleanup.sh
│ │ ├── scale-deployment.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── scale-down-deployment
│ │ ├── cleanup.sh
│ │ ├── scale-down-deployment.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ ├── setup-dev-cluster
│ │ ├── cleanup.sh
│ │ ├── setup-dev-cluster.md
│ │ ├── setup-dev-cluster.yaml
│ │ ├── setup.sh
│ │ └── verify.sh
│ └── statefulset-lifecycle
│ ├── cleanup.sh
│ ├── setup.sh
│ ├── statefulset-lifecycle.yaml
│ └── verify.sh
├── go.mod
├── go.sum
├── hack
│ └── generate-placeholder-ca.sh
├── internal
│ ├── test
│ │ ├── env.go
│ │ ├── kubernetes.go
│ │ ├── mcp.go
│ │ ├── mock_server.go
│ │ ├── test.go
│ │ ├── unstructured_test.go
│ │ └── unstructured.go
│ └── tools
│ └── update-readme
│ └── main.go
├── LICENSE
├── Makefile
├── npm
│ └── kubernetes-mcp-server
│ └── bin
│ └── index.js
├── pkg
│ ├── api
│ │ ├── config.go
│ │ ├── imports_test.go
│ │ ├── kubernetes.go
│ │ ├── params_test.go
│ │ ├── params.go
│ │ ├── prompt_serialization_test.go
│ │ ├── prompts_test.go
│ │ ├── prompts.go
│ │ ├── toolsets_test.go
│ │ └── toolsets.go
│ ├── config
│ │ ├── config_default_overrides.go
│ │ ├── config_default.go
│ │ ├── config_test.go
│ │ ├── config.go
│ │ ├── context.go
│ │ ├── extended.go
│ │ ├── provider_config_test.go
│ │ ├── provider_config.go
│ │ ├── toolset_config_test.go
│ │ └── toolset_config.go
│ ├── helm
│ │ └── helm.go
│ ├── http
│ │ ├── authorization_test.go
│ │ ├── authorization.go
│ │ ├── http_authorization_test.go
│ │ ├── http_mcp_test.go
│ │ ├── http_test.go
│ │ ├── http.go
│ │ ├── middleware.go
│ │ ├── sts_test.go
│ │ ├── sts.go
│ │ └── wellknown.go
│ ├── kiali
│ │ ├── config_test.go
│ │ ├── config.go
│ │ ├── defaults.go
│ │ ├── endpoints.go
│ │ ├── get_mesh_graph.go
│ │ ├── graph.go
│ │ ├── health_calculation.go
│ │ ├── health.go
│ │ ├── istio.go
│ │ ├── kiali_test.go
│ │ ├── kiali.go
│ │ ├── logs.go
│ │ ├── mesh.go
│ │ ├── namespaces.go
│ │ ├── services.go
│ │ ├── traces.go
│ │ ├── types.go
│ │ ├── validations.go
│ │ └── workloads.go
│ ├── kubernetes
│ │ ├── accesscontrol_round_tripper_test.go
│ │ ├── accesscontrol_round_tripper.go
│ │ ├── common_test.go
│ │ ├── configuration.go
│ │ ├── events.go
│ │ ├── impersonate_roundtripper.go
│ │ ├── kubernetes_derived_test.go
│ │ ├── kubernetes.go
│ │ ├── manager_test.go
│ │ ├── manager.go
│ │ ├── namespaces.go
│ │ ├── nodes.go
│ │ ├── openshift.go
│ │ ├── pods.go
│ │ ├── provider_kubeconfig_test.go
│ │ ├── provider_kubeconfig.go
│ │ ├── provider_registry_test.go
│ │ ├── provider_registry.go
│ │ ├── provider_single_test.go
│ │ ├── provider_single.go
│ │ ├── provider_test.go
│ │ ├── provider_watch_test.go
│ │ ├── provider.go
│ │ ├── resources.go
│ │ ├── token.go
│ │ └── watcher
│ │ ├── cluster_test.go
│ │ ├── cluster.go
│ │ ├── kubeconfig_test.go
│ │ ├── kubeconfig.go
│ │ └── watcher.go
│ ├── kubernetes-mcp-server
│ │ └── cmd
│ │ ├── root_sighup_test.go
│ │ ├── root_test.go
│ │ ├── root.go
│ │ └── testdata
│ │ ├── empty-config.toml
│ │ └── valid-config.toml
│ ├── kubevirt
│ │ ├── resources_test.go
│ │ ├── resources.go
│ │ ├── testing
│ │ │ └── helpers.go
│ │ ├── vm_test.go
│ │ └── vm.go
│ ├── mcp
│ │ ├── common_crd_test.go
│ │ ├── common_test.go
│ │ ├── configuration_test.go
│ │ ├── events_test.go
│ │ ├── gosdk.go
│ │ ├── helm_test.go
│ │ ├── kiali_test.go
│ │ ├── kubevirt_test.go
│ │ ├── mcp_middleware_test.go
│ │ ├── mcp_prompts_test.go
│ │ ├── mcp_reload_test.go
│ │ ├── mcp_test.go
│ │ ├── mcp_tools_test.go
│ │ ├── mcp_toolset_prompts_test.go
│ │ ├── mcp_watch_test.go
│ │ ├── mcp.go
│ │ ├── middleware.go
│ │ ├── modules.go
│ │ ├── namespaces_test.go
│ │ ├── nodes_test.go
│ │ ├── nodes_top_test.go
│ │ ├── pods_exec_test.go
│ │ ├── pods_run_test.go
│ │ ├── pods_test.go
│ │ ├── pods_top_test.go
│ │ ├── prompts_config_test.go
│ │ ├── prompts_gosdk_test.go
│ │ ├── prompts_gosdk.go
│ │ ├── resources_test.go
│ │ ├── testdata
│ │ │ ├── helm-chart-no-op
│ │ │ │ └── Chart.yaml
│ │ │ ├── helm-chart-secret
│ │ │ │ ├── Chart.yaml
│ │ │ │ └── templates
│ │ │ │ └── secret.yaml
│ │ │ ├── toolsets-config-tools.json
│ │ │ ├── toolsets-core-tools.json
│ │ │ ├── toolsets-full-tools-multicluster-enum.json
│ │ │ ├── toolsets-full-tools-multicluster.json
│ │ │ ├── toolsets-full-tools-openshift.json
│ │ │ ├── toolsets-full-tools.json
│ │ │ ├── toolsets-helm-tools.json
│ │ │ ├── toolsets-kiali-tools.json
│ │ │ └── toolsets-kubevirt-tools.json
│ │ ├── tool_filter_test.go
│ │ ├── tool_filter.go
│ │ ├── tool_mutator_test.go
│ │ ├── tool_mutator.go
│ │ └── toolsets_test.go
│ ├── openshift
│ │ └── openshift.go
│ ├── output
│ │ ├── output_test.go
│ │ └── output.go
│ ├── prompts
│ │ ├── prompts_test.go
│ │ └── prompts.go
│ ├── toolsets
│ │ ├── config
│ │ │ ├── configuration.go
│ │ │ └── toolset.go
│ │ ├── core
│ │ │ ├── events.go
│ │ │ ├── namespaces.go
│ │ │ ├── nodes.go
│ │ │ ├── pods.go
│ │ │ ├── resources.go
│ │ │ └── toolset.go
│ │ ├── helm
│ │ │ ├── helm.go
│ │ │ └── toolset.go
│ │ ├── kiali
│ │ │ ├── internal
│ │ │ │ └── defaults
│ │ │ │ ├── defaults_override.go
│ │ │ │ └── defaults.go
│ │ │ ├── tools
│ │ │ │ ├── get_mesh_graph.go
│ │ │ │ ├── get_metrics.go
│ │ │ │ ├── get_resource_details.go
│ │ │ │ ├── get_traces.go
│ │ │ │ ├── helpers_test.go
│ │ │ │ ├── helpers.go
│ │ │ │ ├── logs.go
│ │ │ │ └── manage_istio_config.go
│ │ │ └── toolset.go
│ │ ├── kubevirt
│ │ │ ├── toolset.go
│ │ │ └── vm
│ │ │ ├── create
│ │ │ │ ├── tool.go
│ │ │ │ └── vm.yaml.tmpl
│ │ │ └── lifecycle
│ │ │ └── tool.go
│ │ ├── toolsets_test.go
│ │ └── toolsets.go
│ └── version
│ └── version.go
├── python
│ ├── kubernetes_mcp_server
│ │ ├── __init__.py
│ │ ├── __main__.py
│ │ └── kubernetes_mcp_server.py
│ ├── pyproject.toml
│ └── README.md
├── README.md
└── smithery.yaml
```
# Files
--------------------------------------------------------------------------------
/pkg/config/config_test.go:
--------------------------------------------------------------------------------
```go
package config
import (
"errors"
"io/fs"
"os"
"path/filepath"
"strings"
"testing"
"github.com/containers/kubernetes-mcp-server/pkg/api"
"github.com/stretchr/testify/suite"
)
type BaseConfigSuite struct {
suite.Suite
}
func (s *BaseConfigSuite) writeConfig(content string) string {
s.T().Helper()
tempDir := s.T().TempDir()
path := filepath.Join(tempDir, "config.toml")
err := os.WriteFile(path, []byte(content), 0644)
if err != nil {
s.T().Fatalf("Failed to write config file %s: %v", path, err)
}
return path
}
type ConfigSuite struct {
BaseConfigSuite
}
func (s *ConfigSuite) TestReadConfigMissingFile() {
config, err := Read("non-existent-config.toml", "")
s.Run("returns error for missing file", func() {
s.Require().NotNil(err, "Expected error for missing file, got nil")
s.True(errors.Is(err, fs.ErrNotExist), "Expected ErrNotExist, got %v", err)
})
s.Run("returns nil config for missing file", func() {
s.Nil(config, "Expected nil config for missing file")
})
}
func (s *ConfigSuite) TestReadConfigInvalid() {
invalidConfigPath := s.writeConfig(`
[[denied_resources]]
group = "apps"
version = "v1"
kind = "Deployment"
[[denied_resources]]
group = "rbac.authorization.k8s.io"
version = "v1"
kind = "Role
`)
config, err := Read(invalidConfigPath, "")
s.Run("returns error for invalid file", func() {
s.Require().NotNil(err, "Expected error for invalid file, got nil")
})
s.Run("error message contains toml error with line number", func() {
expectedError := "toml: line 9"
s.Truef(strings.Contains(err.Error(), expectedError), "Expected error message to contain line number, got %v", err)
})
s.Run("returns nil config for invalid file", func() {
s.Nil(config, "Expected nil config for missing file")
})
}
func (s *ConfigSuite) TestReadConfigValid() {
validConfigPath := s.writeConfig(`
log_level = 1
port = "9999"
sse_base_url = "https://example.com"
kubeconfig = "./path/to/config"
list_output = "yaml"
read_only = true
disable_destructive = true
toolsets = ["core", "config", "helm", "metrics"]
enabled_tools = ["configuration_view", "events_list", "namespaces_list", "pods_list", "resources_list", "resources_get", "resources_create_or_update", "resources_delete"]
disabled_tools = ["pods_delete", "pods_top", "pods_log", "pods_run", "pods_exec"]
denied_resources = [
{group = "apps", version = "v1", kind = "Deployment"},
{group = "rbac.authorization.k8s.io", version = "v1", kind = "Role"}
]
`)
config, err := Read(validConfigPath, "")
s.Require().NotNil(config)
s.Run("reads and unmarshalls file", func() {
s.Nil(err, "Expected nil error for valid file")
s.Require().NotNil(config, "Expected non-nil config for valid file")
})
s.Run("log_level parsed correctly", func() {
s.Equalf(1, config.LogLevel, "Expected LogLevel to be 1, got %d", config.LogLevel)
})
s.Run("port parsed correctly", func() {
s.Equalf("9999", config.Port, "Expected Port to be 9999, got %s", config.Port)
})
s.Run("sse_base_url parsed correctly", func() {
s.Equalf("https://example.com", config.SSEBaseURL, "Expected SSEBaseURL to be https://example.com, got %s", config.SSEBaseURL)
})
s.Run("kubeconfig parsed correctly", func() {
s.Equalf("./path/to/config", config.KubeConfig, "Expected KubeConfig to be ./path/to/config, got %s", config.KubeConfig)
})
s.Run("list_output parsed correctly", func() {
s.Equalf("yaml", config.ListOutput, "Expected ListOutput to be yaml, got %s", config.ListOutput)
})
s.Run("read_only parsed correctly", func() {
s.Truef(config.ReadOnly, "Expected ReadOnly to be true, got %v", config.ReadOnly)
})
s.Run("disable_destructive parsed correctly", func() {
s.Truef(config.DisableDestructive, "Expected DisableDestructive to be true, got %v", config.DisableDestructive)
})
s.Run("toolsets", func() {
s.Require().Lenf(config.Toolsets, 4, "Expected 4 toolsets, got %d", len(config.Toolsets))
for _, toolset := range []string{"core", "config", "helm", "metrics"} {
s.Containsf(config.Toolsets, toolset, "Expected toolsets to contain %s", toolset)
}
})
s.Run("enabled_tools", func() {
s.Require().Lenf(config.EnabledTools, 8, "Expected 8 enabled tools, got %d", len(config.EnabledTools))
for _, tool := range []string{"configuration_view", "events_list", "namespaces_list", "pods_list", "resources_list", "resources_get", "resources_create_or_update", "resources_delete"} {
s.Containsf(config.EnabledTools, tool, "Expected enabled tools to contain %s", tool)
}
})
s.Run("disabled_tools", func() {
s.Require().Lenf(config.DisabledTools, 5, "Expected 5 disabled tools, got %d", len(config.DisabledTools))
for _, tool := range []string{"pods_delete", "pods_top", "pods_log", "pods_run", "pods_exec"} {
s.Containsf(config.DisabledTools, tool, "Expected disabled tools to contain %s", tool)
}
})
s.Run("denied_resources", func() {
s.Require().Lenf(config.DeniedResources, 2, "Expected 2 denied resources, got %d", len(config.DeniedResources))
s.Run("contains apps/v1/Deployment", func() {
s.Contains(config.DeniedResources, api.GroupVersionKind{Group: "apps", Version: "v1", Kind: "Deployment"},
"Expected denied resources to contain apps/v1/Deployment")
})
s.Run("contains rbac.authorization.k8s.io/v1/Role", func() {
s.Contains(config.DeniedResources, api.GroupVersionKind{Group: "rbac.authorization.k8s.io", Version: "v1", Kind: "Role"},
"Expected denied resources to contain rbac.authorization.k8s.io/v1/Role")
})
})
}
func (s *ConfigSuite) TestReadConfigValidPreservesDefaultsForMissingFields() {
if HasDefaultOverrides() {
s.T().Skip("Skipping test because default configuration overrides are present (this is a downstream fork)")
}
validConfigPath := s.writeConfig(`
port = "1337"
`)
config, err := Read(validConfigPath, "")
s.Require().NotNil(config)
s.Run("reads and unmarshalls file", func() {
s.Nil(err, "Expected nil error for valid file")
s.Require().NotNil(config, "Expected non-nil config for valid file")
})
s.Run("log_level defaulted correctly", func() {
s.Equalf(0, config.LogLevel, "Expected LogLevel to be 0, got %d", config.LogLevel)
})
s.Run("port parsed correctly", func() {
s.Equalf("1337", config.Port, "Expected Port to be 9999, got %s", config.Port)
})
s.Run("list_output defaulted correctly", func() {
s.Equalf("table", config.ListOutput, "Expected ListOutput to be table, got %s", config.ListOutput)
})
s.Run("toolsets defaulted correctly", func() {
s.Require().Lenf(config.Toolsets, 3, "Expected 3 toolsets, got %d", len(config.Toolsets))
for _, toolset := range []string{"core", "config", "helm"} {
s.Containsf(config.Toolsets, toolset, "Expected toolsets to contain %s", toolset)
}
})
}
func (s *ConfigSuite) TestGetSortedConfigFiles() {
tempDir := s.T().TempDir()
// Create test files
files := []string{
"10-first.toml",
"20-second.toml",
"05-before.toml",
"99-last.toml",
".hidden.toml", // should be ignored
"readme.txt", // should be ignored
"invalid", // should be ignored
}
for _, file := range files {
path := filepath.Join(tempDir, file)
err := os.WriteFile(path, []byte(""), 0644)
s.Require().NoError(err)
}
// Create a subdirectory (should be ignored)
subDir := filepath.Join(tempDir, "subdir")
err := os.Mkdir(subDir, 0755)
s.Require().NoError(err)
sorted, err := getSortedConfigFiles(tempDir)
s.Require().NoError(err)
s.Run("returns only .toml files", func() {
s.Len(sorted, 4, "Expected 4 .toml files")
})
s.Run("sorted in lexical order", func() {
expected := []string{
filepath.Join(tempDir, "05-before.toml"),
filepath.Join(tempDir, "10-first.toml"),
filepath.Join(tempDir, "20-second.toml"),
filepath.Join(tempDir, "99-last.toml"),
}
s.Equal(expected, sorted)
})
s.Run("excludes dotfiles", func() {
for _, file := range sorted {
s.NotContains(file, ".hidden")
}
})
s.Run("excludes non-.toml files", func() {
for _, file := range sorted {
s.Contains(file, ".toml")
}
})
}
func (s *ConfigSuite) TestDropInConfigPrecedence() {
tempDir := s.T().TempDir()
// Main config file
mainConfigPath := s.writeConfig(`
log_level = 1
port = "8080"
list_output = "table"
toolsets = ["core", "config"]
`)
// Create drop-in directory
dropInDir := filepath.Join(tempDir, "config.d")
err := os.Mkdir(dropInDir, 0755)
s.Require().NoError(err)
// First drop-in file
dropIn1 := filepath.Join(dropInDir, "10-override.toml")
err = os.WriteFile(dropIn1, []byte(`
log_level = 5
port = "9090"
`), 0644)
s.Require().NoError(err)
// Second drop-in file (should override first)
dropIn2 := filepath.Join(dropInDir, "20-final.toml")
err = os.WriteFile(dropIn2, []byte(`
port = "7777"
list_output = "yaml"
`), 0644)
s.Require().NoError(err)
config, err := Read(mainConfigPath, dropInDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("drop-in overrides main config", func() {
s.Equal(5, config.LogLevel, "log_level from 10-override.toml should override main")
})
s.Run("later drop-in overrides earlier drop-in", func() {
s.Equal("7777", config.Port, "port from 20-final.toml should override 10-override.toml")
})
s.Run("preserves values not in drop-in files", func() {
s.Equal([]string{"core", "config"}, config.Toolsets, "toolsets from main config should be preserved")
})
s.Run("applies all drop-in changes", func() {
s.Equal("yaml", config.ListOutput, "list_output from 20-final.toml should be applied")
})
}
func (s *ConfigSuite) TestDropInConfigMissingDirectory() {
mainConfigPath := s.writeConfig(`
log_level = 3
port = "8080"
`)
config, err := Read(mainConfigPath, "/non/existent/directory")
s.Require().NoError(err, "Should not error for missing drop-in directory")
s.Require().NotNil(config)
s.Run("loads main config successfully", func() {
s.Equal(3, config.LogLevel)
s.Equal("8080", config.Port)
})
}
func (s *ConfigSuite) TestDropInConfigEmptyDirectory() {
mainConfigPath := s.writeConfig(`
log_level = 2
`)
dropInDir := s.T().TempDir()
config, err := Read(mainConfigPath, dropInDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("loads main config successfully", func() {
s.Equal(2, config.LogLevel)
})
}
func (s *ConfigSuite) TestDropInConfigPartialOverride() {
tempDir := s.T().TempDir()
mainConfigPath := s.writeConfig(`
log_level = 1
port = "8080"
list_output = "table"
read_only = false
toolsets = ["core", "config", "helm"]
`)
dropInDir := filepath.Join(tempDir, "config.d")
err := os.Mkdir(dropInDir, 0755)
s.Require().NoError(err)
// Drop-in file with partial config
dropIn := filepath.Join(dropInDir, "10-partial.toml")
err = os.WriteFile(dropIn, []byte(`
read_only = true
`), 0644)
s.Require().NoError(err)
config, err := Read(mainConfigPath, dropInDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("overrides specified field", func() {
s.True(config.ReadOnly, "read_only should be overridden to true")
})
s.Run("preserves all other fields", func() {
s.Equal(1, config.LogLevel)
s.Equal("8080", config.Port)
s.Equal("table", config.ListOutput)
s.Equal([]string{"core", "config", "helm"}, config.Toolsets)
})
}
func (s *ConfigSuite) TestDropInConfigWithArrays() {
tempDir := s.T().TempDir()
mainConfigPath := s.writeConfig(`
toolsets = ["core", "config"]
enabled_tools = ["tool1", "tool2"]
`)
dropInDir := filepath.Join(tempDir, "config.d")
err := os.Mkdir(dropInDir, 0755)
s.Require().NoError(err)
dropIn := filepath.Join(dropInDir, "10-arrays.toml")
err = os.WriteFile(dropIn, []byte(`
toolsets = ["helm", "logs"]
`), 0644)
s.Require().NoError(err)
config, err := Read(mainConfigPath, dropInDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("replaces arrays completely", func() {
s.Equal([]string{"helm", "logs"}, config.Toolsets, "toolsets should be completely replaced")
s.Equal([]string{"tool1", "tool2"}, config.EnabledTools, "enabled_tools should be preserved")
})
}
func (s *ConfigSuite) TestDefaultConfDResolution() {
// Create a temp directory structure:
// tempDir/
// config.toml
// conf.d/
// 10-override.toml
tempDir := s.T().TempDir()
// Create main config file
mainConfigPath := filepath.Join(tempDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(`
log_level = 1
port = "8080"
`), 0644))
// Create default conf.d directory
confDDir := filepath.Join(tempDir, "conf.d")
s.Require().NoError(os.Mkdir(confDDir, 0755))
// Create drop-in file in conf.d
dropIn := filepath.Join(confDDir, "10-override.toml")
s.Require().NoError(os.WriteFile(dropIn, []byte(`
log_level = 5
port = "9090"
`), 0644))
// Read config WITHOUT specifying drop-in directory - should auto-discover conf.d
config, err := Read(mainConfigPath, "")
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("auto-discovers conf.d relative to config file", func() {
s.Equal(5, config.LogLevel, "log_level should be overridden by conf.d/10-override.toml")
s.Equal("9090", config.Port, "port should be overridden by conf.d/10-override.toml")
})
}
func (s *ConfigSuite) TestDefaultConfDNotExist() {
// When conf.d doesn't exist, config should still load without error
mainConfigPath := s.writeConfig(`
log_level = 3
port = "8080"
`)
config, err := Read(mainConfigPath, "")
s.Require().NoError(err, "Should not error when default conf.d doesn't exist")
s.Require().NotNil(config)
s.Run("loads main config when conf.d doesn't exist", func() {
s.Equal(3, config.LogLevel)
s.Equal("8080", config.Port)
})
}
func (s *ConfigSuite) TestStandaloneConfigDir() {
// Test using only --config-dir without --config (standalone mode)
tempDir := s.T().TempDir()
// Create first drop-in file
dropIn1 := filepath.Join(tempDir, "10-base.toml")
s.Require().NoError(os.WriteFile(dropIn1, []byte(`
log_level = 2
port = "8080"
toolsets = ["core"]
`), 0644))
// Create second drop-in file
dropIn2 := filepath.Join(tempDir, "20-override.toml")
s.Require().NoError(os.WriteFile(dropIn2, []byte(`
log_level = 5
list_output = "yaml"
`), 0644))
// Read with empty config path (standalone --config-dir)
config, err := Read("", tempDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("loads config from drop-in directory only", func() {
s.Equal(5, config.LogLevel, "log_level should be from 20-override.toml")
s.Equal("8080", config.Port, "port should be from 10-base.toml")
s.Equal("yaml", config.ListOutput, "list_output should be from 20-override.toml")
s.Equal([]string{"core"}, config.Toolsets, "toolsets should be from 10-base.toml")
})
}
func (s *ConfigSuite) TestStandaloneConfigDirPreservesDefaults() {
// Test that defaults are preserved when using standalone --config-dir
if HasDefaultOverrides() {
s.T().Skip("Skipping test because default configuration overrides are present (this is a downstream fork)")
}
tempDir := s.T().TempDir()
// Create a drop-in file with only partial config
dropIn := filepath.Join(tempDir, "10-partial.toml")
s.Require().NoError(os.WriteFile(dropIn, []byte(`
port = "9999"
`), 0644))
config, err := Read("", tempDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("preserves default values", func() {
s.Equal("9999", config.Port, "port should be from drop-in")
s.Equal("table", config.ListOutput, "list_output should be default")
s.Equal([]string{"core", "config", "helm"}, config.Toolsets, "toolsets should be default")
})
}
func (s *ConfigSuite) TestStandaloneConfigDirEmpty() {
// Test standalone --config-dir with empty directory
if HasDefaultOverrides() {
s.T().Skip("Skipping test because default configuration overrides are present (this is a downstream fork)")
}
tempDir := s.T().TempDir()
config, err := Read("", tempDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("returns defaults for empty directory", func() {
s.Equal("table", config.ListOutput, "list_output should be default")
s.Equal([]string{"core", "config", "helm"}, config.Toolsets, "toolsets should be default")
})
}
func (s *ConfigSuite) TestStandaloneConfigDirNonExistent() {
// Test standalone --config-dir with non-existent directory
if HasDefaultOverrides() {
s.T().Skip("Skipping test because default configuration overrides are present (this is a downstream fork)")
}
config, err := Read("", "/non/existent/directory")
s.Require().NoError(err, "Should not error for non-existent directory")
s.Require().NotNil(config)
s.Run("returns defaults for non-existent directory", func() {
s.Equal("table", config.ListOutput, "list_output should be default")
})
}
func (s *ConfigSuite) TestConfigDirOverridesDefaultConfD() {
// Test that explicit --config-dir overrides default conf.d
tempDir := s.T().TempDir()
// Create main config file
mainConfigPath := filepath.Join(tempDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(`
log_level = 1
port = "8080"
`), 0644))
// Create default conf.d directory with a drop-in
confDDir := filepath.Join(tempDir, "conf.d")
s.Require().NoError(os.Mkdir(confDDir, 0755))
s.Require().NoError(os.WriteFile(filepath.Join(confDDir, "10-default.toml"), []byte(`
log_level = 99
port = "1111"
`), 0644))
// Create custom drop-in directory
customDir := filepath.Join(tempDir, "custom.d")
s.Require().NoError(os.Mkdir(customDir, 0755))
s.Require().NoError(os.WriteFile(filepath.Join(customDir, "10-custom.toml"), []byte(`
log_level = 5
port = "9090"
`), 0644))
// Read with explicit config-dir (should override default conf.d)
config, err := Read(mainConfigPath, customDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("uses explicit config-dir instead of default conf.d", func() {
s.Equal(5, config.LogLevel, "log_level should be from custom.d, not conf.d")
s.Equal("9090", config.Port, "port should be from custom.d, not conf.d")
})
}
func (s *ConfigSuite) TestInvalidTomlInDropIn() {
tempDir := s.T().TempDir()
mainConfigPath := s.writeConfig(`
log_level = 1
`)
// Create drop-in directory with invalid TOML
dropInDir := filepath.Join(tempDir, "config.d")
s.Require().NoError(os.Mkdir(dropInDir, 0755))
// Valid first file
s.Require().NoError(os.WriteFile(filepath.Join(dropInDir, "10-valid.toml"), []byte(`
port = "8080"
`), 0644))
// Invalid second file
s.Require().NoError(os.WriteFile(filepath.Join(dropInDir, "20-invalid.toml"), []byte(`
port = "unclosed string
`), 0644))
config, err := Read(mainConfigPath, dropInDir)
s.Run("returns error for invalid TOML in drop-in", func() {
s.Require().Error(err, "Expected error for invalid TOML")
s.Contains(err.Error(), "20-invalid.toml", "Error should mention the invalid file")
})
s.Run("returns nil config", func() {
s.Nil(config, "Expected nil config when drop-in has invalid TOML")
})
}
func (s *ConfigSuite) TestAbsoluteDropInConfigDir() {
// Test that absolute paths work for --config-dir
tempDir := s.T().TempDir()
// Create main config in one directory
configDir := filepath.Join(tempDir, "config")
s.Require().NoError(os.Mkdir(configDir, 0755))
mainConfigPath := filepath.Join(configDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(`
log_level = 1
`), 0644))
// Create drop-in directory in a completely different location
dropInDir := filepath.Join(tempDir, "somewhere", "else", "conf.d")
s.Require().NoError(os.MkdirAll(dropInDir, 0755))
s.Require().NoError(os.WriteFile(filepath.Join(dropInDir, "10-override.toml"), []byte(`
log_level = 9
port = "7777"
`), 0644))
// Use absolute path for config-dir
absDropInDir, err := filepath.Abs(dropInDir)
s.Require().NoError(err)
config, err := Read(mainConfigPath, absDropInDir)
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("loads from absolute drop-in path", func() {
s.Equal(9, config.LogLevel, "log_level should be from absolute path drop-in")
s.Equal("7777", config.Port, "port should be from absolute path drop-in")
})
}
func (s *ConfigSuite) TestDropInNotADirectory() {
tempDir := s.T().TempDir()
mainConfigPath := s.writeConfig(`
log_level = 1
`)
// Create a file (not a directory) where drop-in dir is expected
notADir := filepath.Join(tempDir, "not-a-dir")
s.Require().NoError(os.WriteFile(notADir, []byte("i am a file"), 0644))
config, err := Read(mainConfigPath, notADir)
s.Run("returns error when drop-in path is not a directory", func() {
s.Require().Error(err)
s.Contains(err.Error(), "not a directory")
})
s.Run("returns nil config", func() {
s.Nil(config)
})
}
func (s *ConfigSuite) TestDeepMerge() {
s.Run("merges flat maps", func() {
dst := map[string]interface{}{
"key1": "value1",
"key2": "value2",
}
src := map[string]interface{}{
"key2": "overridden",
"key3": "value3",
}
deepMerge(dst, src)
s.Equal("value1", dst["key1"], "existing key should be preserved")
s.Equal("overridden", dst["key2"], "overlapping key should be overridden")
s.Equal("value3", dst["key3"], "new key should be added")
})
s.Run("recursively merges nested maps", func() {
dst := map[string]interface{}{
"nested": map[string]interface{}{
"a": "original-a",
"b": "original-b",
},
}
src := map[string]interface{}{
"nested": map[string]interface{}{
"b": "overridden-b",
"c": "new-c",
},
}
deepMerge(dst, src)
nested := dst["nested"].(map[string]interface{})
s.Equal("original-a", nested["a"], "nested key not in src should be preserved")
s.Equal("overridden-b", nested["b"], "nested key in both should be overridden")
s.Equal("new-c", nested["c"], "new nested key should be added")
})
s.Run("overwrites when types differ", func() {
dst := map[string]interface{}{
"key": map[string]interface{}{"nested": "value"},
}
src := map[string]interface{}{
"key": "now-a-string",
}
deepMerge(dst, src)
s.Equal("now-a-string", dst["key"], "map should be replaced by string")
})
s.Run("replaces arrays completely", func() {
dst := map[string]interface{}{
"array": []interface{}{"a", "b", "c"},
}
src := map[string]interface{}{
"array": []interface{}{"x", "y"},
}
deepMerge(dst, src)
s.Equal([]interface{}{"x", "y"}, dst["array"], "arrays should be replaced, not merged")
})
s.Run("deeply nested merge", func() {
dst := map[string]interface{}{
"level1": map[string]interface{}{
"level2": map[string]interface{}{
"level3": map[string]interface{}{
"deep": "original",
"keep": "preserved",
},
},
},
}
src := map[string]interface{}{
"level1": map[string]interface{}{
"level2": map[string]interface{}{
"level3": map[string]interface{}{
"deep": "overridden",
},
},
},
}
deepMerge(dst, src)
level3 := dst["level1"].(map[string]interface{})["level2"].(map[string]interface{})["level3"].(map[string]interface{})
s.Equal("overridden", level3["deep"], "deeply nested key should be overridden")
s.Equal("preserved", level3["keep"], "deeply nested key not in src should be preserved")
})
}
func (s *ConfigSuite) TestDropInWithDeniedResources() {
tempDir := s.T().TempDir()
// Main config with some denied resources
mainConfigPath := filepath.Join(tempDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(`
log_level = 1
denied_resources = [
{group = "apps", version = "v1", kind = "Deployment"}
]
`), 0644))
// Create drop-in directory
dropInDir := filepath.Join(tempDir, "conf.d")
s.Require().NoError(os.Mkdir(dropInDir, 0755))
// Drop-in that replaces denied_resources (arrays are replaced, not merged)
s.Require().NoError(os.WriteFile(filepath.Join(dropInDir, "10-security.toml"), []byte(`
denied_resources = [
{group = "rbac.authorization.k8s.io", version = "v1", kind = "ClusterRole"},
{group = "rbac.authorization.k8s.io", version = "v1", kind = "ClusterRoleBinding"}
]
`), 0644))
config, err := Read(mainConfigPath, "")
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("drop-in replaces denied_resources array", func() {
s.Len(config.DeniedResources, 2, "denied_resources should have 2 entries from drop-in")
s.Contains(config.DeniedResources, api.GroupVersionKind{
Group: "rbac.authorization.k8s.io", Version: "v1", Kind: "ClusterRole",
})
s.Contains(config.DeniedResources, api.GroupVersionKind{
Group: "rbac.authorization.k8s.io", Version: "v1", Kind: "ClusterRoleBinding",
})
})
s.Run("original denied_resources from main config are replaced", func() {
s.NotContains(config.DeniedResources, api.GroupVersionKind{
Group: "apps", Version: "v1", Kind: "Deployment",
}, "original entry should be replaced by drop-in")
})
}
func (s *ConfigSuite) TestRelativeConfigDirPath() {
// Test that relative --config-dir paths are resolved relative to --config file
tempDir := s.T().TempDir()
// Create main config in a subdirectory
configSubDir := filepath.Join(tempDir, "etc", "kmcp")
s.Require().NoError(os.MkdirAll(configSubDir, 0755))
mainConfigPath := filepath.Join(configSubDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(`
log_level = 1
`), 0644))
// Create a custom drop-in dir relative to config (sibling directory)
customDropInDir := filepath.Join(configSubDir, "overrides.d")
s.Require().NoError(os.Mkdir(customDropInDir, 0755))
s.Require().NoError(os.WriteFile(filepath.Join(customDropInDir, "10-override.toml"), []byte(`
log_level = 7
port = "3333"
`), 0644))
// Use relative path for config-dir
config, err := Read(mainConfigPath, "overrides.d")
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("resolves relative config-dir against config file directory", func() {
s.Equal(7, config.LogLevel, "log_level should be from overrides.d")
s.Equal("3333", config.Port, "port should be from overrides.d")
})
}
func (s *ConfigSuite) TestBothConfigAndConfigDirEmpty() {
// Edge case: Read("", "") should return defaults
if HasDefaultOverrides() {
s.T().Skip("Skipping test because default configuration overrides are present (this is a downstream fork)")
}
config, err := Read("", "")
s.Require().NoError(err, "Should not error when both config and config-dir are empty")
s.Require().NotNil(config)
s.Run("returns default configuration", func() {
s.Equal("table", config.ListOutput)
s.Equal([]string{"core", "config", "helm"}, config.Toolsets)
s.Equal(0, config.LogLevel)
})
}
func (s *ConfigSuite) TestMultipleDropInFilesInOrder() {
// Comprehensive test of file ordering with many files
tempDir := s.T().TempDir()
// Create main config
mainConfigPath := filepath.Join(tempDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(`
log_level = 0
port = "initial"
list_output = "table"
`), 0644))
// Create conf.d with multiple files
confDDir := filepath.Join(tempDir, "conf.d")
s.Require().NoError(os.Mkdir(confDDir, 0755))
// Create files in non-alphabetical order to ensure sorting works
files := map[string]string{
"50-middle.toml": `port = "fifty"`,
"10-first.toml": `log_level = 10`,
"90-last.toml": `log_level = 90`,
"30-third.toml": `list_output = "yaml"`,
"70-seven.toml": `port = "seventy"`,
}
for name, content := range files {
s.Require().NoError(os.WriteFile(filepath.Join(confDDir, name), []byte(content), 0644))
}
config, err := Read(mainConfigPath, "")
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("processes files in lexical order", func() {
// log_level: main(0) -> 10-first(10) -> 90-last(90) = 90
s.Equal(90, config.LogLevel, "log_level should be from 90-last.toml (last to set it)")
})
s.Run("last file wins for each field", func() {
// port: main("initial") -> 50-middle("fifty") -> 70-seven("seventy") = "seventy"
s.Equal("seventy", config.Port, "port should be from 70-seven.toml (last to set it)")
// list_output: main("table") -> 30-third("yaml") = "yaml"
s.Equal("yaml", config.ListOutput, "list_output should be from 30-third.toml")
})
}
func (s *ConfigSuite) TestDropInWithNestedConfig() {
// Test that nested config structures (like cluster_provider_configs) merge correctly
tempDir := s.T().TempDir()
mainConfigPath := filepath.Join(tempDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(`
log_level = 1
cluster_provider_strategy = "kubeconfig"
[cluster_provider_configs.kubeconfig]
setting1 = "from-main"
setting2 = "from-main"
`), 0644))
confDDir := filepath.Join(tempDir, "conf.d")
s.Require().NoError(os.Mkdir(confDDir, 0755))
// First drop-in overrides one nested setting
s.Require().NoError(os.WriteFile(filepath.Join(confDDir, "10-partial.toml"), []byte(`
[cluster_provider_configs.kubeconfig]
setting1 = "from-drop-in-1"
`), 0644))
// Second drop-in overrides another nested setting
s.Require().NoError(os.WriteFile(filepath.Join(confDDir, "20-partial.toml"), []byte(`
[cluster_provider_configs.kubeconfig]
setting2 = "from-drop-in-2"
setting3 = "new-in-drop-in-2"
`), 0644))
config, err := Read(mainConfigPath, "")
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("merges nested config from multiple drop-ins", func() {
// The raw ClusterProviderConfigs should have all merged keys
s.NotNil(config.ClusterProviderConfigs["kubeconfig"])
})
}
func (s *ConfigSuite) TestEmptyConfigFile() {
// Test that an empty main config file works correctly
if HasDefaultOverrides() {
s.T().Skip("Skipping test because default configuration overrides are present (this is a downstream fork)")
}
tempDir := s.T().TempDir()
// Create empty main config
mainConfigPath := filepath.Join(tempDir, "config.toml")
s.Require().NoError(os.WriteFile(mainConfigPath, []byte(``), 0644))
// Create conf.d with overrides
confDDir := filepath.Join(tempDir, "conf.d")
s.Require().NoError(os.Mkdir(confDDir, 0755))
s.Require().NoError(os.WriteFile(filepath.Join(confDDir, "10-settings.toml"), []byte(`
log_level = 5
port = "9999"
`), 0644))
config, err := Read(mainConfigPath, "")
s.Require().NoError(err)
s.Require().NotNil(config)
s.Run("applies drop-in on top of defaults when main config is empty", func() {
s.Equal(5, config.LogLevel, "log_level should be from drop-in")
s.Equal("9999", config.Port, "port should be from drop-in")
// Defaults should still be applied for unset values
s.Equal("table", config.ListOutput, "list_output should be default")
s.Equal([]string{"core", "config", "helm"}, config.Toolsets, "toolsets should be default")
})
}
func TestConfig(t *testing.T) {
suite.Run(t, new(ConfigSuite))
}
```
--------------------------------------------------------------------------------
/pkg/mcp/pods_test.go:
--------------------------------------------------------------------------------
```go
package mcp
import (
"regexp"
"strings"
"testing"
"github.com/BurntSushi/toml"
"github.com/stretchr/testify/suite"
"github.com/mark3labs/mcp-go/mcp"
corev1 "k8s.io/api/core/v1"
rbacv1 "k8s.io/api/rbac/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
"sigs.k8s.io/yaml"
)
type PodsSuite struct {
BaseMcpSuite
}
func (s *PodsSuite) TestPodsListInAllNamespaces() {
s.InitMcpClient()
s.Run("pods_list returns pods list in all namespaces", func() {
toolResult, err := s.CallTool("pods_list", map[string]interface{}{})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decoded []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decoded)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns at least 3 items", func() {
s.GreaterOrEqualf(len(decoded), 3, "invalid pods count, expected at least 3, got %v", len(decoded))
})
var aPodInNs1, aPodInNs2 *unstructured.Unstructured
for _, pod := range decoded {
switch pod.GetName() {
case "a-pod-in-ns-1":
aPodInNs1 = &pod
case "a-pod-in-ns-2":
aPodInNs2 = &pod
}
}
s.Run("returns pod in ns-1", func() {
s.Require().NotNil(aPodInNs1, "aPodInNs1 is nil")
s.Equalf("a-pod-in-ns-1", aPodInNs1.GetName(), "invalid pod name, expected a-pod-in-ns-1, got %v", aPodInNs1.GetName())
s.Equalf("ns-1", aPodInNs1.GetNamespace(), "invalid pod namespace, expected ns-1, got %v", aPodInNs1.GetNamespace())
})
s.Run("returns pod in ns-2", func() {
s.Require().NotNil(aPodInNs2, "aPodInNs2 is nil")
s.Equalf("a-pod-in-ns-2", aPodInNs2.GetName(), "invalid pod name, expected a-pod-in-ns-2, got %v", aPodInNs2.GetName())
s.Equalf("ns-2", aPodInNs2.GetNamespace(), "invalid pod namespace, expected ns-2, got %v", aPodInNs2.GetNamespace())
})
s.Run("omits managed fields", func() {
s.Nilf(decoded[1].GetManagedFields(), "managed fields should be omitted, got %v", decoded[1].GetManagedFields())
})
})
}
func (s *PodsSuite) TestPodsListInAllNamespacesUnauthorized() {
s.InitMcpClient()
defer restoreAuth(s.T().Context())
client := kubernetes.NewForConfigOrDie(envTestRestConfig)
// Authorize user only for default/configured namespace
r, _ := client.RbacV1().Roles("default").Create(s.T().Context(), &rbacv1.Role{
ObjectMeta: metav1.ObjectMeta{Name: "allow-pods-list"},
Rules: []rbacv1.PolicyRule{{
Verbs: []string{"get", "list"},
APIGroups: []string{""},
Resources: []string{"pods"},
}},
}, metav1.CreateOptions{})
_, _ = client.RbacV1().RoleBindings("default").Create(s.T().Context(), &rbacv1.RoleBinding{
ObjectMeta: metav1.ObjectMeta{Name: "allow-pods-list"},
Subjects: []rbacv1.Subject{{Kind: "User", Name: envTestUser.Name}},
RoleRef: rbacv1.RoleRef{Kind: "Role", Name: r.Name},
}, metav1.CreateOptions{})
// Deny cluster by removing cluster rule
_ = client.RbacV1().ClusterRoles().Delete(s.T().Context(), "allow-all", metav1.DeleteOptions{})
s.Run("pods_list returns pods list for default namespace only", func() {
toolResult, err := s.CallTool("pods_list", map[string]interface{}{})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed %v", toolResult.Content)
})
var decoded []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decoded)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns at least 1 item", func() {
s.GreaterOrEqualf(len(decoded), 1, "invalid pods count, expected at least 1, got %v", len(decoded))
})
s.Run("all pods are in default namespace", func() {
for _, pod := range decoded {
s.Equalf("default", pod.GetNamespace(), "all pods should be in default namespace, got pod %s in namespace %s", pod.GetName(), pod.GetNamespace())
}
})
s.Run("includes a-pod-in-default", func() {
found := false
for _, pod := range decoded {
if pod.GetName() == "a-pod-in-default" {
found = true
break
}
}
s.Truef(found, "expected to find pod a-pod-in-default")
})
})
}
func (s *PodsSuite) TestPodsListInNamespace() {
s.InitMcpClient()
s.Run("pods_list_in_namespace with nil namespace returns error", func() {
toolResult, _ := s.CallTool("pods_list_in_namespace", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to list pods in namespace, missing argument namespace", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_list_in_namespace(namespace=ns-1) returns pods list", func() {
toolResult, err := s.CallTool("pods_list_in_namespace", map[string]interface{}{
"namespace": "ns-1",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decoded []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decoded)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns 1 item", func() {
s.Lenf(decoded, 1, "invalid pods count, expected 1, got %v", len(decoded))
})
s.Run("returns pod in ns-1", func() {
s.Equalf("a-pod-in-ns-1", decoded[0].GetName(), "invalid pod name, expected a-pod-in-ns-1, got %v", decoded[0].GetName())
s.Equalf("ns-1", decoded[0].GetNamespace(), "invalid pod namespace, expected ns-1, got %v", decoded[0].GetNamespace())
})
s.Run("omits managed fields", func() {
s.Nilf(decoded[0].GetManagedFields(), "managed fields should be omitted, got %v", decoded[0].GetManagedFields())
})
})
}
func (s *PodsSuite) TestPodsListDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [ { version = "v1", kind = "Pod" } ]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("pods_list (denied)", func() {
podsList, err := s.CallTool("pods_list", map[string]interface{}{})
s.Run("has error", func() {
s.Truef(podsList.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := podsList.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to list pods in all namespaces:(.+:)? resource not allowed: /v1, Kind=Pod"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("pods_list_in_namespace (denied)", func() {
podsListInNamespace, err := s.CallTool("pods_list_in_namespace", map[string]interface{}{"namespace": "ns-1"})
s.Run("has error", func() {
s.Truef(podsListInNamespace.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := podsListInNamespace.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to list pods in namespace ns-1:(.+:)? resource not allowed: /v1, Kind=Pod"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
}
func (s *PodsSuite) TestPodsListAsTable() {
s.Cfg.ListOutput = "table"
s.InitMcpClient()
s.Run("pods_list (list_output=table)", func() {
podsList, err := s.CallTool("pods_list", map[string]interface{}{})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsList.IsError, "call tool failed")
})
s.Require().NotNil(podsList, "Expected tool result from call")
outPodsList := podsList.Content[0].(mcp.TextContent).Text
s.Run("returns table with header and rows", func() {
lines := strings.Count(outPodsList, "\n")
s.GreaterOrEqualf(lines, 3, "invalid line count, expected at least 3 (1 header, 2+ rows), got %v", lines)
})
s.Run("returns column headers", func() {
expectedHeaders := "NAMESPACE\\s+APIVERSION\\s+KIND\\s+NAME\\s+READY\\s+STATUS\\s+RESTARTS\\s+AGE\\s+IP\\s+NODE\\s+NOMINATED NODE\\s+READINESS GATES\\s+LABELS"
m, e := regexp.MatchString(expectedHeaders, outPodsList)
s.Truef(m, "Expected headers '%s' not found in output:\n%s", expectedHeaders, outPodsList)
s.NoErrorf(e, "Error matching headers regex: %v", e)
})
s.Run("returns formatted row for a-pod-in-ns-1", func() {
expectedRow := "(?<namespace>ns-1)\\s+" +
"(?<apiVersion>v1)\\s+" +
"(?<kind>Pod)\\s+" +
"(?<name>a-pod-in-ns-1)\\s+" +
"(?<ready>0\\/1)\\s+" +
"(?<status>Pending)\\s+" +
"(?<restarts>0)\\s+" +
"(?<age>(\\d+m)?(\\d+s)?)\\s+" +
"(?<ip><none>)\\s+" +
"(?<node><none>)\\s+" +
"(?<nominated_node><none>)\\s+" +
"(?<readiness_gates><none>)\\s+" +
"(?<labels><none>)"
m, e := regexp.MatchString(expectedRow, outPodsList)
s.Truef(m, "Expected row '%s' not found in output:\n%s", expectedRow, outPodsList)
s.NoErrorf(e, "Error matching a-pod-in-ns-1 regex: %v", e)
})
s.Run("returns formatted row for a-pod-in-default", func() {
expectedRow := "(?<namespace>default)\\s+" +
"(?<apiVersion>v1)\\s+" +
"(?<kind>Pod)\\s+" +
"(?<name>a-pod-in-default)\\s+" +
"(?<ready>0\\/1)\\s+" +
"(?<status>Pending)\\s+" +
"(?<restarts>0)\\s+" +
"(?<age>(\\d+m)?(\\d+s)?)\\s+" +
"(?<ip><none>)\\s+" +
"(?<node><none>)\\s+" +
"(?<nominated_node><none>)\\s+" +
"(?<readiness_gates><none>)\\s+" +
"(?<labels>app=nginx)"
m, e := regexp.MatchString(expectedRow, outPodsList)
s.Truef(m, "Expected row '%s' not found in output:\n%s", expectedRow, outPodsList)
s.NoErrorf(e, "Error matching a-pod-in-default regex: %v", e)
})
})
s.Run("pods_list_in_namespace (list_output=table)", func() {
podsListInNamespace, err := s.CallTool("pods_list_in_namespace", map[string]interface{}{
"namespace": "ns-1",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsListInNamespace.IsError, "call tool failed")
})
s.Require().NotNil(podsListInNamespace, "Expected tool result from call")
outPodsListInNamespace := podsListInNamespace.Content[0].(mcp.TextContent).Text
s.Run("returns table with header and row", func() {
lines := strings.Count(outPodsListInNamespace, "\n")
s.GreaterOrEqualf(lines, 1, "invalid line count, expected at least 1 (1 header, 1+ rows), got %v", lines)
})
s.Run("returns column headers", func() {
expectedHeaders := "NAMESPACE\\s+APIVERSION\\s+KIND\\s+NAME\\s+READY\\s+STATUS\\s+RESTARTS\\s+AGE\\s+IP\\s+NODE\\s+NOMINATED NODE\\s+READINESS GATES\\s+LABELS"
m, e := regexp.MatchString(expectedHeaders, outPodsListInNamespace)
s.Truef(m, "Expected headers '%s' not found in output:\n%s", expectedHeaders, outPodsListInNamespace)
s.NoErrorf(e, "Error matching headers regex: %v", e)
})
s.Run("returns formatted row", func() {
expectedRow := "(?<namespace>ns-1)\\s+" +
"(?<apiVersion>v1)\\s+" +
"(?<kind>Pod)\\s+" +
"(?<name>a-pod-in-ns-1)\\s+" +
"(?<ready>0\\/1)\\s+" +
"(?<status>Pending)\\s+" +
"(?<restarts>0)\\s+" +
"(?<age>(\\d+m)?(\\d+s)?)\\s+" +
"(?<ip><none>)\\s+" +
"(?<node><none>)\\s+" +
"(?<nominated_node><none>)\\s+" +
"(?<readiness_gates><none>)\\s+" +
"(?<labels><none>)"
m, e := regexp.MatchString(expectedRow, outPodsListInNamespace)
s.Truef(m, "Expected row '%s' not found in output:\n%s", expectedRow, outPodsListInNamespace)
s.NoErrorf(e, "Error matching formatted row regex: %v", e)
})
})
}
func (s *PodsSuite) TestPodsGet() {
s.InitMcpClient()
s.Run("pods_get with nil name returns error", func() {
toolResult, _ := s.CallTool("pods_get", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get pod, missing argument name", toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_get(name=not-found) with not found name returns error", func() {
toolResult, _ := s.CallTool("pods_get", map[string]interface{}{"name": "not-found"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get pod not-found in namespace : pods \"not-found\" not found", toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_get(name=a-pod-in-default, namespace=nil), uses configured namespace", func() {
podsGetNilNamespace, err := s.CallTool("pods_get", map[string]interface{}{
"name": "a-pod-in-default",
})
s.Run("returns pod", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsGetNilNamespace.IsError, "call tool failed")
})
var decodedNilNamespace unstructured.Unstructured
err = yaml.Unmarshal([]byte(podsGetNilNamespace.Content[0].(mcp.TextContent).Text), &decodedNilNamespace)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns pod in default", func() {
s.Equalf("a-pod-in-default", decodedNilNamespace.GetName(), "invalid pod name, expected a-pod-in-default, got %v", decodedNilNamespace.GetName())
s.Equalf("default", decodedNilNamespace.GetNamespace(), "invalid pod namespace, expected default, got %v", decodedNilNamespace.GetNamespace())
})
s.Run("omits managed fields", func() {
s.Nilf(decodedNilNamespace.GetManagedFields(), "managed fields should be omitted, got %v", decodedNilNamespace.GetManagedFields())
})
})
s.Run("pods_get(name=a-pod-in-default, namespace=ns-1)", func() {
podsGetInNamespace, err := s.CallTool("pods_get", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
})
s.Run("returns pod", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsGetInNamespace.IsError, "call tool failed")
})
var decodedInNamespace unstructured.Unstructured
err = yaml.Unmarshal([]byte(podsGetInNamespace.Content[0].(mcp.TextContent).Text), &decodedInNamespace)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns pod in ns-1", func() {
s.Equalf("a-pod-in-ns-1", decodedInNamespace.GetName(), "invalid pod name, expected a-pod-in-ns-1, got %v", decodedInNamespace.GetName())
s.Equalf("ns-1", decodedInNamespace.GetNamespace(), "invalid pod namespace, expected ns-1, got %v", decodedInNamespace.GetNamespace())
})
})
}
func (s *PodsSuite) TestPodsGetDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [ { version = "v1", kind = "Pod" } ]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("pods_get (denied)", func() {
podsGet, err := s.CallTool("pods_get", map[string]interface{}{"name": "a-pod-in-default"})
s.Run("has error", func() {
s.Truef(podsGet.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := podsGet.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get pod a-pod-in-default in namespace :(.+:)? resource not allowed: /v1, Kind=Pod"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
}
func (s *PodsSuite) TestPodsDelete() {
s.InitMcpClient()
s.Run("pods_delete with nil name returns error", func() {
toolResult, _ := s.CallTool("pods_delete", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to delete pod, missing argument name", toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_delete(name=not-found) with not found name returns error", func() {
toolResult, _ := s.CallTool("pods_delete", map[string]interface{}{"name": "not-found"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to delete pod not-found in namespace : pods \"not-found\" not found", toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_delete(name=a-pod-to-delete, namespace=nil), uses configured namespace", func() {
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
_, _ = kc.CoreV1().Pods("default").Create(s.T().Context(), &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "a-pod-to-delete"},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}}},
}, metav1.CreateOptions{})
podsDeleteNilNamespace, err := s.CallTool("pods_delete", map[string]interface{}{
"name": "a-pod-to-delete",
})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsDeleteNilNamespace.IsError, "call tool failed")
s.Equalf("Pod deleted successfully", podsDeleteNilNamespace.Content[0].(mcp.TextContent).Text, "invalid tool result content, got %v", podsDeleteNilNamespace.Content[0].(mcp.TextContent).Text)
})
s.Run("deletes Pod", func() {
p, pErr := kc.CoreV1().Pods("default").Get(s.T().Context(), "a-pod-to-delete", metav1.GetOptions{})
s.Truef(pErr != nil || p == nil || p.DeletionTimestamp != nil, "Pod not deleted")
})
})
s.Run("pods_delete(name=a-pod-to-delete-in-ns-1, namespace=ns-1)", func() {
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
_, _ = kc.CoreV1().Pods("ns-1").Create(s.T().Context(), &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "a-pod-to-delete-in-ns-1"},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}}},
}, metav1.CreateOptions{})
podsDeleteInNamespace, err := s.CallTool("pods_delete", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-to-delete-in-ns-1",
})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsDeleteInNamespace.IsError, "call tool failed")
s.Equalf("Pod deleted successfully", podsDeleteInNamespace.Content[0].(mcp.TextContent).Text, "invalid tool result content, got %v", podsDeleteInNamespace.Content[0].(mcp.TextContent).Text)
})
s.Run("deletes Pod", func() {
p, pErr := kc.CoreV1().Pods("ns-1").Get(s.T().Context(), "a-pod-to-delete-in-ns-1", metav1.GetOptions{})
s.Truef(pErr != nil || p == nil || p.DeletionTimestamp != nil, "Pod not deleted")
})
})
s.Run("pods_delete(name=a-managed-pod-to-delete, namespace=ns-1) with managed pod", func() {
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
managedLabels := map[string]string{
"app.kubernetes.io/managed-by": "kubernetes-mcp-server",
"app.kubernetes.io/name": "a-manged-pod-to-delete",
}
_, _ = kc.CoreV1().Pods("default").Create(s.T().Context(), &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "a-managed-pod-to-delete", Labels: managedLabels},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}}},
}, metav1.CreateOptions{})
_, _ = kc.CoreV1().Services("default").Create(s.T().Context(), &corev1.Service{
ObjectMeta: metav1.ObjectMeta{Name: "a-managed-service-to-delete", Labels: managedLabels},
Spec: corev1.ServiceSpec{Selector: managedLabels, Ports: []corev1.ServicePort{{Port: 80}}},
}, metav1.CreateOptions{})
podsDeleteManaged, err := s.CallTool("pods_delete", map[string]interface{}{
"name": "a-managed-pod-to-delete",
})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsDeleteManaged.IsError, "call tool failed")
s.Equalf("Pod deleted successfully", podsDeleteManaged.Content[0].(mcp.TextContent).Text, "invalid tool result content, got %v", podsDeleteManaged.Content[0].(mcp.TextContent).Text)
})
s.Run("deletes Pod and Service", func() {
p, pErr := kc.CoreV1().Pods("default").Get(s.T().Context(), "a-managed-pod-to-delete", metav1.GetOptions{})
s.Truef(pErr != nil || p == nil || p.DeletionTimestamp != nil, "Pod not deleted")
svc, sErr := kc.CoreV1().Services("default").Get(s.T().Context(), "a-managed-service-to-delete", metav1.GetOptions{})
s.Truef(sErr != nil || svc == nil || svc.DeletionTimestamp != nil, "Service not deleted")
})
})
}
func (s *PodsSuite) TestPodsDeleteDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [ { version = "v1", kind = "Pod" } ]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("pods_delete (denied)", func() {
podsDelete, err := s.CallTool("pods_delete", map[string]interface{}{"name": "a-pod-in-default"})
s.Run("has error", func() {
s.Truef(podsDelete.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := podsDelete.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to delete pod a-pod-in-default in namespace :(.+:)? resource not allowed: /v1, Kind=Pod"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
}
func (s *PodsSuite) TestPodsDeleteInOpenShift() {
s.Require().NoError(EnvTestInOpenShift(s.T().Context()), "Expected to configure test for OpenShift")
s.T().Cleanup(func() {
s.Require().NoError(EnvTestInOpenShiftClear(s.T().Context()), "Expected to clear OpenShift test configuration")
})
s.InitMcpClient()
s.Run("pods_delete with managed pod in OpenShift", func() {
managedLabels := map[string]string{
"app.kubernetes.io/managed-by": "kubernetes-mcp-server",
"app.kubernetes.io/name": "a-manged-pod-to-delete",
}
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
_, _ = kc.CoreV1().Pods("default").Create(s.T().Context(), &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "a-managed-pod-to-delete-in-openshift", Labels: managedLabels},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}}},
}, metav1.CreateOptions{})
dynamicClient := dynamic.NewForConfigOrDie(envTestRestConfig)
_, _ = dynamicClient.Resource(schema.GroupVersionResource{Group: "route.openshift.io", Version: "v1", Resource: "routes"}).
Namespace("default").Create(s.T().Context(), &unstructured.Unstructured{Object: map[string]interface{}{
"apiVersion": "route.openshift.io/v1",
"kind": "Route",
"metadata": map[string]interface{}{
"name": "a-managed-route-to-delete",
"labels": managedLabels,
},
}}, metav1.CreateOptions{})
podsDeleteManagedOpenShift, err := s.CallTool("pods_delete", map[string]interface{}{
"name": "a-managed-pod-to-delete-in-openshift",
})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsDeleteManagedOpenShift.IsError, "call tool failed")
s.Equalf("Pod deleted successfully", podsDeleteManagedOpenShift.Content[0].(mcp.TextContent).Text,
"invalid tool result content, got %v", podsDeleteManagedOpenShift.Content[0].(mcp.TextContent).Text)
})
s.Run("deletes Pod and Route", func() {
p, pErr := kc.CoreV1().Pods("default").Get(s.T().Context(), "a-managed-pod-to-delete-in-openshift", metav1.GetOptions{})
s.False(pErr == nil && p != nil && p.DeletionTimestamp == nil, "Pod not deleted")
r, rErr := dynamicClient.
Resource(schema.GroupVersionResource{Group: "route.openshift.io", Version: "v1", Resource: "routes"}).
Namespace("default").Get(s.T().Context(), "a-managed-route-to-delete", metav1.GetOptions{})
s.False(rErr == nil && r != nil && r.GetDeletionTimestamp() == nil, "Route not deleted")
})
})
}
func (s *PodsSuite) TestPodsLog() {
s.InitMcpClient()
s.Run("pods_log with nil name returns error", func() {
toolResult, _ := s.CallTool("pods_log", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get pod log, missing argument name", toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_log with not found name returns error", func() {
toolResult, _ := s.CallTool("pods_log", map[string]interface{}{"name": "not-found"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get pod not-found log in namespace : pods \"not-found\" not found", toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_log(name=a-pod-in-default, namespace=nil), uses configured namespace", func() {
podsLogNilNamespace, err := s.CallTool("pods_log", map[string]interface{}{
"name": "a-pod-in-default",
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsLogNilNamespace.IsError, "call tool failed")
})
s.Run("pods_log(name=a-pod-in-ns-1, namespace=ns-1)", func() {
podsLogInNamespace, err := s.CallTool("pods_log", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsLogInNamespace.IsError, "call tool failed")
})
s.Run("pods_log(name=a-pod-in-ns-1, namespace=ns-1, container=nginx)", func() {
podsContainerLogInNamespace, err := s.CallTool("pods_log", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
"container": "nginx",
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsContainerLogInNamespace.IsError, "call tool failed")
})
s.Run("with non existing container returns error", func() {
toolResult, err := s.CallTool("pods_log", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
"container": "a-not-existing-container",
})
s.Nilf(err, "call tool should not return error object")
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get pod a-pod-in-ns-1 log in namespace ns-1: container a-not-existing-container is not valid for pod a-pod-in-ns-1", toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("pods_log(previous=true) returns previous pod log", func() {
podsPreviousLogInNamespace, err := s.CallTool("pods_log", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
"previous": true,
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsPreviousLogInNamespace.IsError, "call tool failed")
})
s.Run("pods_log(previous=false) returns current pod log", func() {
podsPreviousLogFalse, err := s.CallTool("pods_log", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
"previous": false,
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsPreviousLogFalse.IsError, "call tool failed")
})
s.Run("pods_log(tail=50) returns pod log", func() {
podsTailLines, err := s.CallTool("pods_log", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
"tail": 50,
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(podsTailLines.IsError, "call tool failed")
})
s.Run("with invalid tail returns error", func() {
podsInvalidTailLines, _ := s.CallTool("pods_log", map[string]interface{}{
"namespace": "ns-1",
"name": "a-pod-in-ns-1",
"tail": "invalid",
})
s.Truef(podsInvalidTailLines.IsError, "call tool should fail")
expectedErrorMsg := "failed to parse tail parameter: expected integer"
errMsg := podsInvalidTailLines.Content[0].(mcp.TextContent).Text
s.Containsf(errMsg, expectedErrorMsg, "unexpected error message, expected to contain '%s', got '%s'", expectedErrorMsg, errMsg)
})
}
func (s *PodsSuite) TestPodsLogDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [ { version = "v1", kind = "Pod" } ]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("pods_log (denied)", func() {
podsLog, err := s.CallTool("pods_log", map[string]interface{}{"name": "a-pod-in-default"})
s.Run("has error", func() {
s.Truef(podsLog.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := podsLog.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get pod a-pod-in-default log in namespace :(.+:)? resource not allowed: /v1, Kind=Pod"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
}
func (s *PodsSuite) TestPodsListWithLabelSelector() {
s.InitMcpClient()
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
// Create pods with labels
_, _ = kc.CoreV1().Pods("default").Create(s.T().Context(), &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "pod-with-labels",
Labels: map[string]string{"app": "test", "env": "dev"},
},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}}},
}, metav1.CreateOptions{})
_, _ = kc.CoreV1().Pods("ns-1").Create(s.T().Context(), &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "another-pod-with-labels",
Labels: map[string]string{"app": "test", "env": "prod"},
},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}}},
}, metav1.CreateOptions{})
s.Run("pods_list(labelSelector=app=test) returns filtered pods from configured namespace", func() {
toolResult, err := s.CallTool("pods_list", map[string]interface{}{
"labelSelector": "app=test",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decoded []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decoded)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns 2 pods", func() {
s.Lenf(decoded, 2, "invalid pods count, expected 2, got %v", len(decoded))
})
})
s.Run("pods_list_in_namespace(labelSelector=env=prod, namespace=ns-1) returns filtered pods", func() {
toolResult, err := s.CallTool("pods_list_in_namespace", map[string]interface{}{
"namespace": "ns-1",
"labelSelector": "env=prod",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decoded []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decoded)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns 1 pod", func() {
s.Lenf(decoded, 1, "invalid pods count, expected 1, got %v", len(decoded))
})
s.Run("returns another-pod-with-labels", func() {
s.Equalf("another-pod-with-labels", decoded[0].GetName(), "invalid pod name, expected another-pod-with-labels, got %v", decoded[0].GetName())
})
})
s.Run("pods_list(labelSelector=app=test,env=prod) with multiple label selectors returns filtered pods", func() {
toolResult, err := s.CallTool("pods_list", map[string]interface{}{
"labelSelector": "app=test,env=prod",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decoded []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decoded)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns 1 pod", func() {
s.Lenf(decoded, 1, "invalid pods count, expected 1, got %v", len(decoded))
})
s.Run("returns another-pod-with-labels", func() {
s.Equalf("another-pod-with-labels", decoded[0].GetName(), "invalid pod name, expected another-pod-with-labels, got %v", decoded[0].GetName())
})
})
}
func TestPods(t *testing.T) {
suite.Run(t, new(PodsSuite))
}
```
--------------------------------------------------------------------------------
/pkg/mcp/kubevirt_test.go:
--------------------------------------------------------------------------------
```go
package mcp
import (
"fmt"
"strings"
"testing"
"github.com/BurntSushi/toml"
"github.com/containers/kubernetes-mcp-server/internal/test"
kubevirttesting "github.com/containers/kubernetes-mcp-server/pkg/kubevirt/testing"
"github.com/mark3labs/mcp-go/mcp"
"github.com/stretchr/testify/suite"
"golang.org/x/sync/errgroup"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
"sigs.k8s.io/yaml"
)
var kubevirtApis = []schema.GroupVersionResource{
{Group: "kubevirt.io", Version: "v1", Resource: "virtualmachines"},
{Group: "cdi.kubevirt.io", Version: "v1beta1", Resource: "datasources"},
{Group: "instancetype.kubevirt.io", Version: "v1beta1", Resource: "virtualmachineclusterinstancetypes"},
{Group: "instancetype.kubevirt.io", Version: "v1beta1", Resource: "virtualmachineinstancetypes"},
{Group: "instancetype.kubevirt.io", Version: "v1beta1", Resource: "virtualmachineclusterpreferences"},
{Group: "instancetype.kubevirt.io", Version: "v1beta1", Resource: "virtualmachinepreferences"},
}
type KubevirtSuite struct {
BaseMcpSuite
}
func (s *KubevirtSuite) SetupSuite() {
ctx := s.T().Context()
tasks, _ := errgroup.WithContext(ctx)
for _, api := range kubevirtApis {
gvr := api // capture loop variable
tasks.Go(func() error { return EnvTestEnableCRD(ctx, gvr.Group, gvr.Version, gvr.Resource) })
}
s.Require().NoError(tasks.Wait())
_, err := kubernetes.NewForConfigOrDie(envTestRestConfig).CoreV1().Namespaces().
Create(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "openshift-virtualization-os-images"}}, metav1.CreateOptions{})
s.Require().NoError(err, "failed to create test namespace openshift-virtualization-os-images")
}
func (s *KubevirtSuite) TearDownSuite() {
tasks, _ := errgroup.WithContext(s.T().Context())
for _, api := range kubevirtApis {
gvr := api // capture loop variable
tasks.Go(func() error { return EnvTestDisableCRD(s.T().Context(), gvr.Group, gvr.Version, gvr.Resource) })
}
s.Require().NoError(tasks.Wait())
}
func (s *KubevirtSuite) SetupTest() {
s.BaseMcpSuite.SetupTest()
s.Require().NoError(toml.Unmarshal([]byte(`
toolsets = [ "kubevirt" ]
`), s.Cfg), "Expected to parse toolsets config")
s.InitMcpClient()
}
func (s *KubevirtSuite) TestCreate() {
s.Run("vm_create missing required params", func() {
testCases := []string{"name", "namespace"}
for _, param := range testCases {
s.Run("missing "+param, func() {
params := map[string]interface{}{
"name": "test-vm",
"namespace": "default",
}
delete(params, param)
toolResult, err := s.CallTool("vm_create", params)
s.Require().Nilf(err, "call tool failed %v", err)
s.Truef(toolResult.IsError, "expected call tool to fail due to missing %s", param)
s.Equal(toolResult.Content[0].(mcp.TextContent).Text, param+" parameter required")
})
}
})
s.Run("vm_create with default settings", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm",
"namespace": "default",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("quay.io/containerdisks/fedora:latest", test.FieldString(vm, "spec.template.spec.volumes[0].containerDisk.image"), "invalid default image")
s.Equal("Halted", test.FieldString(vm, "spec.runStrategy"), "invalid default runStrategy")
})
})
s.Run("vm_create(workload=ubuntu, instancetype=u1.medium) with instancetype", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-2",
"namespace": "default",
"workload": "ubuntu",
"instancetype": "u1.medium",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-2", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("quay.io/containerdisks/ubuntu:24.04", test.FieldString(vm, "spec.template.spec.volumes[0].containerDisk.image"), "invalid image for ubuntu workload")
s.Equal("Halted", test.FieldString(vm, "spec.runStrategy"), "invalid default runStrategy")
s.Equal("VirtualMachineClusterInstancetype", test.FieldString(vm, "spec.instancetype.kind"), "invalid memory for u1.medium instanceType")
s.Equal("u1.medium", test.FieldString(vm, "spec.instancetype.name"), "invalid cpu cores for u1.medium instanceType")
})
})
s.Run("vm_create(workload=rhel, preference=rhel.9) with preference", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-3",
"namespace": "default",
"workload": "rhel",
"preference": "rhel.9",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-3", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("rhel", test.FieldString(vm, "spec.template.spec.volumes[0].containerDisk.image"), "invalid image for rhel workload")
s.Equal("Halted", test.FieldString(vm, "spec.runStrategy"), "invalid default runStrategy")
s.Equal("VirtualMachineClusterPreference", test.FieldString(vm, "spec.preference.kind"), "invalid preference kind for rhel.9 preference")
s.Equal("rhel.9", test.FieldString(vm, "spec.preference.name"), "invalid preference name for rhel.9 preference")
})
})
s.Run("vm_create(workload=quay.io/myrepo/myimage:v1.0) with custom container disk", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-4",
"namespace": "default",
"workload": "quay.io/myrepo/myimage:v1.0",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-4", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("quay.io/myrepo/myimage:v1.0", test.FieldString(vm, "spec.template.spec.volumes[0].containerDisk.image"), "invalid image for custom container disk workload")
s.Equal("Halted", test.FieldString(vm, "spec.runStrategy"), "invalid default runStrategy")
})
})
s.Run("with size", func() {
dynamicClient := dynamic.NewForConfigOrDie(envTestRestConfig).Resource(
schema.GroupVersionResource{Group: "instancetype.kubevirt.io", Version: "v1beta1", Resource: "virtualmachineclusterinstancetypes"},
)
instanceTypes := []struct{ instanceType, performance string }{
{"compute", "c1"},
{"general", "u1"},
{"memory", "m1"},
}
for _, size := range []string{"medium", "small", "large"} {
for _, instanceType := range instanceTypes {
labels := map[string]string{}
labels["instancetype.kubevirt.io/class"] = instanceType.instanceType
_, err := dynamicClient.Create(
s.T().Context(),
kubevirttesting.NewUnstructuredInstancetype(fmt.Sprintf("%s.%s", instanceType.performance, size), labels),
metav1.CreateOptions{},
)
s.Require().NoError(err)
}
}
s.Run("vm_create(size=medium) with size hint matching instancetype", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-5",
"namespace": "default",
"size": "medium",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-5", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("VirtualMachineClusterInstancetype", test.FieldString(vm, "spec.instancetype.kind"), "invalid instanceType kind for medium size hint")
s.Equal("u1.medium", test.FieldString(vm, "spec.instancetype.name"), "invalid instanceType name for medium size hint")
})
})
s.Run("vm_create(size=large, performance=compute-optimized) with size and performance hints", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-6",
"namespace": "default",
"size": "large",
"performance": "compute-optimized",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-6", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("VirtualMachineClusterInstancetype", test.FieldString(vm, "spec.instancetype.kind"), "invalid instanceType kind for large size hint")
s.Equal("c1.large", test.FieldString(vm, "spec.instancetype.name"), "invalid instanceType name for large size hint")
})
})
s.Run("vm_create(size=xlarge) with size hint not matching any instancetype", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-7",
"namespace": "default",
"size": "xlarge",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-7", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Falsef(test.FieldExists(vm, "spec.instancetype"), "expected no instancetype to be set for xlarge size hint")
})
})
})
s.Run("with data sources", func() {
dynamicClient := dynamic.NewForConfigOrDie(envTestRestConfig).Resource(
schema.GroupVersionResource{Group: "cdi.kubevirt.io", Version: "v1beta1", Resource: "datasources"},
)
_, err := dynamicClient.Namespace("openshift-virtualization-os-images").Create(
s.T().Context(),
kubevirttesting.NewUnstructuredDataSource("fedora", "openshift-virtualization-os-images", "registry.redhat.io/fedora:latest", "u1.medium", "fedora"),
metav1.CreateOptions{},
)
s.Require().NoError(err)
s.Run("vm_create(workload=fedora) using DataSource with default instancetype and preference", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-8",
"namespace": "default",
"workload": "fedora",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-8", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("Halted", test.FieldString(vm, "spec.runStrategy"), "invalid default runStrategy")
s.Equal("VirtualMachineClusterInstancetype", test.FieldString(vm, "spec.instancetype.kind"), "invalid instanceType kind from DataSource default")
s.Equal("u1.medium", test.FieldString(vm, "spec.instancetype.name"), "invalid instanceType name from DataSource default")
s.Equal("VirtualMachineClusterPreference", test.FieldString(vm, "spec.preference.kind"), "invalid preference kind from DataSource default")
s.Equal("fedora", test.FieldString(vm, "spec.preference.name"), "invalid preference name from DataSource default")
s.Equal("DataSource", test.FieldString(vm, "spec.dataVolumeTemplates[0].spec.sourceRef.kind"), "invalid data source kind in dataVolumeTemplates")
s.Equal("fedora", test.FieldString(vm, "spec.dataVolumeTemplates[0].spec.sourceRef.name"), "invalid data source name in dataVolumeTemplates")
})
})
s.Run("vm_create(workload=rhel) using DataSource partial name match", func() {
_, err := dynamicClient.Namespace("openshift-virtualization-os-images").Create(
s.T().Context(),
kubevirttesting.NewUnstructuredDataSource("rhel9", "openshift-virtualization-os-images", "registry.redhat.io/rhel9:latest", "", "rhel.9"),
metav1.CreateOptions{},
)
s.Require().NoError(err)
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-9",
"namespace": "default",
"workload": "rhel",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-9", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("Halted", test.FieldString(vm, "spec.runStrategy"), "invalid default runStrategy")
s.Equal("VirtualMachineClusterPreference", test.FieldString(vm, "spec.preference.kind"), "invalid preference kind from DataSource default")
s.Equal("rhel.9", test.FieldString(vm, "spec.preference.name"), "invalid preference name from DataSource default")
s.Equal("DataSource", test.FieldString(vm, "spec.dataVolumeTemplates[0].spec.sourceRef.kind"), "invalid data source kind in dataVolumeTemplates")
s.Equal("rhel9", test.FieldString(vm, "spec.dataVolumeTemplates[0].spec.sourceRef.name"), "invalid data source name in dataVolumeTemplates")
})
})
s.Run("vm_create(workload=fedora, size=large) with size hint overriding DataSource default instancetype", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-10",
"namespace": "default",
"workload": "fedora",
"size": "large",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-10", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("Halted", test.FieldString(vm, "spec.runStrategy"), "invalid default runStrategy")
s.Equal("VirtualMachineClusterInstancetype", test.FieldString(vm, "spec.instancetype.kind"), "invalid instanceType kind for large size hint")
s.Equal("u1.large", test.FieldString(vm, "spec.instancetype.name"), "invalid instanceType name for large size hint")
})
})
})
s.Run("with preferences", func() {
dynamicClient := dynamic.NewForConfigOrDie(envTestRestConfig).Resource(
schema.GroupVersionResource{Group: "instancetype.kubevirt.io", Version: "v1beta1", Resource: "virtualmachineclusterpreferences"},
)
for _, preference := range []*unstructured.Unstructured{
kubevirttesting.NewUnstructuredPreference("rhel.9", false),
kubevirttesting.NewUnstructuredPreference("fedora", false),
} {
_, err := dynamicClient.Create(s.T().Context(), preference, metav1.CreateOptions{})
s.Require().NoError(err)
}
s.Run("vm_create(workload=rhel) auto-select preference matching workload name", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-11",
"namespace": "default",
"workload": "rhel",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-11", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("VirtualMachineClusterPreference", test.FieldString(vm, "spec.preference.kind"), "invalid preference kind for rhel.9 preference")
s.Equal("rhel.9", test.FieldString(vm, "spec.preference.name"), "invalid preference name for rhel.9 preference")
})
})
s.Run("vm_create(workload=fedora, preference=custom.preference) with explicit preference overriding auto-selected preference", func() {
toolResult, err := s.CallTool("vm_create", map[string]interface{}{
"name": "test-vm-12",
"namespace": "default",
"workload": "fedora",
"preference": "custom.preference",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine created successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
vm := &decodedResult[0]
s.Equal("test-vm-12", vm.GetName(), "invalid resource name")
s.Equal("default", vm.GetNamespace(), "invalid resource namespace")
s.NotEmptyf(vm.GetUID(), "invalid uid, got %v", vm.GetUID())
s.Equal("custom.preference", test.FieldString(vm, "spec.preference.name"), "invalid preference name for explicit custom.preference")
})
})
})
}
func (s *KubevirtSuite) TestVMLifecycle() {
// Create a test VM in Halted state for start tests
dynamicClient := dynamic.NewForConfigOrDie(envTestRestConfig)
vm := &unstructured.Unstructured{}
vm.SetUnstructuredContent(map[string]interface{}{
"apiVersion": "kubevirt.io/v1",
"kind": "VirtualMachine",
"metadata": map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
},
"spec": map[string]interface{}{
"runStrategy": "Halted",
},
})
_, err := dynamicClient.Resource(schema.GroupVersionResource{
Group: "kubevirt.io",
Version: "v1",
Resource: "virtualmachines",
}).Namespace("default").Create(s.T().Context(), vm, metav1.CreateOptions{})
s.Require().NoError(err, "failed to create test VM")
s.Run("vm_lifecycle missing required params", func() {
testCases := []string{"name", "namespace", "action"}
for _, param := range testCases {
s.Run("missing "+param, func() {
params := map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "start",
}
delete(params, param)
toolResult, err := s.CallTool("vm_lifecycle", params)
s.Require().Nilf(err, "call tool failed %v", err)
s.Truef(toolResult.IsError, "expected call tool to fail due to missing %s", param)
s.Equal(toolResult.Content[0].(mcp.TextContent).Text, param+" parameter required")
})
}
})
s.Run("vm_lifecycle invalid action", func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "invalid",
})
s.Require().Nilf(err, "call tool failed %v", err)
s.Truef(toolResult.IsError, "expected call tool to fail due to invalid action")
s.Truef(strings.Contains(toolResult.Content[0].(mcp.TextContent).Text, "invalid action"),
"Expected invalid action message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("vm_lifecycle action=start on halted VM", func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "start",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine started successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
s.Equal("test-vm-lifecycle", decodedResult[0].GetName(), "invalid resource name")
s.Equal("default", decodedResult[0].GetNamespace(), "invalid resource namespace")
s.Equal("Always",
decodedResult[0].Object["spec"].(map[string]interface{})["runStrategy"].(string),
"expected runStrategy to be Always after start")
})
})
s.Run("vm_lifecycle action=start on already running VM (idempotent)", func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "start",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content showing VM was already running", func() {
s.Nilf(err, "invalid tool result content %v", err)
expectedPrefix := fmt.Sprintf("# VirtualMachine '%s' in namespace '%s' is already running", "test-vm-lifecycle", "default")
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, expectedPrefix),
"Expected already running message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
s.Equal("Always",
decodedResult[0].Object["spec"].(map[string]interface{})["runStrategy"].(string),
"expected runStrategy to remain Always")
})
})
s.Run("vm_lifecycle action=stop on running VM", func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "stop",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine stopped successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
s.Equal("test-vm-lifecycle", decodedResult[0].GetName(), "invalid resource name")
s.Equal("default", decodedResult[0].GetNamespace(), "invalid resource namespace")
s.Equal("Halted",
decodedResult[0].Object["spec"].(map[string]interface{})["runStrategy"].(string),
"expected runStrategy to be Halted after stop")
})
})
s.Run("vm_lifecycle action=stop on already stopped VM (idempotent)", func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "stop",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content showing VM was already stopped", func() {
s.Nilf(err, "invalid tool result content %v", err)
expectedPrefix := fmt.Sprintf("# VirtualMachine '%s' in namespace '%s' is already stopped", "test-vm-lifecycle", "default")
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, expectedPrefix),
"Expected already stopped message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
s.Equal("Halted",
decodedResult[0].Object["spec"].(map[string]interface{})["runStrategy"].(string),
"expected runStrategy to remain Halted")
})
})
s.Run("vm_lifecycle action=restart on stopped VM", func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "restart",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content showing VM restarted from stopped state", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine restarted successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
s.Equal("Always",
decodedResult[0].Object["spec"].(map[string]interface{})["runStrategy"].(string),
"expected runStrategy to be Always after restart from Halted")
})
})
s.Run("vm_lifecycle action=restart on running VM", func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "test-vm-lifecycle",
"namespace": "default",
"action": "restart",
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(toolResult.IsError, "call tool failed")
})
var decodedResult []unstructured.Unstructured
err = yaml.Unmarshal([]byte(toolResult.Content[0].(mcp.TextContent).Text), &decodedResult)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(toolResult.Content[0].(mcp.TextContent).Text, "# VirtualMachine restarted successfully"),
"Expected success message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
s.Require().Lenf(decodedResult, 1, "invalid resource count, expected 1, got %v", len(decodedResult))
s.Equal("test-vm-lifecycle", decodedResult[0].GetName(), "invalid resource name")
s.Equal("default", decodedResult[0].GetNamespace(), "invalid resource namespace")
s.Equal("Always",
decodedResult[0].Object["spec"].(map[string]interface{})["runStrategy"].(string),
"expected runStrategy to be Always after restart")
})
})
s.Run("vm_lifecycle on non-existent VM", func() {
for _, action := range []string{"start", "stop", "restart"} {
s.Run("action="+action, func() {
toolResult, err := s.CallTool("vm_lifecycle", map[string]interface{}{
"name": "non-existent-vm",
"namespace": "default",
"action": action,
})
s.Nilf(err, "call tool failed %v", err)
s.Truef(toolResult.IsError, "expected call tool to fail for non-existent VM")
s.Truef(strings.Contains(toolResult.Content[0].(mcp.TextContent).Text, "failed to get VirtualMachine"),
"Expected error message about VM not found, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
}
})
}
func TestKubevirt(t *testing.T) {
suite.Run(t, new(KubevirtSuite))
}
```
--------------------------------------------------------------------------------
/pkg/mcp/resources_test.go:
--------------------------------------------------------------------------------
```go
package mcp
import (
"regexp"
"strings"
"testing"
"github.com/BurntSushi/toml"
"github.com/mark3labs/mcp-go/mcp"
"github.com/stretchr/testify/suite"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
v1 "k8s.io/api/rbac/v1"
apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
"k8s.io/utils/ptr"
"sigs.k8s.io/yaml"
)
type ResourcesSuite struct {
BaseMcpSuite
}
func (s *ResourcesSuite) TestResourcesList() {
s.InitMcpClient()
s.Run("resources_list with missing apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_list", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to list resources, missing argument apiVersion", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_list with missing kind returns error", func() {
toolResult, _ := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "v1"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to list resources, missing argument kind", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_list with invalid apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "invalid/api/version", "kind": "Pod"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to list resources, invalid argument apiVersion", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_list with nonexistent apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "custom.non.existent.example.com/v1", "kind": "Custom"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf(`failed to list resources: no matches for kind "Custom" in version "custom.non.existent.example.com/v1"`,
toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_list(apiVersion=v1, kind=Namespace) returns namespaces", func() {
namespaces, err := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "v1", "kind": "Namespace"})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(namespaces.IsError, "call tool failed")
})
var decodedNamespaces []unstructured.Unstructured
err = yaml.Unmarshal([]byte(namespaces.Content[0].(mcp.TextContent).Text), &decodedNamespaces)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns more than 2 items", func() {
s.Truef(len(decodedNamespaces) >= 3, "invalid namespace count, expected >2, got %v", len(decodedNamespaces))
})
})
s.Run("resources_list with label selector returns filtered pods", func() {
s.Run("list pods with app=nginx label", func() {
result, err := s.CallTool("resources_list", map[string]interface{}{
"apiVersion": "v1",
"kind": "Pod",
"namespace": "default",
"labelSelector": "app=nginx",
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(result.IsError, "call tool failed")
var decodedPods []unstructured.Unstructured
err = yaml.Unmarshal([]byte(result.Content[0].(mcp.TextContent).Text), &decodedPods)
s.Nilf(err, "invalid tool result content %v", err)
s.Lenf(decodedPods, 1, "expected 1 pod, got %d", len(decodedPods))
s.Equalf("a-pod-in-default", decodedPods[0].GetName(), "expected a-pod-in-default, got %s", decodedPods[0].GetName())
})
s.Run("list pods with multiple label selectors", func() {
result, err := s.CallTool("resources_list", map[string]interface{}{
"apiVersion": "v1",
"kind": "Pod",
"namespace": "default",
"labelSelector": "test-label=test-value,another=value",
})
s.Nilf(err, "call tool failed %v", err)
s.Falsef(result.IsError, "call tool failed")
var decodedPods []unstructured.Unstructured
err = yaml.Unmarshal([]byte(result.Content[0].(mcp.TextContent).Text), &decodedPods)
s.Nilf(err, "invalid tool result content %v", err)
s.Lenf(decodedPods, 0, "expected 0 pods, got %d", len(decodedPods))
})
})
}
func (s *ResourcesSuite) TestResourcesListDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [
{ version = "v1", kind = "Secret" },
{ group = "rbac.authorization.k8s.io", version = "v1" }
]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("resources_list (denied by kind)", func() {
deniedByKind, err := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "v1", "kind": "Secret"})
s.Run("has error", func() {
s.Truef(deniedByKind.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByKind.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to list resources:(.+:)? resource not allowed: /v1, Kind=Secret"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_list (denied by group)", func() {
deniedByGroup, err := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "rbac.authorization.k8s.io/v1", "kind": "Role"})
s.Run("has error", func() {
s.Truef(deniedByGroup.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByGroup.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to list resources:(.+:)? resource not allowed: rbac.authorization.k8s.io/v1, Kind=Role"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_list (not denied) returns list", func() {
allowedResource, _ := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "v1", "kind": "Namespace"})
s.Falsef(allowedResource.IsError, "call tool should not fail")
})
}
func (s *ResourcesSuite) TestResourcesListAsTable() {
s.Cfg.ListOutput = "table"
s.Require().NoError(EnvTestInOpenShift(s.T().Context()), "Expected to configure test for OpenShift")
s.T().Cleanup(func() {
s.Require().NoError(EnvTestInOpenShiftClear(s.T().Context()), "Expected to clear OpenShift test configuration")
})
s.InitMcpClient()
s.Run("resources_list(apiVersion=v1, kind=ConfigMap) (list_output=table)", func() {
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
_, _ = kc.CoreV1().ConfigMaps("default").Create(s.T().Context(), &corev1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{Name: "a-configmap-to-list-as-table", Labels: map[string]string{"resource": "config-map"}},
Data: map[string]string{"key": "value"},
}, metav1.CreateOptions{})
configMapList, err := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "v1", "kind": "ConfigMap"})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(configMapList.IsError, "call tool failed")
})
s.Require().NotNil(configMapList, "Expected tool result from call")
outConfigMapList := configMapList.Content[0].(mcp.TextContent).Text
s.Run("returns column headers for ConfigMap list", func() {
expectedHeaders := "NAMESPACE\\s+APIVERSION\\s+KIND\\s+NAME\\s+DATA\\s+AGE\\s+LABELS"
m, e := regexp.MatchString(expectedHeaders, outConfigMapList)
s.Truef(m, "Expected headers '%s' not found in output:\n%s", expectedHeaders, outConfigMapList)
s.NoErrorf(e, "Error matching headers regex: %v", e)
})
s.Run("returns formatted row for a-configmap-to-list-as-table", func() {
expectedRow := "(?<namespace>default)\\s+" +
"(?<apiVersion>v1)\\s+" +
"(?<kind>ConfigMap)\\s+" +
"(?<name>a-configmap-to-list-as-table)\\s+" +
"(?<data>1)\\s+" +
"(?<age>(\\d+m)?(\\d+s)?)\\s+" +
"(?<labels>resource=config-map)"
m, e := regexp.MatchString(expectedRow, outConfigMapList)
s.Truef(m, "Expected row '%s' not found in output:\n%s", expectedRow, outConfigMapList)
s.NoErrorf(e, "Error matching row regex: %v", e)
})
})
s.Run("resources_list(apiVersion=route.openshift.io/v1, kind=Route) (list_output=table)", func() {
_, _ = dynamic.NewForConfigOrDie(envTestRestConfig).
Resource(schema.GroupVersionResource{Group: "route.openshift.io", Version: "v1", Resource: "routes"}).
Namespace("default").
Create(s.T().Context(), &unstructured.Unstructured{Object: map[string]interface{}{
"apiVersion": "route.openshift.io/v1",
"kind": "Route",
"metadata": map[string]interface{}{
"name": "an-openshift-route-to-list-as-table",
},
}}, metav1.CreateOptions{})
routeList, err := s.CallTool("resources_list", map[string]interface{}{"apiVersion": "route.openshift.io/v1", "kind": "Route"})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(routeList.IsError, "call tool failed")
})
s.Require().NotNil(routeList, "Expected tool result from call")
outRouteList := routeList.Content[0].(mcp.TextContent).Text
s.Run("returns column headers for Route list", func() {
expectedHeaders := "NAMESPACE\\s+APIVERSION\\s+KIND\\s+NAME\\s+AGE\\s+LABELS"
m, e := regexp.MatchString(expectedHeaders, outRouteList)
s.Truef(m, "Expected headers '%s' not found in output:\n%s", expectedHeaders, outRouteList)
s.NoErrorf(e, "Error matching headers regex: %v", e)
})
s.Run("returns formatted row for an-openshift-route-to-list-as-table", func() {
expectedRow := "(?<namespace>default)\\s+" +
"(?<apiVersion>route.openshift.io/v1)\\s+" +
"(?<kind>Route)\\s+" +
"(?<name>an-openshift-route-to-list-as-table)\\s+" +
"(?<age>(\\d+m)?(\\d+s)?)\\s+" +
"(?<labels><none>)"
m, e := regexp.MatchString(expectedRow, outRouteList)
s.Truef(m, "Expected row '%s' not found in output:\n%s", expectedRow, outRouteList)
s.NoErrorf(e, "Error matching row regex: %v", e)
})
})
}
func (s *ResourcesSuite) TestResourcesGet() {
s.InitMcpClient()
s.Run("resources_get with missing apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_get", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get resource, missing argument apiVersion", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_get with missing kind returns error", func() {
toolResult, _ := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "v1"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get resource, missing argument kind", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_get with invalid apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "invalid/api/version", "kind": "Pod", "name": "a-pod"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get resource, invalid argument apiVersion", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_get with nonexistent apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "custom.non.existent.example.com/v1", "kind": "Custom", "name": "a-custom"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf(`failed to get resource: no matches for kind "Custom" in version "custom.non.existent.example.com/v1"`,
toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_get with missing name returns error", func() {
toolResult, _ := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "v1", "kind": "Namespace"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get resource, missing argument name", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_get returns namespace", func() {
namespace, err := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "v1", "kind": "Namespace", "name": "default"})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(namespace.IsError, "call tool failed")
})
var decodedNamespace unstructured.Unstructured
err = yaml.Unmarshal([]byte(namespace.Content[0].(mcp.TextContent).Text), &decodedNamespace)
s.Run("has yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
})
s.Run("returns default namespace", func() {
s.Equalf("default", decodedNamespace.GetName(), "invalid namespace name, expected default, got %v", decodedNamespace.GetName())
})
})
}
func (s *ResourcesSuite) TestResourcesGetDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [
{ version = "v1", kind = "Secret" },
{ group = "rbac.authorization.k8s.io", version = "v1" }
]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
_, _ = kc.CoreV1().Secrets("default").Create(s.T().Context(), &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{Name: "denied-secret"},
}, metav1.CreateOptions{})
_, _ = kc.RbacV1().Roles("default").Create(s.T().Context(), &v1.Role{
ObjectMeta: metav1.ObjectMeta{Name: "denied-role"},
}, metav1.CreateOptions{})
s.Run("resources_get (denied by kind)", func() {
deniedByKind, err := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "v1", "kind": "Secret", "namespace": "default", "name": "denied-secret"})
s.Run("has error", func() {
s.Truef(deniedByKind.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByKind.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get resource:(.+:)? resource not allowed: /v1, Kind=Secret"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_get (denied by group)", func() {
deniedByGroup, err := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "rbac.authorization.k8s.io/v1", "kind": "Role", "namespace": "default", "name": "denied-role"})
s.Run("has error", func() {
s.Truef(deniedByGroup.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByGroup.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get resource:(.+:)? resource not allowed: rbac.authorization.k8s.io/v1, Kind=Role"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_get (not denied) returns resource", func() {
allowedResource, err := s.CallTool("resources_get", map[string]interface{}{"apiVersion": "v1", "kind": "Namespace", "name": "default"})
s.Falsef(allowedResource.IsError, "call tool should not fail")
s.Nilf(err, "call tool should not return error object")
})
}
func (s *ResourcesSuite) TestResourcesCreateOrUpdate() {
s.InitMcpClient()
client := kubernetes.NewForConfigOrDie(envTestRestConfig)
s.Run("resources_create_or_update with nil resource returns error", func() {
toolResult, _ := s.CallTool("resources_create_or_update", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to create or update resources, missing argument resource", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_create_or_update with empty resource returns error", func() {
toolResult, _ := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": ""})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to create or update resources, missing argument resource", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_create_or_update with valid namespaced yaml resource", func() {
configMapYaml := "apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: a-cm-created-or-updated\n namespace: default\n"
resourcesCreateOrUpdateCm1, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": configMapYaml})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(resourcesCreateOrUpdateCm1.IsError, "call tool failed")
})
var decodedCreateOrUpdateCm1 []unstructured.Unstructured
err = yaml.Unmarshal([]byte(resourcesCreateOrUpdateCm1.Content[0].(mcp.TextContent).Text), &decodedCreateOrUpdateCm1)
s.Run("returns yaml content", func() {
s.Nilf(err, "invalid tool result content %v", err)
s.Truef(strings.HasPrefix(resourcesCreateOrUpdateCm1.Content[0].(mcp.TextContent).Text, "# The following resources (YAML) have been created or updated successfully"),
"Expected success message, got %v", resourcesCreateOrUpdateCm1.Content[0].(mcp.TextContent).Text)
s.Lenf(decodedCreateOrUpdateCm1, 1, "invalid resource count, expected 1, got %v", len(decodedCreateOrUpdateCm1))
s.Equalf("a-cm-created-or-updated", decodedCreateOrUpdateCm1[0].GetName(),
"invalid resource name, expected a-cm-created-or-updated, got %v", decodedCreateOrUpdateCm1[0].GetName())
s.NotEmptyf(decodedCreateOrUpdateCm1[0].GetUID(), "invalid uid, got %v", decodedCreateOrUpdateCm1[0].GetUID())
})
s.Run("creates ConfigMap", func() {
cm, _ := client.CoreV1().ConfigMaps("default").Get(s.T().Context(), "a-cm-created-or-updated", metav1.GetOptions{})
s.NotNil(cm, "ConfigMap not found")
})
})
s.Run("resources_create_or_update with valid namespaced json resource", func() {
configMapJson := "{\"apiVersion\": \"v1\", \"kind\": \"ConfigMap\", \"metadata\": {\"name\": \"a-cm-created-or-updated-2\", \"namespace\": \"default\"}}"
resourcesCreateOrUpdateCm2, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": configMapJson})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(resourcesCreateOrUpdateCm2.IsError, "call tool failed")
})
s.Run("creates config map", func() {
cm, _ := client.CoreV1().ConfigMaps("default").Get(s.T().Context(), "a-cm-created-or-updated-2", metav1.GetOptions{})
s.NotNil(cm, "ConfigMap not found")
})
})
s.Run("resources_create_or_update with valid cluster-scoped json resource", func() {
customResourceDefinitionJson := `
{
"apiVersion": "apiextensions.k8s.io/v1",
"kind": "CustomResourceDefinition",
"metadata": {"name": "customs.example.com"},
"spec": {
"group": "example.com",
"versions": [{
"name": "v1","served": true,"storage": true,
"schema": {"openAPIV3Schema": {"type": "object"}}
}],
"scope": "Namespaced",
"names": {"plural": "customs","singular": "custom","kind": "Custom"}
}
}`
resourcesCreateOrUpdateCrd, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": customResourceDefinitionJson})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(resourcesCreateOrUpdateCrd.IsError, "call tool failed")
})
s.Run("creates custom resource definition", func() {
apiExtensionsV1Client := apiextensionsv1.NewForConfigOrDie(envTestRestConfig)
_, err = apiExtensionsV1Client.CustomResourceDefinitions().Get(s.T().Context(), "customs.example.com", metav1.GetOptions{})
s.Nilf(err, "custom resource definition not found")
})
s.Require().NoError(EnvTestWaitForAPIResourceCondition(s.T().Context(), "example.com", "v1", "customs", true))
})
s.Run("resources_create_or_update creates custom resource", func() {
customJson := "{\"apiVersion\": \"example.com/v1\", \"kind\": \"Custom\", \"metadata\": {\"name\": \"a-custom-resource\"}}"
resourcesCreateOrUpdateCustom, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": customJson})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(resourcesCreateOrUpdateCustom.IsError, "call tool failed, got: %v", resourcesCreateOrUpdateCustom.Content)
})
s.Run("creates custom resource", func() {
dynamicClient := dynamic.NewForConfigOrDie(envTestRestConfig)
_, err = dynamicClient.
Resource(schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "customs"}).
Namespace("default").
Get(s.T().Context(), "a-custom-resource", metav1.GetOptions{})
s.Nilf(err, "custom resource not found")
})
})
s.Run("resources_create_or_update with valid namespaced json resource", func() {
customJsonUpdated := "{\"apiVersion\": \"example.com/v1\", \"kind\": \"Custom\", \"metadata\": {\"name\": \"a-custom-resource\",\"annotations\": {\"updated\": \"true\"}}}"
resourcesCreateOrUpdateCustomUpdated, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": customJsonUpdated})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(resourcesCreateOrUpdateCustomUpdated.IsError, "call tool failed")
})
s.Run("updates custom resource", func() {
dynamicClient := dynamic.NewForConfigOrDie(envTestRestConfig)
customResource, _ := dynamicClient.
Resource(schema.GroupVersionResource{Group: "example.com", Version: "v1", Resource: "customs"}).
Namespace("default").
Get(s.T().Context(), "a-custom-resource", metav1.GetOptions{})
s.NotNil(customResource, "custom resource not found")
annotations := customResource.GetAnnotations()
s.Require().NotNil(annotations, "annotations should not be nil")
s.Equalf("true", annotations["updated"], "custom resource not updated")
})
})
}
func (s *ResourcesSuite) TestResourcesCreateOrUpdateDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [
{ version = "v1", kind = "Secret" },
{ group = "rbac.authorization.k8s.io", version = "v1" }
]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("resources_create_or_update (denied by kind)", func() {
secretYaml := "apiVersion: v1\nkind: Secret\nmetadata:\n name: a-denied-secret\n namespace: default\n"
deniedByKind, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": secretYaml})
s.Run("has error", func() {
s.Truef(deniedByKind.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByKind.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to create or update resources:(.+:)? resource not allowed: /v1, Kind=Secret"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_create_or_update (denied by group)", func() {
roleYaml := "apiVersion: rbac.authorization.k8s.io/v1\nkind: Role\nmetadata:\n name: a-denied-role\n namespace: default\n"
deniedByGroup, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": roleYaml})
s.Run("has error", func() {
s.Truef(deniedByGroup.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByGroup.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to create or update resources:(.+:)? resource not allowed: rbac.authorization.k8s.io/v1, Kind=Role"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_create_or_update (not denied) creates or updates resource", func() {
configMapYaml := "apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: a-cm-created-or-updated\n namespace: default\n"
allowedResource, err := s.CallTool("resources_create_or_update", map[string]interface{}{"resource": configMapYaml})
s.Falsef(allowedResource.IsError, "call tool should not fail")
s.Nilf(err, "call tool should not return error object")
})
}
func (s *ResourcesSuite) TestResourcesDelete() {
s.InitMcpClient()
client := kubernetes.NewForConfigOrDie(envTestRestConfig)
s.Run("resources_delete with missing apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_delete", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to delete resource, missing argument apiVersion", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_delete with missing kind returns error", func() {
toolResult, _ := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "v1"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to delete resource, missing argument kind", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_delete with invalid apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "invalid/api/version", "kind": "Pod", "name": "a-pod"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to delete resource, invalid argument apiVersion", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_delete with nonexistent apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "custom.non.existent.example.com/v1", "kind": "Custom", "name": "a-custom"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf(`failed to delete resource: no matches for kind "Custom" in version "custom.non.existent.example.com/v1"`,
toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_delete with missing name returns error", func() {
toolResult, _ := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "v1", "kind": "Namespace"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to delete resource, missing argument name", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_delete with nonexistent resource returns error", func() {
toolResult, _ := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "v1", "kind": "ConfigMap", "name": "nonexistent-configmap"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf(`failed to delete resource: configmaps "nonexistent-configmap" not found`,
toolResult.Content[0].(mcp.TextContent).Text, "invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_delete with valid namespaced resource", func() {
resourcesDeleteCm, err := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "v1", "kind": "ConfigMap", "name": "a-configmap-to-delete"})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(resourcesDeleteCm.IsError, "call tool failed")
s.Equalf("Resource deleted successfully", resourcesDeleteCm.Content[0].(mcp.TextContent).Text,
"invalid tool result content got: %v", resourcesDeleteCm.Content[0].(mcp.TextContent).Text)
})
s.Run("deletes ConfigMap", func() {
_, err := client.CoreV1().ConfigMaps("default").Get(s.T().Context(), "a-configmap-to-delete", metav1.GetOptions{})
s.Error(err, "ConfigMap not deleted")
})
})
s.Run("resources_delete with valid cluster scoped resource", func() {
resourcesDeleteNamespace, err := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "v1", "kind": "Namespace", "name": "ns-to-delete"})
s.Run("returns success", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(resourcesDeleteNamespace.IsError, "call tool failed")
s.Equalf("Resource deleted successfully", resourcesDeleteNamespace.Content[0].(mcp.TextContent).Text,
"invalid tool result content got: %v", resourcesDeleteNamespace.Content[0].(mcp.TextContent).Text)
})
s.Run(" deletes Namespace", func() {
ns, err := client.CoreV1().Namespaces().Get(s.T().Context(), "ns-to-delete", metav1.GetOptions{})
s.Truef(err != nil || (ns != nil && ns.DeletionTimestamp != nil), "Namespace not deleted")
})
})
}
func (s *ResourcesSuite) TestResourcesDeleteDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [
{ version = "v1", kind = "Secret" },
{ group = "rbac.authorization.k8s.io", version = "v1" }
]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
_, _ = kc.CoreV1().ConfigMaps("default").Create(s.T().Context(), &corev1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{Name: "allowed-configmap-to-delete"},
}, metav1.CreateOptions{})
s.Run("resources_delete (denied by kind)", func() {
deniedByKind, err := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "v1", "kind": "Secret", "namespace": "default", "name": "denied-secret"})
s.Run("has error", func() {
s.Truef(deniedByKind.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByKind.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to delete resource:(.+:)? resource not allowed: /v1, Kind=Secret"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_delete (denied by group)", func() {
deniedByGroup, err := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "rbac.authorization.k8s.io/v1", "kind": "Role", "namespace": "default", "name": "denied-role"})
s.Run("has error", func() {
s.Truef(deniedByGroup.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByGroup.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to delete resource:(.+:)? resource not allowed: rbac.authorization.k8s.io/v1, Kind=Role"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_delete (not denied) deletes resource", func() {
allowedResource, err := s.CallTool("resources_delete", map[string]interface{}{"apiVersion": "v1", "kind": "ConfigMap", "name": "allowed-configmap-to-delete"})
s.Falsef(allowedResource.IsError, "call tool should not fail")
s.Nilf(err, "call tool should not return error object")
})
}
func (s *ResourcesSuite) TestResourcesScale() {
s.InitMcpClient()
kc := kubernetes.NewForConfigOrDie(envTestRestConfig)
deploymentName := "deployment-to-scale"
_, _ = kc.AppsV1().Deployments("default").Create(s.T().Context(), &appsv1.Deployment{
ObjectMeta: metav1.ObjectMeta{Name: deploymentName},
Spec: appsv1.DeploymentSpec{
Replicas: ptr.To(int32(2)),
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{"app": deploymentName},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": deploymentName}},
Spec: corev1.PodSpec{
Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}},
},
},
},
}, metav1.CreateOptions{})
s.Run("resources_scale with missing apiVersion returns error", func() {
toolResult, _ := s.CallTool("resources_scale", map[string]interface{}{})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get/update resource scale, missing argument apiVersion", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_scale with missing kind returns error", func() {
toolResult, _ := s.CallTool("resources_scale", map[string]interface{}{"apiVersion": "apps/v1"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get/update resource scale, missing argument kind", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_scale with missing name returns error", func() {
toolResult, _ := s.CallTool("resources_scale", map[string]interface{}{"apiVersion": "apps/v1", "kind": "Deployment"})
s.Truef(toolResult.IsError, "call tool should fail")
s.Equalf("failed to get/update resource scale, missing argument name", toolResult.Content[0].(mcp.TextContent).Text,
"invalid error message, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_scale get returns current scale", func() {
result, err := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "apps/v1",
"kind": "Deployment",
"namespace": "default",
"name": deploymentName,
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(result.IsError, "call tool failed: %v", result.Content)
})
s.Run("returns scale yaml", func() {
content := result.Content[0].(mcp.TextContent).Text
s.Truef(strings.HasPrefix(content, "# Current resource scale (YAML) is below"),
"Expected success message, got %v", content)
var decodedScale unstructured.Unstructured
err = yaml.Unmarshal([]byte(strings.TrimPrefix(content, "# Current resource scale (YAML) is below\n")), &decodedScale)
s.Nilf(err, "invalid tool result content %v", err)
replicas, found, _ := unstructured.NestedInt64(decodedScale.Object, "spec", "replicas")
s.Truef(found, "replicas not found in scale object")
s.Equalf(int64(2), replicas, "expected 2 replicas, got %d", replicas)
})
})
s.Run("resources_scale update changes the scale", func() {
result, err := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "apps/v1",
"kind": "Deployment",
"namespace": "default",
"name": deploymentName,
"scale": 5,
})
s.Run("no error", func() {
s.Nilf(err, "call tool failed %v", err)
s.Falsef(result.IsError, "call tool failed: %v", result.Content)
})
s.Run("returns updated scale yaml", func() {
content := result.Content[0].(mcp.TextContent).Text
var decodedScale unstructured.Unstructured
err = yaml.Unmarshal([]byte(strings.TrimPrefix(content, "# Current resource scale (YAML) is below\n")), &decodedScale)
s.Nilf(err, "invalid tool result content %v", err)
replicas, found, _ := unstructured.NestedInt64(decodedScale.Object, "spec", "replicas")
s.Truef(found, "replicas not found in scale object")
s.Equalf(int64(5), replicas, "expected 5 replicas after update, got %d", replicas)
})
s.Run("deployment was actually scaled", func() {
deployment, _ := kc.AppsV1().Deployments("default").Get(s.T().Context(), deploymentName, metav1.GetOptions{})
s.Equalf(int32(5), *deployment.Spec.Replicas, "expected 5 replicas in deployment, got %d", *deployment.Spec.Replicas)
})
})
s.Run("resources_scale with nonexistent resource returns error", func() {
toolResult, _ := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "apps/v1",
"kind": "Deployment",
"namespace": "default",
"name": "nonexistent-deployment",
})
s.Truef(toolResult.IsError, "call tool should fail")
s.Containsf(toolResult.Content[0].(mcp.TextContent).Text, "not found",
"expected not found error, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
s.Run("resources_scale with resource that does not support scale subresource returns error", func() {
configMapName := "configmap-without-scale"
_, _ = kc.CoreV1().ConfigMaps("default").Create(s.T().Context(), &corev1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{Name: configMapName},
Data: map[string]string{"key": "value"},
}, metav1.CreateOptions{})
toolResult, _ := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "v1",
"kind": "ConfigMap",
"namespace": "default",
"name": configMapName,
})
s.Truef(toolResult.IsError, "call tool should fail")
s.Containsf(toolResult.Content[0].(mcp.TextContent).Text, "the server could not find the requested resource",
"expected scale subresource not found error, got %v", toolResult.Content[0].(mcp.TextContent).Text)
})
}
func (s *ResourcesSuite) TestResourcesScaleDenied() {
s.Require().NoError(toml.Unmarshal([]byte(`
denied_resources = [
{ group = "apps", version = "v1" },
{ group = "", version = "v1", kind = "ReplicationController" }
]
`), s.Cfg), "Expected to parse denied resources config")
s.InitMcpClient()
s.Run("resources_scale get (denied by kind)", func() {
deniedByKind, err := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "v1",
"kind": "ReplicationController",
"namespace": "default",
"name": "nonexistent-rc",
})
s.Run("has error", func() {
s.Truef(deniedByKind.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByKind.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get/update resource scale:(.+:)? resource not allowed: /v1, Kind=ReplicationController"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_scale update (denied by kind)", func() {
deniedByKind, err := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "v1",
"kind": "ReplicationController",
"namespace": "default",
"name": "nonexistent-rc",
"scale": 1337,
})
s.Run("has error", func() {
s.Truef(deniedByKind.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByKind.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get/update resource scale:(.+:)? resource not allowed: /v1, Kind=ReplicationController"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_scale get (denied by group)", func() {
deniedByGroup, err := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "apps/v1",
"kind": "StatefulSet",
"namespace": "default",
"name": "nonexistent-statefulset",
})
s.Run("has error", func() {
s.Truef(deniedByGroup.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByGroup.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get/update resource scale:(.+:)? resource not allowed: apps/v1, Kind=StatefulSet"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
s.Run("resources_scale update (denied by group)", func() {
deniedByGroup, err := s.CallTool("resources_scale", map[string]interface{}{
"apiVersion": "apps/v1",
"kind": "StatefulSet",
"namespace": "default",
"name": "nonexistent-statefulset",
"scale": 1337,
})
s.Run("has error", func() {
s.Truef(deniedByGroup.IsError, "call tool should fail")
s.Nilf(err, "call tool should not return error object")
})
s.Run("describes denial", func() {
msg := deniedByGroup.Content[0].(mcp.TextContent).Text
s.Contains(msg, "resource not allowed:")
expectedMessage := "failed to get/update resource scale:(.+:)? resource not allowed: apps/v1, Kind=StatefulSet"
s.Regexpf(expectedMessage, msg,
"expected descriptive error '%s', got %v", expectedMessage, msg)
})
})
}
func TestResources(t *testing.T) {
suite.Run(t, new(ResourcesSuite))
}
```