Model Control Plane (MCP) Server

import pytest import json import threading import time import requests from unittest.mock import patch, MagicMock import sys import os from http.server import HTTPServer # Add scripts directory to path to import the metrics generator sys.path.append(os.path.join(os.path.dirname(__file__), "..", "scripts")) from k8s_dummy_data_generator import ( MetricsGenerator, MetricsServer, start_server, DEFAULT_POD_COUNT, HTTP_PORT ) class TestK8sMetricsGenerator: """Tests for the Kubernetes metrics generator""" @pytest.fixture def metrics_generator(self): """Create a metrics generator instance for testing""" return MetricsGenerator(pod_count=5) # Use a small number for testing def test_initialization(self, metrics_generator): """Test that metrics generator initializes correctly""" # Check that pods were created assert len(metrics_generator.pods) == 5 # Check each pod has the required attributes for pod in metrics_generator.pods: assert "name" in pod assert "container_name" in pod assert "image" in pod assert "node" in pod assert "namespace" in pod assert "uid" in pod def test_generate_metrics(self, metrics_generator): """Test metrics generation""" metrics = metrics_generator.generate_metrics() # Verify that metrics is a string assert isinstance(metrics, str) # Check that metrics are not empty assert len(metrics) > 0 # Check that metrics contain expected labels and values metrics_lines = metrics.split("\n") assert len(metrics_lines) > 0 # Check for the presence of key metrics cpu_metrics = [line for line in metrics_lines if "container_cpu_usage_seconds_total" in line] memory_metrics = [line for line in metrics_lines if "container_memory_usage_bytes" in line] network_metrics = [line for line in metrics_lines if "container_network" in line] assert len(cpu_metrics) > 0 assert len(memory_metrics) > 0 assert len(network_metrics) > 0 # Check format of a sample metric sample_metric = cpu_metrics[0] assert "{" in sample_metric and "}" in sample_metric # Check for labels assert "container_name=" in sample_metric # Check for specific label assert "kubernetes_io_hostname=" in sample_metric # Check for hostname label # Verify numeric value at the end metric_parts = sample_metric.split("}") assert len(metric_parts) == 2 assert metric_parts[1].strip().replace(".", "", 1).isdigit() # Check if the value is numeric @patch("http.server.HTTPServer") def test_metrics_server(self, mock_http_server): """Test the metrics server initialization""" # Create mock objects for the server mock_server = MagicMock() mock_http_server.return_value = mock_server # Create a metrics generator metrics_generator = MetricsGenerator(pod_count=3) # Create a mock handler function to use instead of the real one def mock_handler_factory(metrics_generator): return lambda *args, **kwargs: None # Create a simple HTTPServer manually to test if it would work server = HTTPServer(('0.0.0.0', HTTP_PORT), MetricsServer) server.metrics_generator = metrics_generator # Verify the server can be created without errors assert hasattr(server, "metrics_generator") # Cleanup - we don't actually start this server del server def test_metrics_server_handler(self): """Test the MetricsServer handler""" # Create mock objects mock_server = MagicMock() mock_generator = MagicMock() mock_generator.generate_metrics.return_value = "test_metric{label='value'} 1.0" mock_server.metrics_generator = mock_generator # Create mock request handler with minimal required attributes mock_wfile = MagicMock() class MockHandler(MetricsServer): def __init__(self): self.server = mock_server self.path = "/metrics" self.wfile = mock_wfile def send_response(self, code): self.response_code = code def send_header(self, name, value): pass def end_headers(self): pass # Create and test the handler handler = MockHandler() handler.do_GET() # Verify that the generator was called mock_generator.generate_metrics.assert_called_once() # Verify that the response code was 200 assert handler.response_code == 200 # Verify that the metrics were written to the response mock_wfile.write.assert_called_once() args, kwargs = mock_wfile.write.call_args assert b"test_metric" in args[0] def test_metrics_server_handler_404(self): """Test the MetricsServer handler for non-metrics paths""" # Create mock objects mock_server = MagicMock() mock_generator = MagicMock() mock_server.metrics_generator = mock_generator # Create mock request handler with minimal required attributes mock_wfile = MagicMock() class MockHandler(MetricsServer): def __init__(self): self.server = mock_server self.path = "/nonexistent" self.wfile = mock_wfile def send_response(self, code): self.response_code = code def send_header(self, name, value): pass def end_headers(self): pass # Create and test the handler handler = MockHandler() handler.do_GET() # Verify that the generator was not called mock_generator.generate_metrics.assert_not_called() # Verify that the response code was 404 assert handler.response_code == 404 # Verify that "Not Found" was written to the response mock_wfile.write.assert_called_once_with(b'Not Found') @pytest.mark.integration def test_metrics_endpoint_integration(self): """ Integration test for the metrics endpoint. This test starts an actual HTTP server and makes a request to it. Run with --run-integration flag to execute this test. """ # Create a metrics generator generator = MetricsGenerator(pod_count=2) port = 9999 # Use a different port for testing # Start the server in a background thread server_thread = threading.Thread( target=start_server, args=(port, generator), daemon=True ) server_thread.start() # Give the server time to start time.sleep(1) try: # Make a request to the metrics endpoint response = requests.get(f"http://localhost:{port}/metrics") # Verify the response assert response.status_code == 200 assert len(response.text) > 0 assert "container_cpu_usage_seconds_total" in response.text # Make a request to a non-existent endpoint response_404 = requests.get(f"http://localhost:{port}/nonexistent") assert response_404.status_code == 404 assert response_404.text == "Not Found" except Exception as e: pytest.fail(f"Failed to connect to metrics server: {e}") finally: # We can't easily stop the server (it's in serve_forever), # but it's a daemon thread so it will be terminated when the test exits pass