"""Tests for the Gradle MCP dashboard."""
import pytest
import socket
from pathlib import Path
from datetime import datetime
from gradle_mcp.dashboard import DaemonMonitor, DaemonInfo, DaemonSession
from gradle_mcp.dashboard.app import find_available_port, create_app
class TestDaemonMonitor:
"""Test suite for DaemonMonitor."""
def test_add_and_get_logs(self):
"""Test adding and retrieving logs."""
monitor = DaemonMonitor(Path("."))
monitor.add_log("12345", "Test message", "INFO")
logs = monitor.get_logs("12345")
assert len(logs) == 1
assert logs[0]["message"] == "Test message"
assert logs[0]["level"] == "INFO"
def test_get_logs_with_limit(self):
"""Test log retrieval respects limit."""
monitor = DaemonMonitor(Path("."))
for i in range(10):
monitor.add_log("12345", f"Message {i}", "INFO")
logs = monitor.get_logs("12345", limit=5)
assert len(logs) == 5
def test_get_logs_returns_most_recent(self):
"""Test that logs returns the most recent entries when limited."""
monitor = DaemonMonitor(Path("."))
for i in range(10):
monitor.add_log("12345", f"Message {i}", "INFO")
logs = monitor.get_logs("12345", limit=5)
# Should return the last 5 messages (5-9)
assert logs[0]["message"] == "Message 5"
assert logs[-1]["message"] == "Message 9"
def test_get_logs_nonexistent_daemon(self):
"""Test getting logs for a daemon that doesn't exist."""
monitor = DaemonMonitor(Path("."))
logs = monitor.get_logs("nonexistent")
assert logs == []
def test_start_and_end_session(self):
"""Test session lifecycle."""
monitor = DaemonMonitor(Path("."))
session_id = monitor.start_session("12345", "build")
sessions = monitor.get_active_sessions()
assert len(sessions) == 1
assert sessions[0].task == "build"
assert sessions[0].daemon_pid == "12345"
monitor.end_session(session_id)
sessions = monitor.get_active_sessions()
assert len(sessions) == 0
def test_get_active_sessions(self):
"""Test getting active sessions."""
monitor = DaemonMonitor(Path("."))
monitor.start_session("11111", "build")
monitor.start_session("22222", "test")
sessions = monitor.get_active_sessions()
assert len(sessions) == 2
def test_session_logs_added_on_start_and_end(self):
"""Test that logs are added when sessions start and end."""
monitor = DaemonMonitor(Path("."))
session_id = monitor.start_session("12345", "build")
# Should have a log entry for session start
logs = monitor.get_logs("12345")
assert any("Started session" in log["message"] for log in logs)
monitor.end_session(session_id)
# Should have a log entry for session end
logs = monitor.get_logs("12345")
assert any("Ended session" in log["message"] for log in logs)
def test_end_nonexistent_session(self):
"""Test ending a session that doesn't exist doesn't raise an error."""
monitor = DaemonMonitor(Path("."))
# Should not raise
monitor.end_session("nonexistent-session-id")
class TestDaemonInfo:
"""Test suite for DaemonInfo dataclass."""
def test_to_dict(self):
"""Test DaemonInfo to_dict conversion."""
now = datetime.now()
info = DaemonInfo(
pid="12345",
status="IDLE",
gradle_version="8.5",
last_activity=now,
)
result = info.to_dict()
assert result["pid"] == "12345"
assert result["status"] == "IDLE"
assert result["gradle_version"] == "8.5"
assert result["last_activity"] == now.isoformat()
def test_to_dict_with_none_values(self):
"""Test DaemonInfo to_dict with None values."""
info = DaemonInfo(
pid="12345",
status="IDLE",
)
result = info.to_dict()
assert result["pid"] == "12345"
assert result["status"] == "IDLE"
assert result["gradle_version"] is None
assert result["last_activity"] is None
class TestDaemonSession:
"""Test suite for DaemonSession dataclass."""
def test_to_dict(self):
"""Test DaemonSession to_dict conversion."""
now = datetime.now()
session = DaemonSession(
session_id="abc123",
daemon_pid="12345",
task="build",
start_time=now,
)
result = session.to_dict()
assert result["session_id"] == "abc123"
assert result["daemon_pid"] == "12345"
assert result["task"] == "build"
assert result["start_time"] == now.isoformat()
assert result["log_count"] == 0
class TestFindAvailablePort:
"""Test suite for find_available_port function."""
def test_finds_port(self):
"""Test finding an available port."""
port = find_available_port(start_port=3333)
assert port >= 3333
def test_returns_different_port_when_occupied(self):
"""Test returns next port when start port is occupied."""
# Occupy a port
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("127.0.0.1", 13333))
try:
port = find_available_port(start_port=13333)
assert port > 13333
finally:
sock.close()
def test_found_port_is_actually_available(self):
"""Test that the found port can actually be bound."""
port = find_available_port(start_port=14000)
# Verify we can bind to it
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.bind(("127.0.0.1", port))
finally:
sock.close()
class TestFlaskApp:
"""Test suite for Flask application."""
def test_create_app(self):
"""Test Flask app creation."""
monitor = DaemonMonitor(Path("."))
app, socketio = create_app(monitor)
assert app is not None
assert socketio is not None
def test_app_has_daemon_monitor_in_config(self):
"""Test that daemon monitor is stored in app config."""
monitor = DaemonMonitor(Path("."))
app, _ = create_app(monitor)
assert app.config["daemon_monitor"] is monitor
def test_api_daemons_endpoint(self):
"""Test the /api/daemons endpoint."""
monitor = DaemonMonitor(Path("."))
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/api/daemons")
assert response.status_code == 200
data = response.get_json()
# Should return a list (may be empty if no daemons running)
assert isinstance(data, list)
def test_api_sessions_endpoint(self):
"""Test the /api/sessions endpoint."""
monitor = DaemonMonitor(Path("."))
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/api/sessions")
assert response.status_code == 200
data = response.get_json()
# Should return a list
assert isinstance(data, list)
def test_api_sessions_returns_active_sessions(self):
"""Test that /api/sessions returns active sessions."""
monitor = DaemonMonitor(Path("."))
monitor.start_session("12345", "build")
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/api/sessions")
assert response.status_code == 200
data = response.get_json()
assert len(data) == 1
assert data[0]["task"] == "build"
assert data[0]["daemon_pid"] == "12345"
def test_api_logs_endpoint(self):
"""Test the /api/logs/<daemon_pid> endpoint."""
monitor = DaemonMonitor(Path("."))
monitor.add_log("12345", "Test log message", "INFO")
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/api/logs/12345")
assert response.status_code == 200
data = response.get_json()
assert len(data) == 1
assert data[0]["message"] == "Test log message"
def test_api_logs_with_limit_parameter(self):
"""Test that /api/logs respects the limit parameter."""
monitor = DaemonMonitor(Path("."))
for i in range(10):
monitor.add_log("12345", f"Log {i}", "INFO")
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/api/logs/12345?limit=3")
assert response.status_code == 200
data = response.get_json()
assert len(data) == 3
def test_index_endpoint(self):
"""Test the index page endpoint."""
monitor = DaemonMonitor(Path("."))
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/")
assert response.status_code == 200
# Should return HTML content
assert b"<!DOCTYPE html>" in response.data or b"<html" in response.data
def test_log_viewer_endpoint(self):
"""Test the log viewer page endpoint."""
monitor = DaemonMonitor(Path("."))
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/logs/12345")
assert response.status_code == 200
# Should return HTML content
assert b"<!DOCTYPE html>" in response.data or b"<html" in response.data
def test_daemon_details_page(self):
"""Test the daemon details page renders."""
monitor = DaemonMonitor(Path("."))
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/daemon/12345")
assert response.status_code == 200
# Should return HTML content
assert b"<!DOCTYPE html>" in response.data or b"<html" in response.data
# Check for the Gradle MCP Dashboard title/branding
assert b"Gradle MCP Dashboard" in response.data
# Verify the daemon PID is set in the JavaScript variable
assert b'const DAEMON_PID = "12345"' in response.data
def test_api_daemon_details_not_found(self):
"""Test 404 when daemon doesn't exist."""
monitor = DaemonMonitor(Path("."))
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/api/daemons/99999/details")
assert response.status_code == 404
data = response.get_json()
assert "error" in data
def test_api_daemon_details_success(self):
"""Test getting daemon details for an existing daemon."""
from unittest.mock import patch, MagicMock
from datetime import datetime
monitor = DaemonMonitor(Path("."))
# Create a mock daemon
mock_daemon = DaemonInfo(
pid="12345",
status="IDLE",
gradle_version="8.5",
last_activity=datetime.now(),
)
# Mock the get_daemons method to return our mock daemon
with patch.object(monitor, 'get_daemons', return_value=[mock_daemon]):
# Also mock the GradleWrapper.get_daemon_specific_config method
mock_config = {
"runtime_jvm_args": ["-Xmx2g", "-XX:MaxMetaspaceSize=512m"],
"jvm_args": "-Xmx2g",
"daemon_enabled": True,
"parallel_enabled": False,
"caching_enabled": True,
"max_workers": 4,
"distribution_url": "https://services.gradle.org/distributions/gradle-8.5-bin.zip",
"gradle_version": "8.5",
}
with patch('gradle_mcp.gradle.GradleWrapper') as MockWrapper:
mock_wrapper_instance = MagicMock()
mock_wrapper_instance.get_daemon_specific_config.return_value = mock_config
MockWrapper.return_value = mock_wrapper_instance
app, _ = create_app(monitor)
with app.test_client() as client:
response = client.get("/api/daemons/12345/details")
assert response.status_code == 200
data = response.get_json()
# Verify response structure
assert "daemon" in data
assert "config" in data
assert "active_builds" in data
# Verify daemon info
assert data["daemon"]["pid"] == "12345"
assert data["daemon"]["status"] == "IDLE"
assert data["daemon"]["gradle_version"] == "8.5"
# Verify config
assert "runtime_jvm_args" in data["config"]
