tests/gappsscript/manual_test.py

"""
Manual E2E test script for Apps Script integration.

This script tests Apps Script tools against the real Google API.
Requires valid OAuth credentials and enabled Apps Script API.

Usage:
    python tests/gappsscript/manual_test.py

Environment Variables:
    GOOGLE_CLIENT_SECRET_PATH: Path to client_secret.json (default: ./client_secret.json)
    GOOGLE_TOKEN_PATH: Path to store OAuth token (default: ./test_token.pickle)

Note: This will create real Apps Script projects in your account.
      Delete test projects manually after running.
"""

import asyncio
import sys
import os

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))

from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import pickle


SCOPES = [
    "https://www.googleapis.com/auth/script.projects",
    "https://www.googleapis.com/auth/script.deployments",
    "https://www.googleapis.com/auth/script.processes",
    "https://www.googleapis.com/auth/drive.readonly",  # For listing script projects
    "https://www.googleapis.com/auth/userinfo.email",  # Basic user info
    "openid",  # Required by Google OAuth
]

# Allow http://localhost for OAuth (required for headless auth)
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"

# Default paths (can be overridden via environment variables)
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
DEFAULT_CLIENT_SECRET = os.path.join(PROJECT_ROOT, "client_secret.json")
DEFAULT_TOKEN_PATH = os.path.join(PROJECT_ROOT, "test_token.pickle")


def get_credentials():
    """
    Get OAuth credentials for Apps Script API.

    Credential paths can be configured via environment variables:
    - GOOGLE_CLIENT_SECRET_PATH: Path to client_secret.json
    - GOOGLE_TOKEN_PATH: Path to store/load OAuth token

    Returns:
        Credentials object
    """
    creds = None
    token_path = os.environ.get("GOOGLE_TOKEN_PATH", DEFAULT_TOKEN_PATH)
    client_secret_path = os.environ.get(
        "GOOGLE_CLIENT_SECRET_PATH", DEFAULT_CLIENT_SECRET
    )

    if os.path.exists(token_path):
        with open(token_path, "rb") as token:
            creds = pickle.load(token)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            if not os.path.exists(client_secret_path):
                print(f"Error: {client_secret_path} not found")
                print("\nTo fix this:")
                print("1. Go to Google Cloud Console > APIs & Services > Credentials")
                print("2. Create an OAuth 2.0 Client ID (Desktop application type)")
                print("3. Download the JSON and save as client_secret.json")
                print(f"\nExpected path: {client_secret_path}")
                print("\nOr set GOOGLE_CLIENT_SECRET_PATH environment variable")
                sys.exit(1)

            flow = InstalledAppFlow.from_client_secrets_file(client_secret_path, SCOPES)
            # Set redirect URI to match client_secret.json
            flow.redirect_uri = "http://localhost"
            # Headless flow: user copies redirect URL after auth
            auth_url, _ = flow.authorization_url(prompt="consent")
            print("\n" + "=" * 60)
            print("HEADLESS AUTH")
            print("=" * 60)
            print("\n1. Open this URL in any browser:\n")
            print(auth_url)
            print("\n2. Sign in and authorize the app")
            print("3. You'll be redirected to http://localhost (won't load)")
            print("4. Copy the FULL URL from browser address bar")
            print("   (looks like: http://localhost/?code=4/0A...&scope=...)")
            print("5. Paste it below:\n")
            redirect_response = input("Paste full redirect URL: ").strip()
            flow.fetch_token(authorization_response=redirect_response)
            creds = flow.credentials

        with open(token_path, "wb") as token:
            pickle.dump(creds, token)

    return creds


async def test_list_projects(drive_service):
    """Test listing Apps Script projects using Drive API"""
    print("\n=== Test: List Projects ===")

    from gappsscript.apps_script_tools import _list_script_projects_impl

    try:
        result = await _list_script_projects_impl(
            service=drive_service, user_google_email="test@example.com", page_size=10
        )
        print(result)
        return True
    except Exception as e:
        print(f"Error: {e}")
        return False


async def test_create_project(service):
    """Test creating a new Apps Script project"""
    print("\n=== Test: Create Project ===")

    from gappsscript.apps_script_tools import _create_script_project_impl

    try:
        result = await _create_script_project_impl(
            service=service,
            user_google_email="test@example.com",
            title="MCP Test Project",
        )
        print(result)

        if "Script ID:" in result:
            script_id = result.split("Script ID: ")[1].split("\n")[0]
            return script_id
        return None
    except Exception as e:
        print(f"Error: {e}")
        return None


async def test_get_project(service, script_id):
    """Test retrieving project details"""
    print(f"\n=== Test: Get Project {script_id} ===")

    from gappsscript.apps_script_tools import _get_script_project_impl

    try:
        result = await _get_script_project_impl(
            service=service, user_google_email="test@example.com", script_id=script_id
        )
        print(result)
        return True
    except Exception as e:
        print(f"Error: {e}")
        return False


async def test_update_content(service, script_id):
    """Test updating script content"""
    print(f"\n=== Test: Update Content {script_id} ===")

    from gappsscript.apps_script_tools import _update_script_content_impl

    files = [
        {
            "name": "appsscript",
            "type": "JSON",
            "source": """{
  "timeZone": "America/New_York",
  "dependencies": {},
  "exceptionLogging": "STACKDRIVER",
  "runtimeVersion": "V8"
}""",
        },
        {
            "name": "Code",
            "type": "SERVER_JS",
            "source": """function testFunction() {
  Logger.log('Hello from MCP test!');
  return 'Test successful';
}""",
        },
    ]

    try:
        result = await _update_script_content_impl(
            service=service,
            user_google_email="test@example.com",
            script_id=script_id,
            files=files,
        )
        print(result)
        return True
    except Exception as e:
        print(f"Error: {e}")
        return False


async def test_run_function(service, script_id):
    """Test running a script function"""
    print(f"\n=== Test: Run Function {script_id} ===")

    from gappsscript.apps_script_tools import _run_script_function_impl

    try:
        result = await _run_script_function_impl(
            service=service,
            user_google_email="test@example.com",
            script_id=script_id,
            function_name="testFunction",
            dev_mode=True,
        )
        print(result)
        return True
    except Exception as e:
        print(f"Error: {e}")
        return False


async def test_create_deployment(service, script_id):
    """Test creating a deployment"""
    print(f"\n=== Test: Create Deployment {script_id} ===")

    from gappsscript.apps_script_tools import _create_deployment_impl

    try:
        result = await _create_deployment_impl(
            service=service,
            user_google_email="test@example.com",
            script_id=script_id,
            description="MCP Test Deployment",
        )
        print(result)

        if "Deployment ID:" in result:
            deployment_id = result.split("Deployment ID: ")[1].split("\n")[0]
            return deployment_id
        return None
    except Exception as e:
        print(f"Error: {e}")
        return None


async def test_list_deployments(service, script_id):
    """Test listing deployments"""
    print(f"\n=== Test: List Deployments {script_id} ===")

    from gappsscript.apps_script_tools import _list_deployments_impl

    try:
        result = await _list_deployments_impl(
            service=service, user_google_email="test@example.com", script_id=script_id
        )
        print(result)
        return True
    except Exception as e:
        print(f"Error: {e}")
        return False


async def test_list_processes(service):
    """Test listing script processes"""
    print("\n=== Test: List Processes ===")

    from gappsscript.apps_script_tools import _list_script_processes_impl

    try:
        result = await _list_script_processes_impl(
            service=service, user_google_email="test@example.com", page_size=10
        )
        print(result)
        return True
    except Exception as e:
        print(f"Error: {e}")
        return False


async def cleanup_test_project(service, script_id):
    """
    Cleanup test project (requires Drive API).
    Note: Apps Script API does not have a delete endpoint.
    Projects must be deleted via Drive API by moving to trash.
    """
    print(f"\n=== Cleanup: Delete Project {script_id} ===")
    print("Note: Apps Script projects must be deleted via Drive API")
    print(f"Please manually delete: https://script.google.com/d/{script_id}/edit")


async def run_all_tests():
    """Run all manual tests"""
    print("=" * 60)
    print("Apps Script MCP Manual Test Suite")
    print("=" * 60)

    print("\nGetting OAuth credentials...")
    creds = get_credentials()

    print("Building API services...")
    script_service = build("script", "v1", credentials=creds)
    drive_service = build("drive", "v3", credentials=creds)

    test_script_id = None
    deployment_id = None

    try:
        success = await test_list_projects(drive_service)
        if not success:
            print("\nWarning: List projects failed")

        test_script_id = await test_create_project(script_service)
        if test_script_id:
            print(f"\nCreated test project: {test_script_id}")

            await test_get_project(script_service, test_script_id)
            await test_update_content(script_service, test_script_id)

            await asyncio.sleep(2)

            await test_run_function(script_service, test_script_id)

            deployment_id = await test_create_deployment(script_service, test_script_id)
            if deployment_id:
                print(f"\nCreated deployment: {deployment_id}")

            await test_list_deployments(script_service, test_script_id)
        else:
            print("\nSkipping tests that require a project (creation failed)")

        await test_list_processes(script_service)

    finally:
        if test_script_id:
            await cleanup_test_project(script_service, test_script_id)

    print("\n" + "=" * 60)
    print("Manual Test Suite Complete")
    print("=" * 60)


def main():
    """Main entry point"""
    import argparse

    parser = argparse.ArgumentParser(description="Manual E2E test for Apps Script")
    parser.add_argument(
        "--yes", "-y", action="store_true", help="Skip confirmation prompt"
    )
    args = parser.parse_args()

    print("\nIMPORTANT: This script will:")
    print("1. Create a test Apps Script project in your account")
    print("2. Run various operations on it")
    print("3. Leave the project for manual cleanup")
    print("\nYou must manually delete the test project after running this.")

    if not args.yes:
        response = input("\nContinue? (yes/no): ")
        if response.lower() not in ["yes", "y"]:
            print("Aborted")
            return

    asyncio.run(run_all_tests())


if __name__ == "__main__":
    main()