diff --git a/server/mcp_server_vefaas_function/pyproject.toml b/server/mcp_server_vefaas_function/pyproject.toml index ee08fbe1..0bed2dbe 100644 --- a/server/mcp_server_vefaas_function/pyproject.toml +++ b/server/mcp_server_vefaas_function/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mcp-server-vefaas-function" -version = "0.0.6" +version = "0.0.7" description = "MCP server for managing veFaaS (Volc Engine Function as a Service) functions" readme = "README.md" requires-python = ">=3.12" diff --git a/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_cli_sdk/deploy.py b/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_cli_sdk/deploy.py index 0350baa4..43f27cab 100644 --- a/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_cli_sdk/deploy.py +++ b/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_cli_sdk/deploy.py @@ -72,12 +72,66 @@ # veFaaS CLI config .vefaas/ +vefaas.yaml """ # Default Caddyfile name for static sites DEFAULT_CADDYFILE_NAME = "DefaultCaddyFile" +def generate_app_name_from_path(project_path: str) -> str: + """ + Generate application name from project path. + + Rules: + 1. Get project folder name + 2. Convert to lowercase + 3. Replace non-alphanumeric characters with hyphens + 4. Remove consecutive hyphens + 5. Remove leading/trailing hyphens + 6. Truncate if too long + 7. Add random suffix to avoid conflicts + + Args: + project_path: Absolute path to project + + Returns: + Processed app name, e.g., "my-project-abc123" + """ + import re + import random + import string + + # Get folder name + folder_name = os.path.basename(os.path.normpath(project_path)) + + # Convert to lowercase + name = folder_name.lower() + + # Replace non-alphanumeric with hyphens + name = re.sub(r'[^a-z0-9]', '-', name) + + # Remove consecutive hyphens + name = re.sub(r'-+', '-', name) + + # Remove leading/trailing hyphens + name = name.strip('-') + + # Use default if empty + if not name: + name = "app" + + # Truncate to reasonable length (reserve space for suffix) + max_base_len = 20 + if len(name) > max_base_len: + name = name[:max_base_len].rstrip('-') + + # Add random suffix to avoid conflicts + suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6)) + + return f"{name}-{suffix}" + + def read_gitignore_patterns(base_dir: str) -> List[str]: """Read .gitignore file patterns. Ported from vefaas-cli.""" gitignore_path = os.path.join(base_dir, ".gitignore") @@ -117,7 +171,7 @@ def create_ignore_filter( ) -> pathspec.PathSpec: """Create a pathspec filter from gitignore/vefaasignore patterns. Ported from vefaas-cli.""" all_patterns = gitignore_patterns + vefaasignore_patterns + (additional_patterns or []) - return pathspec.PathSpec.from_lines("gitwildmatch", all_patterns) + return pathspec.PathSpec.from_lines("gitignore", all_patterns) def render_default_caddyfile_content() -> str: @@ -245,10 +299,12 @@ def _check_api_error(self, result: dict, action: str) -> None: # Check for common error patterns and provide clear guidance if "already exists" in message.lower() or "duplicate" in message.lower(): raise ValueError( - f"[{action}] Name already exists: {message}\n" - "To update an existing application, get the application_id from `.vefaas/config.json` or console, " - "then call deploy_application with application_id parameter. " - "Do NOT use function_id directly - always use application_id for updates." + f"[{action}] NAME_CONFLICT: {message}\n" + "**YOU MUST ASK THE USER** to choose one of the following options:\n" + " 1. Update existing application: Get application_id from `.vefaas/config.json` or console, " + "then call deploy_application with application_id parameter\n" + " 2. Deploy as new application: Call deploy_application with a different name\n" + "DO NOT automatically choose an option. Present both choices to the user and wait for their decision." ) elif "not found" in message.lower(): raise ValueError(f"[{action}] Resource not found: {message}") @@ -382,6 +438,10 @@ def get_dependency_install_status(self, function_id: str) -> dict: """Get dependency installation task status""" return self.call("GetDependencyInstallTaskStatus", {"FunctionId": function_id}) + def create_dependency_install_task(self, function_id: str) -> dict: + """Create dependency installation task for Python projects""" + return self.call("CreateDependencyInstallTask", {"FunctionId": function_id}) + # ========== Application Operations ========== def get_application(self, app_id: str) -> dict: @@ -547,7 +607,7 @@ def wait_for_application_deploy( return {"success": True, "access_url": access_url} if status.lower() in ("deploy_fail", "deleted", "delete_fail"): - # Try to get detailed error from GetReleaseStatus (like vefaas-cli) + # Try to get detailed error from GetReleaseStatus error_details = {} function_id = None try: @@ -617,8 +677,10 @@ def wait_for_dependency_install( """Wait for Python dependency installation to complete.""" start_time = time.time() last_status = "" + poll_count = 0 while time.time() - start_time < timeout_seconds: + poll_count += 1 try: result = client.get_dependency_install_status(function_id) status = result.get("Result", {}).get("Status", "") @@ -627,16 +689,29 @@ def wait_for_dependency_install( logger.info(f"[dependency] Installation status: {status}") last_status = status + # Success status if status.lower() in ("succeeded", "success", "done"): return {"success": True, "status": status} + # Failed status if status.lower() == "failed": raise ValueError("Dependency installation failed") + # In-progress status (Dequeued = queued, InProgress = installing) + if status.lower() in ("dequeued", "inprogress", "in_progress", "pending"): + # Normal intermediate status, continue polling + pass + elif not status and poll_count > 3: + # Empty status may indicate no dependencies to install + return {"success": True, "status": "no_dependency"} + except ValueError: raise except Exception as e: logger.warning(f"[dependency] Error checking status: {e}") + # Multiple failures may indicate no dependency install task + if poll_count > 5: + return {"success": True, "status": "skipped"} time.sleep(poll_interval_seconds) @@ -702,7 +777,30 @@ def package_directory(directory: str, base_dir: Optional[str] = None, include_gi continue file_path = os.path.join(root, file) - zf.write(file_path, arcname) + + # Use ZipInfo to preserve file permissions (especially executable) + info = zipfile.ZipInfo(arcname) + + # Get original file permissions + file_stat = os.stat(file_path) + original_mode = file_stat.st_mode & 0o777 + + # Grant execute permission (755) for script files + script_extensions = ('.sh', '.bash', '.py', '.pl', '.rb') + if file.lower().endswith(script_extensions): + # Ensure execute permission: original | 0o755 + final_mode = original_mode | 0o755 + else: + final_mode = original_mode + + # Unix permissions stored in high 16 bits of external_attr + # Format: (permissions << 16) | (file_type << 28) + # 0o100000 = regular file + info.external_attr = (final_mode << 16) | (0o100000 << 16) + + # Read file content and write to zip + with open(file_path, 'rb') as f: + zf.writestr(info, f.read()) buffer.seek(0) zip_bytes = buffer.read() @@ -757,16 +855,21 @@ def log(msg: str): else: log(f"[config] Config region ({config_region}) differs from target region ({client.region}), will create new application") + # Auto-generate app name from project path if not provided for new app if not config.name and not config.application_id: - raise ValueError("Must provide name or application_id") + config.name = generate_app_name_from_path(config.project_path) + log(f"[config] Auto-generated app name: {config.name}") # 0. Early check for duplicate application name if config.name and not config.application_id: existing_app_id = client.find_application_by_name(config.name) if existing_app_id: raise ValueError( - f"Application name '{config.name}' already exists (ID: {existing_app_id}). " - f"To update this application, pass application_id='{existing_app_id}' parameter." + f"NAME_CONFLICT: Application name '{config.name}' already exists (existing_application_id: {existing_app_id}). " + f"**YOU MUST ASK THE USER** to choose one of the following options:\n" + f" 1. Update existing application: Call deploy_application with application_id='{existing_app_id}'\n" + f" 2. Deploy as new application: Call deploy_application with a different name\n" + f"DO NOT automatically choose an option. Present both choices to the user and wait for their decision." ) # 0.5 Early check: if updating existing app and deployment is in progress, return early @@ -963,8 +1066,13 @@ def log(msg: str): # 6. Wait for dependency installation (Python) if is_python: - log("[6/7] Waiting for dependency installation...") + log("[6/7] Installing dependencies...") try: + # Trigger dependency install task + client.create_dependency_install_task(target_function_id) + log(" → Dependency installation task created") + + # Wait for installation to complete wait_for_dependency_install(client, target_function_id) log(" → Dependencies installed") except Exception as e: diff --git a/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_server.py b/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_server.py index cf59858e..79f66778 100644 --- a/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_server.py +++ b/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_server.py @@ -192,7 +192,36 @@ def update_function(function_id: str, region: Optional[str] = None, except ApiException as e: raise ValueError(f"Failed to update function config: {str(e)}") - result["next_step"] = "Call release_function to publish changes (it will handle dependency installation automatically)" + # Save config to vefaas.yaml if project_path is provided + if project_path: + try: + from .vefaas_cli_sdk import write_config, VefaasConfig, FunctionConfig + + # Get function info for config + try: + func_req = volcenginesdkvefaas.GetFunctionRequest(id=function_id) + func_resp = api_instance.get_function(func_req) + runtime = func_resp.runtime + func_name = func_resp.name + except Exception: + runtime = None + func_name = None + + save_config = VefaasConfig( + function=FunctionConfig( + id=function_id, + runtime=runtime, + region=region, + ), + name=func_name, + command=command, + ) + write_config(project_path, save_config) + result["config_saved"] = True + except Exception as e: + logger.warning(f"Failed to save config: {e}") + + result["next_step"] = "Code updated. If you want to publish changes, call release_function. Note: release_function will trigger dependency installation and deploy to production." result["platform_url"] = f"https://console.volcengine.com/vefaas/region:vefaas+{region}/function/detail/{function_id}" return json.dumps(result, ensure_ascii=False, indent=2) @@ -256,64 +285,31 @@ def release_function(function_id: str, region: Optional[str] = None, skip_depend result["dependency_triggered"] = False result["dependency_status"] = "skipped" else: - logger.info("Checking if dependency installation is needed...") + logger.info("Triggering dependency installation...") + # Use SDK client for dependency operations + from .vefaas_cli_sdk import VeFaaSClient, wait_for_dependency_install + client = VeFaaSClient(ak, sk, token, region) + try: - dep_body = {"FunctionId": function_id} - now = datetime.datetime.utcnow() - dep_resp = request( - "POST", now, {}, {}, ak, sk, token, "CreateDependencyInstallTask", json.dumps(dep_body), region - ) + client.create_dependency_install_task(function_id) logger.info("Dependency install task created, waiting for completion...") result["dependency_triggered"] = True + + # Step 2: Wait for dependency installation using SDK logic + dep_result = wait_for_dependency_install(client, function_id, timeout_seconds=300) + result["dependency_status"] = dep_result.get("status", "succeeded") + logger.info(f"Dependency installation completed: {result['dependency_status']}") + + except ValueError as e: + # Dependency installation failed + result["dependency_triggered"] = True + result["dependency_status"] = "failed" + raise ValueError(f"Dependency installation failed: {e}") except Exception as e: # Dependency install may fail if no requirements.txt/package.json, that's OK - logger.info(f"Dependency install skipped or failed: {str(e)}") + logger.info(f"Dependency install skipped or not needed: {str(e)}") result["dependency_triggered"] = False - - # Step 2: Wait for dependency installation to complete - if result.get("dependency_triggered"): - timeout_seconds = 120 - poll_interval_seconds = 5 - start_time = time.time() - dep_status = None - - while time.time() - start_time < timeout_seconds: - try: - now = datetime.datetime.utcnow() - status_resp = request( - "POST", now, {}, {}, ak, sk, token, "GetDependencyInstallTaskStatus", - json.dumps({"FunctionId": function_id}), region, 5 - ) - dep_status = status_resp.get("Result", {}).get("Status") - - if dep_status == "InProgress" or dep_status is None: - time.sleep(poll_interval_seconds) - continue - else: - break - except Exception as ex: - logger.warning(f"Failed to get dependency status: {ex}") - break - - if dep_status == "Failed": - # Try to get log for debugging - try: - now = datetime.datetime.utcnow() - log_resp = request( - "POST", now, {}, {}, ak, sk, token, - "GetDependencyInstallTaskLogDownloadURI", - json.dumps({"FunctionId": function_id}), region, 5 - ) - log_url = log_resp.get("Result", {}).get("DownloadURL", "") - result["dependency_status"] = "failed" - result["dependency_log_url"] = log_url - raise ValueError(f"Dependency installation failed. Check logs: {log_url}") - except ValueError: - raise - except Exception: - raise ValueError("Dependency installation failed") - - result["dependency_status"] = "succeeded" if dep_status == "Succeeded" else dep_status + result["dependency_status"] = "skipped" # Step 3: Submit release request logger.info("Submitting release request...") @@ -1068,6 +1064,13 @@ def list_function_triggers(function_id: str, region: Optional[str] = None): **RECOMMENDED**: Call this BEFORE `deploy_application` to ensure correct configuration! +**Supported Runtimes**: +- **Node.js**: Next.js, Nuxt, Vite, VitePress, Rspress, Astro, Express, SvelteKit, Remix, CRA, Angular, Gatsby, etc. +- **Python**: FastAPI, Flask, Streamlit, Django, etc. +- **Static Sites**: HTML, Hugo, MkDocs, etc. + +> **Note**: Other runtimes (e.g., Go, Java, Rust, PHP) are NOT currently supported. If your project uses an unsupported runtime, you will need to deploy manually via the veFaaS console. + This tool analyzes the project structure and automatically detects: - Framework (Next.js, Vite, FastAPI, Flask, Streamlit, etc.) - Runtime and startup command @@ -1125,7 +1128,7 @@ def detect_project(project_path: str): This is the **recommended tool** for deploying applications. It handles the entire workflow automatically: 1. Detect project configuration -2. Build project (if needed) +2. Build project (for Node.js/static sites only, Python projects automatically skip this step) 3. Package and upload code to cloud storage 4. Create/update function with code 5. Wait for dependencies (Python) @@ -1139,20 +1142,21 @@ def detect_project(project_path: str): - This means subsequent deployments only need `project_path` - no need to specify IDs again. **Scenarios**: -- **New deployment**: Provide `project_path` + `name` + `start_command` + `build_command` (non-Python) + `port` +- **New Python app**: Provide `project_path` + `name` + `start_command` + `port` +- **New Node.js app**: Provide `project_path` + `name` + `start_command` + `build_command` + `port` - **Update existing app**: Just provide `project_path` (IDs read from config automatically) - **Update by ID**: Use `application_id` to update an existing application - **Deploy to different region**: Provide `project_path` + `name` + `region` (existing config for other regions is ignored) Args: - project_path: Absolute path to the project root directory (required) - - name: Application name (required for NEW apps only) + - name: Application name (required for NEW apps, will auto-generate from folder name if not provided) - application_id: Application ID for updates (auto-read from config if exists) - region: Region (cn-beijing, cn-shanghai, cn-guangzhou, ap-southeast-1) - - build_command: Build command (e.g., "npm run build"). Required for non-Python runtimes unless skip_build=True. + - build_command: Build command (e.g., "npm run build"). Required for non-Python runtimes. - start_command: Startup command. **REQUIRED**. Use detect_project to auto-detect. - port: Service port. **IMPORTANT: Must match the actual listening port in start_command** (e.g., if start_command has --port 3000, then port must be 3000) - - skip_build: Skip build step (default False). Set to True if project is already built. + - skip_build: Skip build step (default False). **Note: Python projects automatically skip build, this parameter is ignored for Python.** - gateway_name: API gateway name (optional, auto-selects first available gateway if not specified) Returns: @@ -1163,8 +1167,10 @@ def detect_project(project_path: str): **Common Errors**: - "start_command is required": Call `detect_project` first or provide start_command -- "build_command is required": Provide build_command or set skip_build=True if already built -- "Name already exists": Use the `application_id` from `.vefaas/config.json` or console, then retry with `application_id` parameter +- "build_command is required": For non-Python projects, provide build_command or set skip_build=True +- "Name already exists (NAME_CONFLICT)": **YOU MAY ASK THE USER** whether to update existing or create new. Present these options: + 1. Update existing: retry with `application_id` parameter (ID is in error message) + 2. Create new: retry with a different `name` parameter - "deploy_fail": The returned error will include detailed error_message and error_logs_uri **Retry/Redeployment**: @@ -1415,7 +1421,7 @@ def vefaas_deploy_guide(): The tool will auto-read config and update the application ### Common Issues -- **Name exists**: Use get_application_detail to get ID, then retry with application_id parameter +- **NAME_CONFLICT (Name exists)**: **Ask the user** to choose: 1) Update existing (use application_id) or 2) Create new (use different name) - **start_command error**: Use detect_project to get correct command - **Deployment failed**: Check error_details for error message and logs diff --git a/server/mcp_server_vefaas_function/tests/__init__.py b/server/mcp_server_vefaas_function/tests/__init__.py new file mode 100644 index 00000000..65140f2e --- /dev/null +++ b/server/mcp_server_vefaas_function/tests/__init__.py @@ -0,0 +1 @@ +# tests package diff --git a/server/mcp_server_vefaas_function/tests/test_e2e.py b/server/mcp_server_vefaas_function/tests/test_e2e.py new file mode 100644 index 00000000..d0802390 --- /dev/null +++ b/server/mcp_server_vefaas_function/tests/test_e2e.py @@ -0,0 +1,507 @@ +#!/usr/bin/env python3 +""" +veFaaS MCP Server Test Script + +This script tests the full workflow of veFaaS function and application deployment. + +Test Scenarios: +1. Application Deployment (Python/Node.js) + - Deploy new application + - Update and redeploy + +Usage: + # Run all tests + python test_e2e.py + + # Run specific scenario + python test_e2e.py --scenario app-python + python test_e2e.py --scenario app-node + +Environment Variables: + VOLCENGINE_ACCESS_KEY_ID: Access Key ID (required) + VOLCENGINE_SECRET_ACCESS_KEY: Secret Access Key (required) + VOLCENGINE_SECURITY_TOKEN: Security Token (optional, for STS) + TEST_REGION: Region to test (default: cn-beijing) +""" + +from mcp_server_vefaas_function.vefaas_cli_sdk import ( + VeFaaSClient, + DeployConfig, + deploy_application, + auto_detect, +) +import os +import sys +import json +import time +import shutil +import tempfile +import argparse +import logging + +# Add src to path for imports +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) + + +# ========== Test Project Templates ========== + +PYTHON_PROJECT = { + "app.py": '''from fastapi import FastAPI + +app = FastAPI() + +@app.get("/") +def root(): + return {"message": "Hello from Python!"} + +@app.get("/health") +def health(): + return {"status": "ok"} +''', + "requirements.txt": "fastapi\nuvicorn\n", + "run.sh": '''#!/bin/bash +exec python -m uvicorn app:app --host 0.0.0.0 --port 8000 +''', +} + +PYTHON_PROJECT_UPDATED = { + "app.py": '''from fastapi import FastAPI + +app = FastAPI() + +@app.get("/") +def root(): + return {"message": "Hello from Python! (Updated)"} + +@app.get("/health") +def health(): + return {"status": "ok", "version": "2.0"} +''', + "requirements.txt": "fastapi\nuvicorn\n", + "run.sh": '''#!/bin/bash +exec python -m uvicorn app:app --host 0.0.0.0 --port 8000 +''', +} + +NODE_PROJECT = { + "index.js": '''const http = require('http'); + +const server = http.createServer((req, res) => { + if (req.url === '/health') { + res.writeHead(200, {'Content-Type': 'application/json'}); + res.end(JSON.stringify({status: 'ok'})); + } else { + res.writeHead(200, {'Content-Type': 'application/json'}); + res.end(JSON.stringify({message: 'Hello from Node.js!'})); + } +}); + +server.listen(8000, '0.0.0.0', () => { + console.log('Server running on port 8000'); +}); +''', + "package.json": '''{ + "name": "test-node-app", + "version": "1.0.0", + "main": "index.js", + "scripts": { + "start": "node index.js" + } +} +''', +} + +NODE_PROJECT_UPDATED = { + "index.js": '''const http = require('http'); + +const server = http.createServer((req, res) => { + if (req.url === '/health') { + res.writeHead(200, {'Content-Type': 'application/json'}); + res.end(JSON.stringify({status: 'ok', version: '2.0'})); + } else { + res.writeHead(200, {'Content-Type': 'application/json'}); + res.end(JSON.stringify({message: 'Hello from Node.js! (Updated)'})); + } +}); + +server.listen(8000, '0.0.0.0', () => { + console.log('Server running on port 8000'); +}); +''', + "package.json": '''{ + "name": "test-node-app", + "version": "2.0.0", + "main": "index.js", + "scripts": { + "start": "node index.js" + } +} +''', +} + + +# ========== Helper Functions ========== + +def get_credentials(): + """Get credentials from environment variables""" + ak = os.environ.get("VOLCENGINE_ACCESS_KEY_ID") + sk = os.environ.get("VOLCENGINE_SECRET_ACCESS_KEY") + token = os.environ.get("VOLCENGINE_SECURITY_TOKEN") + + if not ak or not sk: + raise ValueError( + "Missing credentials. Set VOLCENGINE_ACCESS_KEY_ID and VOLCENGINE_SECRET_ACCESS_KEY" + ) + + return ak, sk, token + + +def create_temp_project(files: dict, name: str) -> str: + """Create a temporary project directory with given files""" + temp_dir = tempfile.mkdtemp(prefix=f"vefaas_test_{name}_") + + for filename, content in files.items(): + filepath = os.path.join(temp_dir, filename) + with open(filepath, "w") as f: + f.write(content) + + # Make .sh files executable + if filename.endswith(".sh"): + os.chmod(filepath, 0o755) + + logger.info(f"Created temp project: {temp_dir}") + return temp_dir + + +def update_project(project_path: str, files: dict): + """Update project files""" + for filename, content in files.items(): + filepath = os.path.join(project_path, filename) + with open(filepath, "w") as f: + f.write(content) + logger.info(f"Updated project files in: {project_path}") + + +def cleanup_project(project_path: str): + """Remove temporary project directory""" + if project_path and os.path.exists(project_path): + shutil.rmtree(project_path) + logger.info(f"Cleaned up: {project_path}") + + +def generate_unique_name(prefix: str) -> str: + """Generate unique name for testing""" + import random + import string + suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6)) + return f"{prefix}-{suffix}" + + +# ========== Test Scenarios ========== + +class TestResult: + def __init__(self, name: str): + self.name = name + self.steps = [] + self.success = True + self.error = None + self.function_id = None + self.application_id = None + self.access_url = None + + def add_step(self, step: str, success: bool, details: str = ""): + self.steps.append({ + "step": step, + "success": success, + "details": details + }) + if not success: + self.success = False + + def summary(self) -> str: + lines = [f"\n{'='*60}", f"Test: {self.name}", f"Result: {'✅ PASSED' if self.success else '❌ FAILED'}", ""] + + for step in self.steps: + icon = "✅" if step["success"] else "❌" + lines.append(f" {icon} {step['step']}") + if step["details"]: + lines.append(f" {step['details']}") + + if self.function_id: + lines.append(f"\n Function ID: {self.function_id}") + if self.application_id: + lines.append(f" Application ID: {self.application_id}") + if self.access_url: + lines.append(f" Access URL: {self.access_url}") + + lines.append("=" * 60) + return "\n".join(lines) + + +def test_application_workflow(runtime: str, region: str) -> TestResult: + """ + Test application deployment workflow using deploy_application: + 1. Deploy new application + 2. Update and redeploy + """ + result = TestResult(f"Application Workflow ({runtime})") + project_path = None + + try: + ak, sk, token = get_credentials() + client = VeFaaSClient(ak, sk, token, region) + + # Select project template + if runtime == "python": + initial_files = PYTHON_PROJECT + updated_files = PYTHON_PROJECT_UPDATED + start_command = "./run.sh" + else: + initial_files = NODE_PROJECT + updated_files = NODE_PROJECT_UPDATED + start_command = "node index.js" + + # Step 1: Create temp project + project_path = create_temp_project(initial_files, runtime) + result.add_step("Create temp project", True, project_path) + + # Step 2: Detect project + detection = auto_detect(project_path) + result.add_step("Detect project", True, f"Framework: {detection.framework}, Runtime: {detection.runtime}") + + # Step 3: Deploy application (uses deploy_application from SDK) + app_name = generate_unique_name(f"app-{runtime}") + logger.info(f"Deploying application: {app_name}") + + config = DeployConfig( + project_path=project_path, + name=app_name, + start_command=start_command, + port=8000, + skip_build=True, + ) + + deploy_result = deploy_application(config, client) + + if not deploy_result.success: + raise ValueError(f"Deployment failed: {deploy_result.error}") + + result.function_id = deploy_result.function_id + result.application_id = deploy_result.application_id + result.access_url = deploy_result.access_url + result.add_step("Deploy application (v1)", True, f"URL: {deploy_result.access_url}") + + # Step 4: Update project files + update_project(project_path, updated_files) + result.add_step("Update project files", True) + + # Step 5: Redeploy application (uses deploy_application with application_id) + config2 = DeployConfig( + project_path=project_path, + application_id=deploy_result.application_id, + start_command=start_command, + port=8000, + skip_build=True, + ) + + deploy_result2 = deploy_application(config2, client) + + if not deploy_result2.success: + raise ValueError(f"Redeployment failed: {deploy_result2.error}") + + result.add_step("Redeploy application (v2)", True) + + logger.info(result.summary()) + + except Exception as e: + result.error = str(e) + result.add_step("Test execution", False, str(e)) + logger.error(f"Test failed: {e}") + + finally: + if project_path: + cleanup_project(project_path) + + return result + + +def test_function_local_dev_workflow(runtime: str, region: str) -> TestResult: + """ + Test function local development workflow using MCP tools: + 1. Deploy initial application (to get a function) + 2. Update function code (simulate local development) + 3. Release function + + This simulates the scenario where a developer creates a function, + then makes local code changes and uses update_function + release_function + to iterate without going through full deploy_application flow. + """ + result = TestResult(f"Function Local Dev ({runtime})") + project_path = None + + # Import the MCP tool functions + from mcp_server_vefaas_function.vefaas_server import ( + zip_and_encode_folder, + ) + from mcp_server_vefaas_function.vefaas_cli_sdk import ( + wait_for_function_release, + wait_for_dependency_install, + ) + + try: + ak, sk, token = get_credentials() + client = VeFaaSClient(ak, sk, token, region) + + # Select project template + if runtime == "python": + initial_files = PYTHON_PROJECT + updated_files = PYTHON_PROJECT_UPDATED + start_command = "./run.sh" + else: + initial_files = NODE_PROJECT + updated_files = NODE_PROJECT_UPDATED + start_command = "node index.js" + + # Step 1: Create temp project + project_path = create_temp_project(initial_files, runtime) + result.add_step("Create temp project", True, project_path) + + # Step 2: Create initial function via deploy_application + func_name = generate_unique_name(f"func-{runtime}") + logger.info(f"Creating function via deploy_application: {func_name}") + + config = DeployConfig( + project_path=project_path, + name=func_name, + start_command=start_command, + port=8000, + skip_build=True, + ) + deploy_result = deploy_application(config, client) + + if not deploy_result.success: + raise ValueError(f"Initial deployment failed: {deploy_result.error}") + + function_id = deploy_result.function_id + result.function_id = function_id + result.application_id = deploy_result.application_id + result.access_url = deploy_result.access_url + result.add_step("Create function (initial deploy)", True, f"Function ID: {function_id}") + + # Step 3: Update project files (simulate local development) + update_project(project_path, updated_files) + result.add_step("Update local files", True, "Simulating code changes") + + # Step 4: Upload new code using update_function pattern + # (Similar to what MCP update_function tool does internally) + logger.info(f"Updating function code: {function_id}") + + # Package and upload code + zip_bytes, size, err = zip_and_encode_folder(project_path) + if err: + raise ValueError(f"Failed to package code: {err}") + + source_location = client.upload_to_tos(zip_bytes) + client.update_function(function_id, source=source_location) + result.add_step("Update function code", True, f"Uploaded {size} bytes") + + # Step 5: Install dependencies (Python only) + if runtime == "python": + try: + client.create_dependency_install_task(function_id) + wait_for_dependency_install(client, function_id, timeout_seconds=300) + result.add_step("Install dependencies", True) + except Exception as e: + result.add_step("Install dependencies", False, str(e)) + + # Step 6: Release function + logger.info(f"Releasing function: {function_id}") + try: + client.release_function(function_id) + wait_for_function_release(client, function_id, timeout_seconds=180) + result.add_step("Release function", True) + except Exception as e: + result.add_step("Release function", False, str(e)) + + logger.info(result.summary()) + + except Exception as e: + result.error = str(e) + result.add_step("Test execution", False, str(e)) + logger.error(f"Test failed: {e}") + + finally: + if project_path: + cleanup_project(project_path) + + return result + + +# ========== Main ========== + +def main(): + parser = argparse.ArgumentParser(description="veFaaS MCP Server E2E Tests") + parser.add_argument( + "--scenario", + choices=["all", "app-python", "app-node", "func-python", "func-node"], + default="all", + help="Test scenario to run" + ) + parser.add_argument( + "--region", + default=os.environ.get("TEST_REGION", "cn-beijing"), + help="Region to test" + ) + + args = parser.parse_args() + + print("\n" + "=" * 60) + print("veFaaS MCP Server E2E Tests") + print(f"Region: {args.region}") + print(f"Scenario: {args.scenario}") + print("=" * 60 + "\n") + + results = [] + + # Application deployment tests + if args.scenario in ("all", "app-python"): + results.append(test_application_workflow("python", args.region)) + + if args.scenario in ("all", "app-node"): + results.append(test_application_workflow("node", args.region)) + + # Function local development tests + if args.scenario in ("all", "func-python"): + results.append(test_function_local_dev_workflow("python", args.region)) + + if args.scenario in ("all", "func-node"): + results.append(test_function_local_dev_workflow("node", args.region)) + + # Print summary + print("\n" + "=" * 60) + print("TEST SUMMARY") + print("=" * 60) + + passed = sum(1 for r in results if r.success) + failed = len(results) - passed + + for r in results: + icon = "✅" if r.success else "❌" + print(f" {icon} {r.name}") + + print(f"\nTotal: {len(results)} | Passed: {passed} | Failed: {failed}") + print("=" * 60 + "\n") + + return 0 if failed == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_server_test.py b/server/mcp_server_vefaas_function/tests/test_unit.py similarity index 66% rename from server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_server_test.py rename to server/mcp_server_vefaas_function/tests/test_unit.py index ba51672f..303de5ba 100644 --- a/server/mcp_server_vefaas_function/src/mcp_server_vefaas_function/vefaas_server_test.py +++ b/server/mcp_server_vefaas_function/tests/test_unit.py @@ -233,6 +233,127 @@ def test_detect_vite_project_is_static(self): self.assertTrue(result.is_static) self.assertIn("caddy", result.start_command.lower()) + def test_detect_fastapi_project(self): + """Test that FastAPI project is detected correctly""" + from mcp_server_vefaas_function.vefaas_cli_sdk.detector import auto_detect + + # Create FastAPI project structure + with open(os.path.join(self.temp_dir, "requirements.txt"), "w") as f: + f.write("fastapi\nuvicorn\n") + with open(os.path.join(self.temp_dir, "app.py"), "w") as f: + f.write("from fastapi import FastAPI\napp = FastAPI()\n") + + result = auto_detect(self.temp_dir) + + self.assertEqual(result.framework, "fastapi") + self.assertEqual(result.runtime, "native-python3.12/v1") + self.assertFalse(result.is_static) + self.assertIn("uvicorn", result.start_command) + + def test_detect_flask_project(self): + """Test that Flask project is detected correctly""" + from mcp_server_vefaas_function.vefaas_cli_sdk.detector import auto_detect + + # Create Flask project structure + with open(os.path.join(self.temp_dir, "requirements.txt"), "w") as f: + f.write("flask\ngunicorn\n") + with open(os.path.join(self.temp_dir, "app.py"), "w") as f: + f.write("from flask import Flask\napp = Flask(__name__)\n") + + result = auto_detect(self.temp_dir) + + self.assertEqual(result.framework, "flask") + self.assertEqual(result.runtime, "native-python3.12/v1") + self.assertFalse(result.is_static) + + +class TestConfig(unittest.TestCase): + """Test cases for configuration reading/writing""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.temp_dir) + + def test_write_and_read_config(self): + """Test writing and reading configuration""" + from mcp_server_vefaas_function.vefaas_cli_sdk.config import ( + write_config, read_config, VefaasConfig, FunctionConfig + ) + + config = VefaasConfig( + function=FunctionConfig( + id="test-func-id", + region="cn-beijing", + runtime="native-python3.12/v1", + application_id="test-app-id", + ), + name="test-app", + command="./run.sh", + ) + + # Write config + write_config(self.temp_dir, config) + + # Verify files exist + self.assertTrue(os.path.exists(os.path.join(self.temp_dir, ".vefaas", "config.json"))) + self.assertTrue(os.path.exists(os.path.join(self.temp_dir, "vefaas.yaml"))) + + # Read config back + loaded_config = read_config(self.temp_dir) + self.assertIsNotNone(loaded_config) + self.assertEqual(loaded_config.function.id, "test-func-id") + self.assertEqual(loaded_config.function.region, "cn-beijing") + self.assertEqual(loaded_config.function.application_id, "test-app-id") + + def test_get_linked_ids(self): + """Test getting linked IDs from config""" + from mcp_server_vefaas_function.vefaas_cli_sdk.config import ( + write_config, get_linked_ids, VefaasConfig, FunctionConfig + ) + + config = VefaasConfig( + function=FunctionConfig( + id="func-123", + application_id="app-456", + ), + ) + write_config(self.temp_dir, config) + + func_id, app_id = get_linked_ids(self.temp_dir) + self.assertEqual(func_id, "func-123") + self.assertEqual(app_id, "app-456") + + +class TestGenerateAppName(unittest.TestCase): + """Test cases for app name generation""" + + def test_generate_app_name_from_path(self): + """Test generating app name from project path""" + from mcp_server_vefaas_function.vefaas_cli_sdk.deploy import generate_app_name_from_path + + # Test with simple path + name = generate_app_name_from_path("/path/to/my-project") + self.assertTrue(name.startswith("my-project-")) + self.assertEqual(len(name), len("my-project-") + 6) # 6 char suffix + + def test_generate_app_name_handles_special_chars(self): + """Test that special characters are replaced""" + from mcp_server_vefaas_function.vefaas_cli_sdk.deploy import generate_app_name_from_path + + name = generate_app_name_from_path("/path/to/My_Project.Name") + # Should be lowercase with hyphens + self.assertTrue(name.startswith("my-project-name-")) + + def test_generate_app_name_truncates_long_names(self): + """Test that long names are truncated""" + from mcp_server_vefaas_function.vefaas_cli_sdk.deploy import generate_app_name_from_path + + name = generate_app_name_from_path("/path/to/this-is-a-very-long-project-name-that-should-be-truncated") + # Base name should be max 20 chars + hyphen + 6 char suffix + self.assertLessEqual(len(name), 20 + 1 + 6) + if __name__ == "__main__": unittest.main()