File: main.py

package info (click to toggle)
python-azure 20250603%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 851,724 kB
  • sloc: python: 7,362,925; ansic: 804; javascript: 287; makefile: 195; sh: 145; xml: 109
file content (142 lines) | stat: -rw-r--r-- 4,900 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
from mcp.server.fastmcp import FastMCP
import subprocess
from typing import Dict, List, Optional, Any
import logging
import sys
import os
from pathlib import Path

logger = logging.getLogger()
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stderr)
logger.addHandler(handler)

# Log the Python executable and environment information
logger.info(f"Running with Python executable: {sys.executable}")
logger.info(f"Virtual environment path: {os.environ.get('VIRTUAL_ENV', 'Not running in a virtual environment')}")
logger.info(f"Working directory: {os.getcwd()}")

_REPO_ROOT = Path(__file__).parents[3]
_TOX_INI_PATH = os.path.abspath(_REPO_ROOT / "eng" / "tox" / "tox.ini")

# Initialize server
mcp = FastMCP("validation")

def run_command(command: List[str], cwd: Optional[str] = None) -> Dict[str, Any]:
    """Run a command and return the result."""
    logger.info(f"Running command: {' '.join(command)}")
    logger.info(f"Virtual environment path: {os.environ.get('VIRTUAL_ENV', 'Not running in a virtual environment')}")
    
    try:
        if not cwd:
            cwd = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))
            logger.info(f"Using current working directory: {cwd}")
        
        # On Windows, use shell=True for better command handling
        use_cmd = os.name == "nt"
        
        if use_cmd:
            # Convert list to single command string for Windows shell
            command = ["cmd.exe", "/c"] + command
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                cwd=cwd,
                stdin=subprocess.DEVNULL,
            )
        else:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                cwd=cwd,
            )
        
        return {
            "success": result.returncode == 0,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "code": result.returncode
        }
    except Exception as e:
        logger.error(f"Error running command: {e}")
        return {
            "success": False,
            "stdout": "",
            "stderr": str(e),
            "code": 1,
        }

@mcp.tool("verify_setup")
def verify_setup_tool() -> Dict[str, Any]:
    """Verify machine is set up correctly for development.

    """
    def verify_installation(command: List[str], name: str) -> Dict[str, Any]:
        """Helper function to verify installation of a tool."""
        logger.info(f"Checking installation of {name} with command: {command}")
        
        result = run_command(command)
        
        if not result["success"]:
            logger.error(f"{name} verification failed. Exit code: {result['code']}")
            logger.error(f"stderr: {result['stderr']}")
            return {
                "success": False,
                "message": f"{name} is not installed or not available in PATH.",
                "details": {
                    "stdout": result["stdout"].split(f"{name}: commands")[1] if f"{name}: commands" in result["stdout"] else result["stdout"],
                    "stderr": result["stderr"],
                    "exit_code": result["code"]
                }
            }
            
        version_output = result["stdout"].strip() or "No version output"
        logger.info(f"{name} version output: '{version_output}'")
        
        return {
            "success": True,
            "message": f"{name} is installed. Version: {version_output}"
        }

    results = {
        "node": verify_installation(["node", "--version"], "Node.js"),
        "python": verify_installation(["python", "--version"], "Python")
    }

    # Check if tox is installed
    logger.info("Checking tox installation...")
    
    tox_command = ["tox", "--version", "-c", _TOX_INI_PATH]
    results["tox"] = verify_installation(tox_command, "tox")

    return results


@mcp.tool("tox")
def tox_tool(package_path: str, environment: str, config_file: Optional[str] = None) -> Dict[str, Any]:
    """Run tox tests on a Python package.
    
    Args:
        package_path: Path to the Python package to test
        environment: tox environment to run (e.g., 'pylint', 'mypy')
        config_file: Optional path to a tox configuration file
    """
    # Normalize config file path
    config_path = config_file if config_file is not None else _TOX_INI_PATH
    
    command = ["tox", "run"]
        
    command.extend(["-e", environment])
    command.extend(["-c", config_path])
    command.extend(["--root", package_path])
    
    logger.info(f"Running tox with command: {command}")
    logger.info(f"Working directory: {package_path}")
    
    return run_command(command, cwd=package_path)

# Run the MCP server
if __name__ == "__main__":
    mcp.run(transport='stdio')