forked from Shifty/pyserveX
1349 lines
47 KiB
Python
1349 lines
47 KiB
Python
"""
|
|
Comprehensive tests for pyservectl CLI tool.
|
|
|
|
Tests cover:
|
|
- CLI main entry point and commands
|
|
- State management
|
|
- Output formatting utilities
|
|
- Service runner
|
|
- Config management commands
|
|
- Init command templates
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
import yaml
|
|
from click.testing import CliRunner
|
|
|
|
from pyserve.ctl.main import cli, Context, DEFAULT_CONFIG, DEFAULT_STATE_DIR
|
|
from pyserve.ctl.state import (
|
|
ProjectState,
|
|
ServiceHealth,
|
|
ServiceState,
|
|
StateManager,
|
|
)
|
|
from pyserve.ctl.output import (
|
|
console,
|
|
create_services_table,
|
|
format_bytes,
|
|
format_health,
|
|
format_status,
|
|
format_uptime,
|
|
print_error,
|
|
print_info,
|
|
print_success,
|
|
print_warning,
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Fixtures
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def runner():
|
|
"""Click CLI test runner."""
|
|
return CliRunner()
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Temporary directory for tests."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
|
|
@pytest.fixture
|
|
def state_dir(temp_dir):
|
|
"""Temporary state directory."""
|
|
state_path = temp_dir / ".pyserve"
|
|
state_path.mkdir()
|
|
return state_path
|
|
|
|
|
|
@pytest.fixture
|
|
def state_manager(state_dir):
|
|
"""StateManager instance with temp directory."""
|
|
return StateManager(state_dir, project="test-project")
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_config(temp_dir):
|
|
"""Create a sample config.yaml file."""
|
|
config_content = """
|
|
server:
|
|
host: 0.0.0.0
|
|
port: 8080
|
|
backlog: 100
|
|
proxy_timeout: 30.0
|
|
|
|
logging:
|
|
level: INFO
|
|
console_output: true
|
|
|
|
extensions:
|
|
- type: process_orchestration
|
|
config:
|
|
port_range: [9000, 9999]
|
|
apps:
|
|
- name: api
|
|
path: /api
|
|
app_path: myapp.api:app
|
|
workers: 2
|
|
health_check_path: /health
|
|
- name: admin
|
|
path: /admin
|
|
app_path: myapp.admin:app
|
|
workers: 1
|
|
- type: routing
|
|
config:
|
|
regex_locations:
|
|
"=/health":
|
|
return: "200 OK"
|
|
"""
|
|
config_path = temp_dir / "config.yaml"
|
|
config_path.write_text(config_content)
|
|
return config_path
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_runner(temp_dir, sample_config):
|
|
"""CLI runner with isolated environment."""
|
|
runner = CliRunner()
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
# Copy config to current directory
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
yield runner
|
|
|
|
|
|
# =============================================================================
|
|
# Context Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestContext:
|
|
"""Tests for Context class."""
|
|
|
|
def test_context_defaults(self):
|
|
"""Test Context default values."""
|
|
ctx = Context()
|
|
assert ctx.config_file == DEFAULT_CONFIG
|
|
assert ctx.state_dir == Path(DEFAULT_STATE_DIR)
|
|
assert ctx.verbose is False
|
|
assert ctx.debug is False
|
|
assert ctx.project is None
|
|
|
|
def test_context_config_lazy_load(self, temp_dir, sample_config):
|
|
"""Test lazy loading of config."""
|
|
ctx = Context()
|
|
ctx.config_file = str(sample_config)
|
|
|
|
# First access should load config
|
|
config = ctx.config
|
|
assert config is not None
|
|
assert config.server.port == 8080
|
|
|
|
# Second access should return cached config
|
|
assert ctx.config is config
|
|
|
|
def test_context_config_missing_file(self, temp_dir):
|
|
"""Test config loading with missing file."""
|
|
ctx = Context()
|
|
ctx.config_file = str(temp_dir / "nonexistent.yaml")
|
|
|
|
# Should return default config
|
|
config = ctx.config
|
|
assert config is not None
|
|
|
|
def test_context_state_lazy_load(self, state_dir):
|
|
"""Test lazy loading of state."""
|
|
ctx = Context()
|
|
ctx.state_dir = state_dir
|
|
ctx.project = "test"
|
|
|
|
state = ctx.state
|
|
assert state is not None
|
|
assert isinstance(state, StateManager)
|
|
|
|
# Second access should return cached state
|
|
assert ctx.state is state
|
|
|
|
|
|
# =============================================================================
|
|
# CLI Main Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestCLIMain:
|
|
"""Tests for main CLI entry point."""
|
|
|
|
def test_cli_no_command_shows_help(self, runner):
|
|
"""Test CLI without command shows help."""
|
|
result = runner.invoke(cli)
|
|
assert result.exit_code == 0
|
|
assert "PyServeCTL" in result.output
|
|
assert "Quick Start:" in result.output
|
|
|
|
def test_cli_help(self, runner):
|
|
"""Test --help option."""
|
|
result = runner.invoke(cli, ["--help"])
|
|
assert result.exit_code == 0
|
|
assert "PyServeCTL" in result.output
|
|
assert "Commands:" in result.output
|
|
|
|
def test_cli_version(self, runner):
|
|
"""Test --version option."""
|
|
result = runner.invoke(cli, ["--version"])
|
|
assert result.exit_code == 0
|
|
assert "pyservectl" in result.output
|
|
|
|
def test_cli_verbose_flag(self, runner):
|
|
"""Test --verbose flag is passed to context."""
|
|
result = runner.invoke(cli, ["--verbose", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_cli_debug_flag(self, runner):
|
|
"""Test --debug flag is passed to context."""
|
|
result = runner.invoke(cli, ["--debug", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_cli_config_option(self, runner, temp_dir, sample_config):
|
|
"""Test --config option."""
|
|
result = runner.invoke(cli, ["-c", str(sample_config), "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_cli_project_option(self, runner):
|
|
"""Test --project option."""
|
|
result = runner.invoke(cli, ["-p", "myproject", "--help"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
# =============================================================================
|
|
# State Management Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestServiceHealth:
|
|
"""Tests for ServiceHealth dataclass."""
|
|
|
|
def test_default_values(self):
|
|
"""Test default values."""
|
|
health = ServiceHealth()
|
|
assert health.status == "unknown"
|
|
assert health.last_check is None
|
|
assert health.failures == 0
|
|
assert health.response_time_ms is None
|
|
|
|
def test_custom_values(self):
|
|
"""Test custom values."""
|
|
health = ServiceHealth(
|
|
status="healthy",
|
|
last_check=1234567890.0,
|
|
failures=2,
|
|
response_time_ms=50.5,
|
|
)
|
|
assert health.status == "healthy"
|
|
assert health.last_check == 1234567890.0
|
|
assert health.failures == 2
|
|
assert health.response_time_ms == 50.5
|
|
|
|
|
|
class TestServiceState:
|
|
"""Tests for ServiceState dataclass."""
|
|
|
|
def test_default_values(self):
|
|
"""Test default values."""
|
|
state = ServiceState(name="test")
|
|
assert state.name == "test"
|
|
assert state.state == "stopped"
|
|
assert state.pid is None
|
|
assert state.port == 0
|
|
assert state.workers == 0
|
|
assert state.started_at is None
|
|
assert state.restart_count == 0
|
|
assert isinstance(state.health, ServiceHealth)
|
|
|
|
def test_uptime_not_started(self):
|
|
"""Test uptime when not started."""
|
|
state = ServiceState(name="test")
|
|
assert state.uptime == 0.0
|
|
|
|
def test_uptime_running(self):
|
|
"""Test uptime calculation."""
|
|
state = ServiceState(name="test", started_at=time.time() - 100)
|
|
assert 99 <= state.uptime <= 101
|
|
|
|
def test_to_dict(self):
|
|
"""Test serialization to dict."""
|
|
state = ServiceState(
|
|
name="api",
|
|
state="running",
|
|
pid=1234,
|
|
port=9000,
|
|
workers=2,
|
|
)
|
|
data = state.to_dict()
|
|
|
|
assert data["name"] == "api"
|
|
assert data["state"] == "running"
|
|
assert data["pid"] == 1234
|
|
assert data["port"] == 9000
|
|
assert data["workers"] == 2
|
|
assert "health" in data
|
|
|
|
def test_from_dict(self):
|
|
"""Test deserialization from dict."""
|
|
data = {
|
|
"name": "api",
|
|
"state": "running",
|
|
"pid": 1234,
|
|
"port": 9000,
|
|
"workers": 2,
|
|
"started_at": 1234567890.0,
|
|
"restart_count": 1,
|
|
"health": {"status": "healthy", "failures": 0},
|
|
"config_hash": "abc123",
|
|
}
|
|
state = ServiceState.from_dict(data)
|
|
|
|
assert state.name == "api"
|
|
assert state.state == "running"
|
|
assert state.pid == 1234
|
|
assert state.health.status == "healthy"
|
|
|
|
|
|
class TestProjectState:
|
|
"""Tests for ProjectState dataclass."""
|
|
|
|
def test_default_values(self):
|
|
"""Test default values."""
|
|
state = ProjectState()
|
|
assert state.version == "1.0"
|
|
assert state.project == ""
|
|
assert state.services == {}
|
|
|
|
def test_to_dict(self):
|
|
"""Test serialization."""
|
|
state = ProjectState(
|
|
project="myproject",
|
|
config_file="config.yaml",
|
|
)
|
|
state.services["api"] = ServiceState(name="api", state="running")
|
|
|
|
data = state.to_dict()
|
|
assert data["project"] == "myproject"
|
|
assert "api" in data["services"]
|
|
|
|
def test_from_dict(self):
|
|
"""Test deserialization."""
|
|
data = {
|
|
"version": "1.0",
|
|
"project": "myproject",
|
|
"config_file": "config.yaml",
|
|
"config_hash": "",
|
|
"started_at": None,
|
|
"daemon_pid": None,
|
|
"services": {
|
|
"api": {
|
|
"name": "api",
|
|
"state": "running",
|
|
"pid": 1234,
|
|
"port": 9000,
|
|
"workers": 1,
|
|
"started_at": None,
|
|
"restart_count": 0,
|
|
"health": {},
|
|
"config_hash": "",
|
|
}
|
|
},
|
|
}
|
|
state = ProjectState.from_dict(data)
|
|
assert state.project == "myproject"
|
|
assert "api" in state.services
|
|
|
|
|
|
class TestStateManager:
|
|
"""Tests for StateManager class."""
|
|
|
|
def test_init(self, state_dir):
|
|
"""Test initialization."""
|
|
manager = StateManager(state_dir, project="test")
|
|
assert manager.state_dir == state_dir
|
|
assert manager.project == "test"
|
|
|
|
def test_paths(self, state_manager, state_dir):
|
|
"""Test path properties."""
|
|
assert state_manager.state_file == state_dir / "state.json"
|
|
assert state_manager.pid_file == state_dir / "pyserve.pid"
|
|
assert state_manager.socket_file == state_dir / "pyserve.sock"
|
|
assert state_manager.logs_dir == state_dir / "logs"
|
|
|
|
def test_ensure_dirs(self, state_manager):
|
|
"""Test directory creation."""
|
|
state_manager.ensure_dirs()
|
|
assert state_manager.state_dir.exists()
|
|
assert state_manager.logs_dir.exists()
|
|
|
|
def test_load_empty(self, state_manager):
|
|
"""Test loading when no state file exists."""
|
|
state = state_manager.load()
|
|
assert isinstance(state, ProjectState)
|
|
assert state.project == "test-project"
|
|
|
|
def test_save_and_load(self, state_manager):
|
|
"""Test save and load cycle."""
|
|
state_manager.update_service("api", state="running", pid=1234, port=9000)
|
|
|
|
# Create new manager to force reload
|
|
manager2 = StateManager(state_manager.state_dir, project="test-project")
|
|
manager2._state = None # Force reload
|
|
state = manager2.load()
|
|
|
|
assert "api" in state.services
|
|
assert state.services["api"].pid == 1234
|
|
|
|
def test_update_service_new(self, state_manager):
|
|
"""Test updating a new service."""
|
|
service = state_manager.update_service("api", state="running", port=9000)
|
|
|
|
assert service.name == "api"
|
|
assert service.state == "running"
|
|
assert service.port == 9000
|
|
|
|
def test_update_service_existing(self, state_manager):
|
|
"""Test updating an existing service."""
|
|
state_manager.update_service("api", state="running", port=9000)
|
|
service = state_manager.update_service("api", state="stopped", pid=None)
|
|
|
|
assert service.state == "stopped"
|
|
assert service.port == 9000 # Should preserve
|
|
|
|
def test_remove_service(self, state_manager):
|
|
"""Test removing a service."""
|
|
state_manager.update_service("api", state="running")
|
|
state_manager.remove_service("api")
|
|
|
|
assert state_manager.get_service("api") is None
|
|
|
|
def test_get_service(self, state_manager):
|
|
"""Test getting a service."""
|
|
state_manager.update_service("api", state="running")
|
|
|
|
service = state_manager.get_service("api")
|
|
assert service is not None
|
|
assert service.name == "api"
|
|
|
|
assert state_manager.get_service("nonexistent") is None
|
|
|
|
def test_get_all_services(self, state_manager):
|
|
"""Test getting all services."""
|
|
state_manager.update_service("api", state="running")
|
|
state_manager.update_service("admin", state="stopped")
|
|
|
|
services = state_manager.get_all_services()
|
|
assert len(services) == 2
|
|
assert "api" in services
|
|
assert "admin" in services
|
|
|
|
def test_clear(self, state_manager):
|
|
"""Test clearing state."""
|
|
state_manager.update_service("api", state="running")
|
|
state_manager.clear()
|
|
|
|
assert len(state_manager.get_all_services()) == 0
|
|
|
|
def test_daemon_pid_not_running(self, state_manager):
|
|
"""Test daemon PID when not running."""
|
|
assert state_manager.is_daemon_running() is False
|
|
assert state_manager.get_daemon_pid() is None
|
|
|
|
def test_daemon_pid_operations(self, state_manager):
|
|
"""Test daemon PID set/get/clear."""
|
|
# Set PID (but process won't actually exist)
|
|
state_manager.set_daemon_pid(99999)
|
|
|
|
# Should return None since process doesn't exist
|
|
assert state_manager.is_daemon_running() is False
|
|
|
|
# Clear
|
|
state_manager.clear_daemon_pid()
|
|
assert not state_manager.pid_file.exists()
|
|
|
|
def test_get_service_log_file(self, state_manager):
|
|
"""Test getting service log file path."""
|
|
log_file = state_manager.get_service_log_file("api")
|
|
assert log_file == state_manager.logs_dir / "api.log"
|
|
|
|
def test_compute_config_hash(self, state_manager, temp_dir):
|
|
"""Test config hash computation."""
|
|
config_path = temp_dir / "test.yaml"
|
|
config_path.write_text("test: value")
|
|
|
|
hash1 = state_manager.compute_config_hash(str(config_path))
|
|
assert len(hash1) == 16
|
|
|
|
# Same content should give same hash
|
|
hash2 = state_manager.compute_config_hash(str(config_path))
|
|
assert hash1 == hash2
|
|
|
|
# Different content should give different hash
|
|
config_path.write_text("test: different")
|
|
hash3 = state_manager.compute_config_hash(str(config_path))
|
|
assert hash1 != hash3
|
|
|
|
def test_compute_config_hash_missing_file(self, state_manager):
|
|
"""Test config hash with missing file."""
|
|
hash_val = state_manager.compute_config_hash("/nonexistent/file.yaml")
|
|
assert hash_val == ""
|
|
|
|
|
|
# =============================================================================
|
|
# Output Formatting Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestOutputFormatting:
|
|
"""Tests for output formatting utilities."""
|
|
|
|
def test_format_status(self):
|
|
"""Test status formatting."""
|
|
assert "running" in format_status("running")
|
|
assert "stopped" in format_status("stopped")
|
|
assert "failed" in format_status("failed")
|
|
assert "starting" in format_status("starting")
|
|
assert "stopping" in format_status("stopping")
|
|
assert "restarting" in format_status("restarting")
|
|
assert "pending" in format_status("pending")
|
|
assert format_status("unknown") == "unknown"
|
|
|
|
def test_format_health(self):
|
|
"""Test health formatting."""
|
|
assert "healthy" in format_health("healthy")
|
|
assert "unhealthy" in format_health("unhealthy")
|
|
assert "degraded" in format_health("degraded")
|
|
assert "unknown" in format_health("unknown")
|
|
assert format_health("-") is not None
|
|
|
|
def test_format_uptime_zero(self):
|
|
"""Test uptime formatting for zero."""
|
|
assert format_uptime(0) == "-"
|
|
assert format_uptime(-10) == "-"
|
|
|
|
def test_format_uptime_seconds(self):
|
|
"""Test uptime formatting for seconds."""
|
|
assert format_uptime(30) == "30s"
|
|
assert format_uptime(59) == "59s"
|
|
|
|
def test_format_uptime_minutes(self):
|
|
"""Test uptime formatting for minutes."""
|
|
assert format_uptime(60) == "1m 0s"
|
|
assert format_uptime(90) == "1m 30s"
|
|
assert format_uptime(3599) == "59m 59s"
|
|
|
|
def test_format_uptime_hours(self):
|
|
"""Test uptime formatting for hours."""
|
|
assert format_uptime(3600) == "1h 0m"
|
|
assert format_uptime(7200) == "2h 0m"
|
|
assert format_uptime(5400) == "1h 30m"
|
|
|
|
def test_format_uptime_days(self):
|
|
"""Test uptime formatting for days."""
|
|
assert format_uptime(86400) == "1d 0h"
|
|
assert format_uptime(172800) == "2d 0h"
|
|
assert format_uptime(90000) == "1d 1h"
|
|
|
|
def test_format_bytes(self):
|
|
"""Test bytes formatting."""
|
|
assert format_bytes(500) == "500.0B"
|
|
assert format_bytes(1024) == "1.0KB"
|
|
assert format_bytes(1536) == "1.5KB"
|
|
assert format_bytes(1048576) == "1.0MB"
|
|
assert format_bytes(1073741824) == "1.0GB"
|
|
|
|
def test_create_services_table(self):
|
|
"""Test services table creation."""
|
|
table = create_services_table()
|
|
assert table is not None
|
|
# Check columns exist
|
|
column_names = [col.header for col in table.columns]
|
|
assert "NAME" in column_names
|
|
assert "STATUS" in column_names
|
|
assert "PORTS" in column_names
|
|
|
|
|
|
# =============================================================================
|
|
# Init Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestInitCommand:
|
|
"""Tests for init command."""
|
|
|
|
def test_init_basic_template(self, runner, temp_dir):
|
|
"""Test init with basic template."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["init"])
|
|
|
|
assert result.exit_code == 0
|
|
assert Path("config.yaml").exists()
|
|
assert Path("static").exists()
|
|
assert Path("templates").exists()
|
|
assert Path("logs").exists()
|
|
assert Path(".pyserve").exists()
|
|
|
|
def test_init_orchestration_template(self, runner, temp_dir):
|
|
"""Test init with orchestration template."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["init", "-t", "orchestration"])
|
|
|
|
assert result.exit_code == 0
|
|
config = Path("config.yaml").read_text()
|
|
assert "process_orchestration" in config
|
|
|
|
def test_init_asgi_template(self, runner, temp_dir):
|
|
"""Test init with asgi template."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["init", "-t", "asgi"])
|
|
|
|
assert result.exit_code == 0
|
|
config = Path("config.yaml").read_text()
|
|
assert "asgi_mount" in config
|
|
|
|
def test_init_full_template(self, runner, temp_dir):
|
|
"""Test init with full template."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["init", "-t", "full"])
|
|
|
|
assert result.exit_code == 0
|
|
config = Path("config.yaml").read_text()
|
|
assert "process_orchestration" in config
|
|
assert "routing" in config
|
|
|
|
def test_init_custom_output(self, runner, temp_dir):
|
|
"""Test init with custom output file."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["init", "-o", "custom.yaml"])
|
|
|
|
assert result.exit_code == 0
|
|
assert Path("custom.yaml").exists()
|
|
|
|
def test_init_no_overwrite(self, runner, temp_dir):
|
|
"""Test init won't overwrite without force."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text("existing content")
|
|
|
|
result = runner.invoke(cli, ["init"], input="n\n")
|
|
|
|
assert Path("config.yaml").read_text() == "existing content"
|
|
|
|
def test_init_force_overwrite(self, runner, temp_dir):
|
|
"""Test init with force flag."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text("existing content")
|
|
|
|
result = runner.invoke(cli, ["init", "-f"])
|
|
|
|
assert result.exit_code == 0
|
|
assert Path("config.yaml").read_text() != "existing content"
|
|
|
|
def test_init_list_templates(self, runner, temp_dir):
|
|
"""Test listing available templates."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["init", "--list-templates"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "basic" in result.output
|
|
assert "orchestration" in result.output
|
|
assert "asgi" in result.output
|
|
assert "full" in result.output
|
|
|
|
|
|
# =============================================================================
|
|
# Config Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestConfigCommand:
|
|
"""Tests for config command."""
|
|
|
|
def test_config_validate_success(self, runner, temp_dir, sample_config):
|
|
"""Test successful config validation."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["config", "validate"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "valid" in result.output.lower()
|
|
|
|
def test_config_validate_missing_file(self, runner, temp_dir):
|
|
"""Test validation with missing config."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["config", "validate"])
|
|
|
|
assert result.exit_code != 0
|
|
|
|
def test_config_show_yaml(self, runner, temp_dir, sample_config):
|
|
"""Test showing config as YAML."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["config", "show"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_config_show_json(self, runner, temp_dir, sample_config):
|
|
"""Test showing config as JSON."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["config", "show", "--format", "json"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_config_show_section(self, runner, temp_dir, sample_config):
|
|
"""Test showing specific section."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["config", "show", "--section", "server"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_config_get_value(self, runner, temp_dir, sample_config):
|
|
"""Test getting a specific value."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["config", "get", "server.port"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "8080" in result.output
|
|
|
|
def test_config_get_nested(self, runner, temp_dir, sample_config):
|
|
"""Test getting nested value."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["config", "get", "server.host"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "0.0.0.0" in result.output
|
|
|
|
def test_config_set_value(self, runner, temp_dir, sample_config):
|
|
"""Test setting a value."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["config", "set", "server.port", "9090"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
# Verify change
|
|
config = yaml.safe_load(Path("config.yaml").read_text())
|
|
assert config["server"]["port"] == 9090
|
|
|
|
def test_config_diff(self, runner, temp_dir):
|
|
"""Test comparing two configs."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
config1 = {"server": {"port": 8080}}
|
|
config2 = {"server": {"port": 9090}}
|
|
|
|
Path("config1.yaml").write_text(yaml.dump(config1))
|
|
Path("config2.yaml").write_text(yaml.dump(config2))
|
|
|
|
result = runner.invoke(cli, ["config", "diff", "config1.yaml", "config2.yaml"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
|
|
# =============================================================================
|
|
# PS/Status Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestPsCommand:
|
|
"""Tests for ps/status command."""
|
|
|
|
def test_ps_no_services(self, runner, temp_dir, sample_config):
|
|
"""Test ps with no running services."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["ps"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "No services" in result.output or "running" not in result.output.lower()
|
|
|
|
def test_ps_with_services(self, runner, temp_dir, sample_config):
|
|
"""Test ps with services in state."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
# Create state with services
|
|
state_dir = Path(".pyserve")
|
|
state_dir.mkdir(exist_ok=True)
|
|
state_data = {
|
|
"version": "1.0",
|
|
"project": "test",
|
|
"config_file": "",
|
|
"config_hash": "",
|
|
"started_at": None,
|
|
"daemon_pid": None,
|
|
"services": {
|
|
"api": {
|
|
"name": "api",
|
|
"state": "running",
|
|
"pid": 1234,
|
|
"port": 9000,
|
|
"workers": 2,
|
|
"started_at": time.time(),
|
|
"restart_count": 0,
|
|
"health": {"status": "healthy"},
|
|
"config_hash": "",
|
|
}
|
|
},
|
|
}
|
|
(state_dir / "state.json").write_text(json.dumps(state_data))
|
|
|
|
result = runner.invoke(cli, ["ps"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "api" in result.output
|
|
|
|
def test_ps_show_all(self, runner, temp_dir, sample_config):
|
|
"""Test ps with --all flag."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["ps", "-a"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_ps_quiet(self, runner, temp_dir, sample_config):
|
|
"""Test ps with --quiet flag."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
# Create state with services
|
|
state_dir = Path(".pyserve")
|
|
state_dir.mkdir(exist_ok=True)
|
|
state_data = {
|
|
"version": "1.0",
|
|
"project": "test",
|
|
"config_file": "",
|
|
"config_hash": "",
|
|
"started_at": None,
|
|
"daemon_pid": None,
|
|
"services": {
|
|
"api": {
|
|
"name": "api",
|
|
"state": "running",
|
|
"pid": 1234,
|
|
"port": 9000,
|
|
"workers": 2,
|
|
"started_at": time.time(),
|
|
"restart_count": 0,
|
|
"health": {},
|
|
"config_hash": "",
|
|
}
|
|
},
|
|
}
|
|
(state_dir / "state.json").write_text(json.dumps(state_data))
|
|
|
|
result = runner.invoke(cli, ["ps", "-q"])
|
|
|
|
assert result.exit_code == 0
|
|
# Quiet mode should only show service names
|
|
assert "api" in result.output
|
|
|
|
def test_ps_json_format(self, runner, temp_dir, sample_config):
|
|
"""Test ps with JSON output."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
# Create state with services
|
|
state_dir = Path(".pyserve")
|
|
state_dir.mkdir(exist_ok=True)
|
|
state_data = {
|
|
"version": "1.0",
|
|
"project": "test",
|
|
"config_file": "",
|
|
"config_hash": "",
|
|
"started_at": None,
|
|
"daemon_pid": None,
|
|
"services": {
|
|
"api": {
|
|
"name": "api",
|
|
"state": "running",
|
|
"pid": 1234,
|
|
"port": 9000,
|
|
"workers": 2,
|
|
"started_at": time.time(),
|
|
"restart_count": 0,
|
|
"health": {},
|
|
"config_hash": "",
|
|
}
|
|
},
|
|
}
|
|
(state_dir / "state.json").write_text(json.dumps(state_data))
|
|
|
|
result = runner.invoke(cli, ["ps", "--format", "json"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_status_alias(self, runner, temp_dir, sample_config):
|
|
"""Test that 'status' is alias for 'ps'."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["status"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
|
|
# =============================================================================
|
|
# Up/Down Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestUpDownCommands:
|
|
"""Tests for up and down commands."""
|
|
|
|
def test_up_no_config(self, runner, temp_dir):
|
|
"""Test up without config file."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
result = runner.invoke(cli, ["up"])
|
|
|
|
assert result.exit_code != 0
|
|
|
|
def test_up_help(self, runner):
|
|
"""Test up command help."""
|
|
result = runner.invoke(cli, ["up", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Start services" in result.output
|
|
assert "--detach" in result.output
|
|
assert "--scale" in result.output
|
|
|
|
def test_down_no_daemon(self, runner, temp_dir, sample_config):
|
|
"""Test down when no daemon running."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["down"])
|
|
|
|
# Should succeed even with no daemon
|
|
assert result.exit_code == 0
|
|
|
|
def test_down_help(self, runner):
|
|
"""Test down command help."""
|
|
result = runner.invoke(cli, ["down", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Stop" in result.output
|
|
assert "--timeout" in result.output
|
|
|
|
|
|
# =============================================================================
|
|
# Start/Stop/Restart Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestServiceCommands:
|
|
"""Tests for start/stop/restart commands."""
|
|
|
|
def test_start_help(self, runner):
|
|
"""Test start command help."""
|
|
result = runner.invoke(cli, ["start", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Start one or more services" in result.output
|
|
|
|
def test_stop_help(self, runner):
|
|
"""Test stop command help."""
|
|
result = runner.invoke(cli, ["stop", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Stop one or more services" in result.output
|
|
|
|
def test_restart_help(self, runner):
|
|
"""Test restart command help."""
|
|
result = runner.invoke(cli, ["restart", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Restart one or more services" in result.output
|
|
|
|
def test_start_requires_services(self, runner, temp_dir, sample_config):
|
|
"""Test start requires service names."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["start"])
|
|
|
|
# Should fail without service names
|
|
assert result.exit_code != 0
|
|
|
|
def test_stop_requires_services(self, runner, temp_dir, sample_config):
|
|
"""Test stop requires service names."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["stop"])
|
|
|
|
# Should fail without service names
|
|
assert result.exit_code != 0
|
|
|
|
|
|
# =============================================================================
|
|
# Logs Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestLogsCommand:
|
|
"""Tests for logs command."""
|
|
|
|
def test_logs_help(self, runner):
|
|
"""Test logs command help."""
|
|
result = runner.invoke(cli, ["logs", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "View service logs" in result.output
|
|
assert "--follow" in result.output
|
|
assert "--tail" in result.output
|
|
|
|
def test_logs_no_services(self, runner, temp_dir, sample_config):
|
|
"""Test logs with no services."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
Path(".pyserve").mkdir()
|
|
|
|
result = runner.invoke(cli, ["logs"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
def test_logs_with_log_file(self, runner, temp_dir, sample_config):
|
|
"""Test logs with existing log file."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
# Create state and logs
|
|
state_dir = Path(".pyserve")
|
|
state_dir.mkdir()
|
|
logs_dir = state_dir / "logs"
|
|
logs_dir.mkdir()
|
|
|
|
(logs_dir / "api.log").write_text("2024-01-01 12:00:00 INFO: Test log\n")
|
|
|
|
state_data = {
|
|
"version": "1.0",
|
|
"project": "test",
|
|
"config_file": "",
|
|
"config_hash": "",
|
|
"started_at": None,
|
|
"daemon_pid": None,
|
|
"services": {
|
|
"api": {
|
|
"name": "api",
|
|
"state": "running",
|
|
"pid": 1234,
|
|
"port": 9000,
|
|
"workers": 1,
|
|
"started_at": None,
|
|
"restart_count": 0,
|
|
"health": {},
|
|
"config_hash": "",
|
|
}
|
|
},
|
|
}
|
|
(state_dir / "state.json").write_text(json.dumps(state_data))
|
|
|
|
result = runner.invoke(cli, ["logs", "api"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Test log" in result.output
|
|
|
|
|
|
# =============================================================================
|
|
# Health Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestHealthCommand:
|
|
"""Tests for health command."""
|
|
|
|
def test_health_help(self, runner):
|
|
"""Test health command help."""
|
|
result = runner.invoke(cli, ["health", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Check health" in result.output
|
|
assert "--timeout" in result.output
|
|
|
|
def test_health_no_services(self, runner, temp_dir, sample_config):
|
|
"""Test health with no services."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
Path(".pyserve").mkdir()
|
|
|
|
result = runner.invoke(cli, ["health"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "No services" in result.output
|
|
|
|
|
|
# =============================================================================
|
|
# Scale Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestScaleCommand:
|
|
"""Tests for scale command."""
|
|
|
|
def test_scale_help(self, runner):
|
|
"""Test scale command help."""
|
|
result = runner.invoke(cli, ["scale", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Scale services" in result.output
|
|
assert "SERVICE=NUM" in result.output
|
|
|
|
def test_scale_invalid_format(self, runner, temp_dir, sample_config):
|
|
"""Test scale with invalid format."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["scale", "api"]) # Missing =NUM
|
|
|
|
assert result.exit_code != 0
|
|
|
|
def test_scale_requires_argument(self, runner, temp_dir, sample_config):
|
|
"""Test scale requires argument."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
result = runner.invoke(cli, ["scale"])
|
|
|
|
assert result.exit_code != 0
|
|
|
|
|
|
# =============================================================================
|
|
# Top Command Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestTopCommand:
|
|
"""Tests for top command."""
|
|
|
|
def test_top_help(self, runner):
|
|
"""Test top command help."""
|
|
result = runner.invoke(cli, ["top", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "Live monitoring" in result.output
|
|
assert "--refresh" in result.output
|
|
|
|
def test_top_no_daemon(self, runner, temp_dir, sample_config):
|
|
"""Test top when no daemon running."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
Path(".pyserve").mkdir()
|
|
|
|
result = runner.invoke(cli, ["top"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "No services running" in result.output
|
|
|
|
|
|
# =============================================================================
|
|
# Service Runner Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestServiceRunner:
|
|
"""Tests for ServiceRunner class."""
|
|
|
|
def test_parse_services_empty(self, temp_dir):
|
|
"""Test parsing with no services."""
|
|
from pyserve.config import Config
|
|
from pyserve.ctl._runner import ServiceRunner
|
|
|
|
config = Config()
|
|
state_manager = StateManager(temp_dir / ".pyserve")
|
|
|
|
runner = ServiceRunner(config, state_manager)
|
|
assert len(runner.get_services()) == 0
|
|
|
|
def test_parse_services_from_config(self, temp_dir, sample_config):
|
|
"""Test parsing services from config."""
|
|
from pyserve.config import Config
|
|
from pyserve.ctl._runner import ServiceRunner
|
|
|
|
config = Config.from_yaml(str(sample_config))
|
|
state_manager = StateManager(temp_dir / ".pyserve")
|
|
|
|
runner = ServiceRunner(config, state_manager)
|
|
services = runner.get_services()
|
|
|
|
assert "api" in services
|
|
assert "admin" in services
|
|
assert services["api"].workers == 2
|
|
assert services["admin"].workers == 1
|
|
|
|
def test_get_service(self, temp_dir, sample_config):
|
|
"""Test getting single service."""
|
|
from pyserve.config import Config
|
|
from pyserve.ctl._runner import ServiceRunner
|
|
|
|
config = Config.from_yaml(str(sample_config))
|
|
state_manager = StateManager(temp_dir / ".pyserve")
|
|
|
|
runner = ServiceRunner(config, state_manager)
|
|
|
|
service = runner.get_service("api")
|
|
assert service is not None
|
|
assert service.name == "api"
|
|
|
|
assert runner.get_service("nonexistent") is None
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestIntegration:
|
|
"""Integration tests for complete workflows."""
|
|
|
|
def test_init_then_validate(self, runner, temp_dir):
|
|
"""Test init followed by config validate."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
# Init
|
|
result = runner.invoke(cli, ["init"])
|
|
assert result.exit_code == 0
|
|
|
|
# Validate
|
|
result = runner.invoke(cli, ["config", "validate"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_init_then_show_config(self, runner, temp_dir):
|
|
"""Test init followed by config show."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
# Init
|
|
result = runner.invoke(cli, ["init", "-t", "orchestration"])
|
|
assert result.exit_code == 0
|
|
|
|
# Show config
|
|
result = runner.invoke(cli, ["config", "show"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_state_persistence(self, runner, temp_dir):
|
|
"""Test that state persists across commands."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
runner.invoke(cli, ["init"])
|
|
|
|
# Create state manually
|
|
state_dir = Path(".pyserve")
|
|
state_manager = StateManager(state_dir)
|
|
state_manager.update_service("api", state="running", pid=1234, port=9000)
|
|
|
|
# Check state with ps
|
|
result = runner.invoke(cli, ["ps"])
|
|
assert result.exit_code == 0
|
|
assert "api" in result.output
|
|
|
|
def test_environment_variables(self, runner, temp_dir, sample_config):
|
|
"""Test environment variable support."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
# Test PYSERVE_CONFIG env var
|
|
result = runner.invoke(
|
|
cli,
|
|
["config", "show"],
|
|
env={"PYSERVE_CONFIG": "config.yaml"}
|
|
)
|
|
assert result.exit_code == 0
|
|
|
|
|
|
# =============================================================================
|
|
# Edge Cases and Error Handling
|
|
# =============================================================================
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Tests for edge cases and error handling."""
|
|
|
|
def test_corrupted_state_file(self, runner, temp_dir, sample_config):
|
|
"""Test handling of corrupted state file."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
|
|
state_dir = Path(".pyserve")
|
|
state_dir.mkdir()
|
|
(state_dir / "state.json").write_text("invalid json {{{")
|
|
|
|
# Should handle gracefully
|
|
result = runner.invoke(cli, ["ps"])
|
|
assert result.exit_code == 0
|
|
|
|
def test_invalid_yaml_config(self, runner, temp_dir):
|
|
"""Test handling of invalid YAML config."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text("invalid: yaml: content:")
|
|
|
|
result = runner.invoke(cli, ["config", "validate"])
|
|
assert result.exit_code != 0
|
|
|
|
def test_missing_required_fields(self, runner, temp_dir):
|
|
"""Test config with missing fields."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text("extensions: []")
|
|
|
|
result = runner.invoke(cli, ["config", "validate"])
|
|
# Should validate (minimal config is OK)
|
|
assert result.exit_code == 0
|
|
|
|
def test_empty_config_file(self, runner, temp_dir):
|
|
"""Test empty config file."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text("")
|
|
|
|
result = runner.invoke(cli, ["config", "validate"])
|
|
assert result.exit_code != 0
|
|
|
|
def test_nonexistent_service(self, runner, temp_dir, sample_config):
|
|
"""Test operations on nonexistent service."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
Path("config.yaml").write_text(sample_config.read_text())
|
|
Path(".pyserve").mkdir()
|
|
|
|
result = runner.invoke(cli, ["logs", "nonexistent"])
|
|
# Should handle gracefully
|
|
assert result.exit_code == 0
|
|
|
|
def test_special_characters_in_service_name(self, runner, temp_dir):
|
|
"""Test handling of special characters."""
|
|
with runner.isolated_filesystem(temp_dir=str(temp_dir)):
|
|
runner.invoke(cli, ["init"])
|
|
|
|
state_dir = Path(".pyserve")
|
|
state_manager = StateManager(state_dir)
|
|
state_manager.update_service("api-v2", state="running")
|
|
state_manager.update_service("admin_service", state="stopped")
|
|
|
|
result = runner.invoke(cli, ["ps", "-a"])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|