sugar-crm/test_api.py

#!/usr/bin/env python3
"""
SugarCRM 6.5 CE REST v4.1 API Test
===================================
Endpunkt: http://localhost:{PORT}/service/v4_1/rest.php
"""
import http.client
import json
import hashlib
import urllib.parse
import sys
import os
from datetime import datetime

# Configurable via env vars
BASE_HOST = os.environ.get("SUGARCRM_HOST", "localhost")
BASE_PORT = os.environ.get("SUGARCRM_PORT", "2080")
ENDPOINT = "/service/v4_1/rest.php"
USER = os.environ.get("SUGARCRM_USER", "admin")
PASSWORD = os.environ.get("SUGARCRM_PASSWORD", "admin123")

BASE_URL = f"{BASE_HOST}:{BASE_PORT}"

def call_api(method, rest_data):
    """Aufruf der SugarCRM REST v4.1 API"""
    conn = http.client.HTTPConnection(BASE_URL, timeout=30)

    body = urllib.parse.urlencode({
        "method": method,
        "input_type": "JSON",
        "response_type": "JSON",
        "rest_data": json.dumps(rest_data)
    })

    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    conn.request("POST", ENDPOINT, body, headers)
    resp = conn.getresponse()

    if resp.status == 302:
        location = resp.getheader("Location", "")
        print(f"  ⚠️  Redirect to: {location}")
        print("  (Admin Wizard may be active - restart container to fix)")
        return None

    data = json.loads(resp.read().decode())
    conn.close()
    return data


def main():
    print("=" * 60)
    print("🍬 SugarCRM 6.5.26 CE REST API Test")
    print(f"   URL: http://{BASE_URL}")
    print("=" * 60)

    # 1. Login
    pwd_hash = hashlib.md5(PASSWORD.encode()).hexdigest()
    print(f"\n1️⃣  LOGIN (user={USER}, md5={pwd_hash[:8]}...)")

    result = call_api("login", {
        "user_auth": {"user_name": USER, "password": pwd_hash},
        "application_name": "API Test Script"
    })

    if not result:
        print("❌ LOGIN FAILED (Redirect - Admin Wizard active?)")
        print("   Run: docker compose restart sugarcrm")
        sys.exit(1)

    if result.get("id"):
        session = result["id"]
        print(f"   ✅ Session: {session[:20]}...")
    else:
        print(f"   ❌ Login Error: {result.get('name', 'Unknown')} - {result.get('description', '')}")
        sys.exit(1)

    # 2. Available modules
    print(f"\n2️⃣  AVAILABLE MODULES")
    result = call_api("get_available_modules", {"session": session})
    if result and "modules" in result:
        modules = result["modules"]
        print(f"   ✅ {len(modules)} modules available")
        print(f"   First 10: {', '.join(m['module_key'] for m in modules[:10])}")
    else:
        print(f"   ❌ Error: {result}")

    # 3. Accounts list
    print(f"\n3️⃣  ACCOUNTS (get_entry_list)")
    result = call_api("get_entry_list", {
        "session": session,
        "module_name": "Accounts",
        "query": "",
        "order_by": "",
        "offset": 0,
        "select_fields": ["name", "id", "date_entered"],
        "link_name_to_fields_array": [],
        "max_results": 5,
        "deleted": 0
    })
    if result and "entry_list" in result:
        count = result.get("result_count", 0)
        print(f"   ✅ {count} accounts found")
        for entry in result["entry_list"][:5]:
            vals = {n["name"]: n["value"] for n in entry["name_value_list"]}
            print(f"      - {vals.get('name', 'N/A')} (ID: {vals.get('id', 'N/A')[:10]}...)")
    else:
        print(f"   ❌ Error: {result}")

    # 4. Create test account
    print(f"\n4️⃣  CREATE ACCOUNT (set_entry)")
    test_name = f"Test API Account {datetime.now().strftime('%H:%M:%S')}"
    result = call_api("set_entry", {
        "session": session,
        "module_name": "Accounts",
        "name_value_list": {
            "name": {"name": "name", "value": test_name},
            "account_type": {"name": "account_type", "value": "Customer"},
        }
    })
    if result and result.get("id"):
        print(f"   ✅ Account '{test_name}' created (ID: {result['id'][:10]}...)")

    # 5. Session count
    print(f"\n5️⃣  MODULE COUNT (get_entries_count)")
    result = call_api("get_entries_count", {
        "session": session,
        "module_name": "Accounts",
        "query": "",
        "deleted": 0
    })
    if result:
        print(f"   ✅ Total accounts: {result.get('result_count', 'unknown')}")

    print(f"\n{'=' * 60}")
    print("✅ ALL API TESTS PASSED!")
    print(f"{'=' * 60}")


if __name__ == "__main__":
    main()