mirror of
https://github.com/katanemo/plano.git
synced 2026-06-17 15:25:17 +02:00
Musa
parent
2d0dcce365
commit
67a52fbcd7
5 changed files with 21 additions and 865 deletions
|
|
@ -1,218 +0,0 @@
|
|||
import pytest
|
||||
import os
|
||||
import tempfile
|
||||
from unittest import mock
|
||||
from click.testing import CliRunner
|
||||
from planoai.main import validate, validate_config_file
|
||||
|
||||
|
||||
class TestValidateConfigFile:
|
||||
"""Tests for the validate_config_file function."""
|
||||
|
||||
def test_valid_config(self, tmp_path):
|
||||
"""Test validation of a valid config file."""
|
||||
config_content = """
|
||||
version: v0.3.0
|
||||
|
||||
listeners:
|
||||
- type: model
|
||||
name: llm_gateway
|
||||
port: 12000
|
||||
|
||||
model_providers:
|
||||
- model: openai/gpt-4o-mini
|
||||
access_key: $OPENAI_API_KEY
|
||||
default: true
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
result = validate_config_file(str(config_file))
|
||||
|
||||
assert result["valid"] is True
|
||||
assert len(result["errors"]) == 0
|
||||
assert result["config"] is not None
|
||||
assert len(result["summary"]["model_providers"]) == 1
|
||||
assert result["summary"]["model_providers"][0]["model"] == "openai/gpt-4o-mini"
|
||||
assert result["summary"]["model_providers"][0]["default"] is True
|
||||
assert "OPENAI_API_KEY" in result["summary"]["env_vars_required"]
|
||||
|
||||
def test_invalid_yaml_syntax(self, tmp_path):
|
||||
"""Test validation fails for invalid YAML syntax."""
|
||||
config_content = """
|
||||
version: v0.3.0
|
||||
listeners:
|
||||
- type: model
|
||||
name: [invalid yaml
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
result = validate_config_file(str(config_file))
|
||||
|
||||
assert result["valid"] is False
|
||||
assert any("Invalid YAML" in error for error in result["errors"])
|
||||
|
||||
def test_file_not_found(self):
|
||||
"""Test validation fails for non-existent file."""
|
||||
result = validate_config_file("/nonexistent/path/config.yaml")
|
||||
|
||||
assert result["valid"] is False
|
||||
assert any("not found" in error for error in result["errors"])
|
||||
|
||||
def test_multiple_model_providers(self, tmp_path):
|
||||
"""Test config with multiple model providers."""
|
||||
config_content = """
|
||||
version: v0.3.0
|
||||
|
||||
listeners:
|
||||
- type: model
|
||||
name: llm_gateway
|
||||
port: 12000
|
||||
|
||||
model_providers:
|
||||
- model: openai/gpt-4o-mini
|
||||
access_key: $OPENAI_API_KEY
|
||||
default: true
|
||||
|
||||
- model: anthropic/claude-3-5-sonnet
|
||||
access_key: $ANTHROPIC_API_KEY
|
||||
|
||||
- model: mistral/mistral-large
|
||||
access_key: $MISTRAL_API_KEY
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
result = validate_config_file(str(config_file))
|
||||
|
||||
assert result["valid"] is True
|
||||
assert len(result["summary"]["model_providers"]) == 3
|
||||
assert set(result["summary"]["env_vars_required"]) == {
|
||||
"OPENAI_API_KEY",
|
||||
"ANTHROPIC_API_KEY",
|
||||
"MISTRAL_API_KEY",
|
||||
}
|
||||
|
||||
def test_legacy_listener_format(self, tmp_path):
|
||||
"""Test config with legacy listener format."""
|
||||
config_content = """
|
||||
version: v0.1.0
|
||||
|
||||
listeners:
|
||||
egress_traffic:
|
||||
address: 0.0.0.0
|
||||
port: 12000
|
||||
ingress_traffic:
|
||||
address: 0.0.0.0
|
||||
port: 10000
|
||||
|
||||
llm_providers:
|
||||
- model: openai/gpt-4o
|
||||
access_key: $OPENAI_API_KEY
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
result = validate_config_file(str(config_file))
|
||||
|
||||
assert result["valid"] is True
|
||||
assert len(result["summary"]["listeners"]) == 2
|
||||
|
||||
# Check listener ports are correctly extracted
|
||||
ports = [l["port"] for l in result["summary"]["listeners"]]
|
||||
assert 12000 in ports
|
||||
assert 10000 in ports
|
||||
|
||||
|
||||
class TestValidateCommand:
|
||||
"""Tests for the CLI validate command."""
|
||||
|
||||
def test_validate_command_with_valid_file(self, tmp_path):
|
||||
"""Test validate command with a valid config file."""
|
||||
config_content = """
|
||||
version: v0.3.0
|
||||
|
||||
listeners:
|
||||
- type: model
|
||||
name: llm_gateway
|
||||
port: 12000
|
||||
|
||||
model_providers:
|
||||
- model: openai/gpt-4o-mini
|
||||
access_key: $OPENAI_API_KEY
|
||||
default: true
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(validate, [str(config_file)])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "valid" in result.output.lower() or "✓" in result.output
|
||||
|
||||
def test_validate_command_with_invalid_file(self, tmp_path):
|
||||
"""Test validate command fails with invalid config."""
|
||||
config_content = """
|
||||
version: v0.3.0
|
||||
invalid_yaml: [broken
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(validate, [str(config_file)])
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert "invalid" in result.output.lower() or "✗" in result.output
|
||||
|
||||
def test_validate_command_auto_detect_config(self, tmp_path, monkeypatch):
|
||||
"""Test validate command auto-detects config.yaml in current directory."""
|
||||
config_content = """
|
||||
version: v0.3.0
|
||||
|
||||
listeners:
|
||||
- type: model
|
||||
name: test
|
||||
port: 12000
|
||||
|
||||
model_providers:
|
||||
- model: openai/gpt-4o
|
||||
access_key: $OPENAI_API_KEY
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
# Change to the temp directory
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(validate, [])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "valid" in result.output.lower() or "✓" in result.output
|
||||
|
||||
def test_validate_command_quiet_mode(self, tmp_path):
|
||||
"""Test validate command with --quiet flag."""
|
||||
config_content = """
|
||||
version: v0.3.0
|
||||
|
||||
listeners:
|
||||
- type: model
|
||||
name: llm_gateway
|
||||
port: 12000
|
||||
|
||||
model_providers:
|
||||
- model: openai/gpt-4o-mini
|
||||
access_key: $OPENAI_API_KEY
|
||||
"""
|
||||
config_file = tmp_path / "config.yaml"
|
||||
config_file.write_text(config_content)
|
||||
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(validate, [str(config_file), "--quiet"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
# Quiet mode should have minimal output (no tables)
|
||||
assert "Model Providers" not in result.output
|
||||
|
|
@ -1,163 +0,0 @@
|
|||
import pytest
|
||||
from unittest import mock
|
||||
from planoai.main import (
|
||||
get_version,
|
||||
get_latest_version,
|
||||
parse_version,
|
||||
check_version_status,
|
||||
PYPI_URL,
|
||||
)
|
||||
|
||||
|
||||
class TestParseVersion:
|
||||
"""Tests for version string parsing."""
|
||||
|
||||
def test_parse_simple_version(self):
|
||||
assert parse_version("1.0.0") == (1, 0, 0)
|
||||
assert parse_version("0.4.1") == (0, 4, 1)
|
||||
assert parse_version("10.20.30") == (10, 20, 30)
|
||||
|
||||
def test_parse_two_part_version(self):
|
||||
assert parse_version("1.0") == (1, 0)
|
||||
assert parse_version("2.5") == (2, 5)
|
||||
|
||||
def test_parse_version_with_prerelease(self):
|
||||
# Pre-release suffixes should be stripped
|
||||
assert parse_version("0.4.1a1") == (0, 4, 1)
|
||||
assert parse_version("1.0.0beta2") == (1, 0, 0)
|
||||
assert parse_version("2.0.0rc1") == (2, 0, 0)
|
||||
|
||||
|
||||
class TestCheckVersionStatus:
|
||||
"""Tests for version comparison logic."""
|
||||
|
||||
def test_current_equals_latest(self):
|
||||
status = check_version_status("0.4.1", "0.4.1")
|
||||
assert status["is_outdated"] is False
|
||||
assert status["current"] == "0.4.1"
|
||||
assert status["latest"] == "0.4.1"
|
||||
assert status["message"] is None
|
||||
|
||||
def test_current_is_outdated(self):
|
||||
status = check_version_status("0.4.1", "0.5.0")
|
||||
assert status["is_outdated"] is True
|
||||
assert status["current"] == "0.4.1"
|
||||
assert status["latest"] == "0.5.0"
|
||||
assert "Update available" in status["message"]
|
||||
assert "0.5.0" in status["message"]
|
||||
|
||||
def test_current_is_newer(self):
|
||||
# Dev version might be newer than PyPI
|
||||
status = check_version_status("0.5.0", "0.4.1")
|
||||
assert status["is_outdated"] is False
|
||||
assert status["message"] is None
|
||||
|
||||
def test_major_version_outdated(self):
|
||||
status = check_version_status("0.4.1", "1.0.0")
|
||||
assert status["is_outdated"] is True
|
||||
|
||||
def test_minor_version_outdated(self):
|
||||
status = check_version_status("0.4.1", "0.5.0")
|
||||
assert status["is_outdated"] is True
|
||||
|
||||
def test_patch_version_outdated(self):
|
||||
status = check_version_status("0.4.1", "0.4.2")
|
||||
assert status["is_outdated"] is True
|
||||
|
||||
def test_latest_is_none(self):
|
||||
# When PyPI check fails
|
||||
status = check_version_status("0.4.1", None)
|
||||
assert status["is_outdated"] is False
|
||||
assert status["latest"] is None
|
||||
assert status["message"] is None
|
||||
|
||||
|
||||
class TestGetLatestVersion:
|
||||
"""Tests for PyPI version fetching."""
|
||||
|
||||
def test_successful_fetch(self):
|
||||
mock_response = mock.Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"info": {"version": "0.5.0"}}
|
||||
|
||||
with mock.patch("requests.get", return_value=mock_response):
|
||||
version = get_latest_version()
|
||||
assert version == "0.5.0"
|
||||
|
||||
def test_network_error(self):
|
||||
import requests
|
||||
|
||||
with mock.patch(
|
||||
"requests.get", side_effect=requests.RequestException("Network error")
|
||||
):
|
||||
version = get_latest_version()
|
||||
assert version is None
|
||||
|
||||
def test_timeout(self):
|
||||
import requests
|
||||
|
||||
with mock.patch("requests.get", side_effect=requests.Timeout("Timeout")):
|
||||
version = get_latest_version()
|
||||
assert version is None
|
||||
|
||||
def test_invalid_json(self):
|
||||
mock_response = mock.Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.side_effect = ValueError("Invalid JSON")
|
||||
|
||||
with mock.patch("requests.get", return_value=mock_response):
|
||||
version = get_latest_version()
|
||||
assert version is None
|
||||
|
||||
def test_404_response(self):
|
||||
mock_response = mock.Mock()
|
||||
mock_response.status_code = 404
|
||||
|
||||
with mock.patch("requests.get", return_value=mock_response):
|
||||
version = get_latest_version()
|
||||
assert version is None
|
||||
|
||||
|
||||
class TestVersionCheckIntegration:
|
||||
"""Integration tests simulating version check scenarios."""
|
||||
|
||||
def test_outdated_version_message(self, capsys):
|
||||
"""Simulate an outdated version scenario."""
|
||||
from rich.console import Console
|
||||
|
||||
console = Console(force_terminal=True)
|
||||
current_version = "0.4.1"
|
||||
|
||||
# Mock PyPI returning a newer version
|
||||
mock_response = mock.Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"info": {"version": "0.5.0"}}
|
||||
|
||||
with mock.patch("requests.get", return_value=mock_response):
|
||||
latest = get_latest_version()
|
||||
status = check_version_status(current_version, latest)
|
||||
|
||||
assert status["is_outdated"] is True
|
||||
assert status["latest"] == "0.5.0"
|
||||
|
||||
def test_up_to_date_version(self):
|
||||
"""Simulate an up-to-date version scenario."""
|
||||
current_version = "0.4.1"
|
||||
|
||||
mock_response = mock.Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"info": {"version": "0.4.1"}}
|
||||
|
||||
with mock.patch("requests.get", return_value=mock_response):
|
||||
latest = get_latest_version()
|
||||
status = check_version_status(current_version, latest)
|
||||
|
||||
assert status["is_outdated"] is False
|
||||
|
||||
def test_skip_version_check_env_var(self, monkeypatch):
|
||||
"""Test that PLANO_SKIP_VERSION_CHECK skips the check."""
|
||||
monkeypatch.setenv("PLANO_SKIP_VERSION_CHECK", "1")
|
||||
|
||||
import os
|
||||
|
||||
assert os.environ.get("PLANO_SKIP_VERSION_CHECK") == "1"
|
||||
Loading…
Add table
Add a link
Reference in a new issue