sugar-crm/test_api_extended.py

#!/usr/bin/env python3
"""
SugarCRM 6.5 CE — Erweiterte API-Tests
========================================
Testet alle wichtigen REST v4.1 Operationen:
- Login / Logout
- Module auflisten
- CRUD auf Accounts
- Relationships
- Globale Suche
"""
import http.client
import json
import hashlib
import urllib.parse
import sys
import os
from datetime import datetime

BASE_HOST = os.environ.get("SUGARCRM_HOST", "localhost")
BASE_PORT = os.environ.get("SUGARCRM_PORT", "2080")
ENDPOINT = "/service/v4_1/rest.php"
BASE_URL = f"{BASE_HOST}:{BASE_PORT}"

USER = os.environ.get("SUGARCRM_USER", "admin")
PASSWORD = os.environ.get("SUGARCRM_PASSWORD", "admin123")
PWD_HASH = hashlib.md5(PASSWORD.encode()).hexdigest()

failures = 0
test_account_id = None
session = None


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"}
    try:
        conn.request("POST", ENDPOINT, body, headers)
        resp = conn.getresponse()
        if resp.status == 302:
            return {"_error": f"Redirect (AdminWizard active?)"}
        data = json.loads(resp.read().decode())
        conn.close()
        return data
    except Exception as e:
        return {"_error": str(e)}


def test(name, fn):
    global failures
    try:
        result = fn()
        if result:
            print(f"   ✅ {name}")
            return result
        else:
            print(f"   ❌ {name}")
            failures += 1
            return None
    except Exception as e:
        print(f"   ❌ {name}: {e}")
        failures += 1
        return None


def main():
    global session, test_account_id

    print("=" * 60)
    print("🍬 SugarCRM 6.5.26 CE — Erweiterte API-Tests")
    print(f"   URL: http://{BASE_URL}")
    print("=" * 60)

    # === AUTH ===
    print("\n📌 AUTHENTIFIZIERUNG")

    def do_login():
        global session
        result = call_api("login", {
            "user_auth": {"user_name": USER, "password": PWD_HASH},
            "application_name": "Extended Test Suite"
        })
        if result.get("_error"):
            return False
        if result.get("id"):
            session = result["id"]
            return True
        return False

    if not test("Login", do_login):
        print("\n❌ Login fehlgeschlagen. Tests abgebrochen.")
        print("   Prüfe: Läuft der Container? Admin Wizard deaktiviert?")
        sys.exit(1)

    # === MODULES ===
    print("\n📌 MODULE")

    def get_modules():
        result = call_api("get_available_modules", {"session": session})
        if "_error" in result:
            return False
        if result and "modules" in result:
            count = len(result["modules"])
            # Zeige alle Module
            names = [m["module_key"] for m in result["modules"]]
            print(f"      Verfügbar ({count}): {', '.join(names)}")
            return True
        return False

    test("Verfügbare Module", get_modules)

    def get_module_fields():
        result = call_api("get_module_fields", {
            "session": session,
            "module_name": "Accounts",
            "fields": []
        })
        if "_error" in result:
            return False
        if result and "module_fields" in result:
            fields = [f["name"] for f in result["module_fields"].values()]
            print(f"      Accounts Felder ({len(fields)}): {', '.join(fields)}")
            return True
        return False

    test("Accounts Felder", get_module_fields)

    # === CRUD: Accounts ===
    print("\n📌 CRUD: ACCOUNTS")

    def create_account():
        global test_account_id
        name = f"API Test {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        result = call_api("set_entry", {
            "session": session,
            "module_name": "Accounts",
            "name_value_list": {
                "name": {"name": "name", "value": name},
                "account_type": {"name": "account_type", "value": "Customer"},
                "industry": {"name": "industry", "value": "Technology"},
                "phone_office": {"name": "phone_office", "value": "+49 30 1234567"},
                "description": {"name": "description", "value": "API erstellter Test-Account"},
            }
        })
        if result.get("id"):
            test_account_id = result["id"]
            print(f"      Name: '{name}', ID: {test_account_id[:10]}...")
            return True
        return False

    test("Account erstellen (set_entry)", create_account)

    def read_account():
        if not test_account_id:
            return False
        result = call_api("get_entry", {
            "session": session,
            "module_name": "Accounts",
            "id": test_account_id,
            "select_fields": ["name", "account_type", "industry", "phone_office", "description"]
        })
        if "_error" in result:
            return False
        if result and "entry_list" in result:
            vals = {n["name"]: n["value"] for n in result["entry_list"][0]["name_value_list"]}
            print(f"      Name: {vals.get('name')}, Typ: {vals.get('account_type')}, "
                  f"Tel: {vals.get('phone_office')}, Branche: {vals.get('industry')}")
            return True
        return False

    test("Account lesen (get_entry)", read_account)

    def update_account():
        if not test_account_id:
            return False
        result = call_api("set_entry", {
            "session": session,
            "module_name": "Accounts",
            "name_value_list": {
                "id": {"name": "id", "value": test_account_id},
                "phone_office": {"name": "phone_office", "value": "+49 30 999888777"},
                "description": {"name": "description", "value": "AKTUALISIERT via API"},
            }
        })
        return bool(result.get("id"))

    test("Account updaten (set_entry mit ID)", update_account)

    def delete_account():
        if not test_account_id:
            return False
        result = call_api("set_entry", {
            "session": session,
            "module_name": "Accounts",
            "name_value_list": {
                "id": {"name": "id", "value": test_account_id},
                "deleted": {"name": "deleted", "value": "1"},
            }
        })
        return bool(result.get("id"))

    test("Account löschen (soft delete)", delete_account)

    # === LIST ===
    print("\n📌 LIST & FILTER")

    def list_accounts():
        result = call_api("get_entry_list", {
            "session": session,
            "module_name": "Accounts",
            "query": "",
            "order_by": "date_entered DESC",
            "offset": 0,
            "select_fields": ["name", "id", "date_entered", "account_type"],
            "max_results": 5,
            "deleted": 0
        })
        if "_error" in result:
            return False
        if result and "entry_list" in result:
            count = result.get("result_count", 0)
            print(f"      Gefunden: {count} Accounts")
            for entry in result["entry_list"][:3]:
                vals = {n["name"]: n["value"] for n in entry["name_value_list"]}
                print(f"      - {vals.get('name', 'N/A')} ({vals.get('account_type', 'N/A')})")
            return True
        return False

    test("Accounts auflisten", list_accounts)

    def search_accounts():
        result = call_api("search_by_module", {
            "session": session,
            "search_string": "API",
            "modules": ["Accounts"],
            "offset": 0,
            "max_results": 10,
        })
        if result and "entry_list" in result:
            count = result.get("entry_list", [])
            print(f"      Suche 'API': {len(count)} Treffer in Accounts")
            return True
        return False

    test("Globale Suche nach 'API'", search_accounts)
            "search_string": "API",
            "modules": ["Accounts"],
            "offset": 0,
            "max_results": 10,
        })
        if result and "entry_list" in result:
            count = result.get("entry_list", [])
            print(f"      Suche 'API': {len(count)} Treffer in Accounts")
            return True
        return False

    test("Globale Suche nach 'API'", search_by_module)

    # === RELATIONSHIPS ===
    print("\n📌 RELATIONSHIPS")

    def create_contact():
        result = call_api("set_entry", {
            "session": session,
            "module_name": "Contacts",
            "name_value_list": {
                "first_name": {"name": "first_name", "value": "Max"},
                "last_name": {"name": "last_name", "value": "Mustermann"},
                "email1": {"name": "email1", "value": "max@example.com"},
            }
        })
        if result.get("id"):
            print(f"      Contact ID: {result['id'][:10]}...")
            return result["id"]
        return None

    contact_id = test("Kontakt erstellen", create_contact)

    def create_account_for_rel():
        result = call_api("set_entry", {
            "session": session,
            "module_name": "Accounts",
            "name_value_list": {
                "name": {"name": "name", "value": f"Rel Test {datetime.now().strftime('%H:%M')}"},
            }
        })
        if result.get("id"):
            return result["id"]
        return None

    account_id = test("Account für Relationship", create_account_for_rel)

    def set_relationship():
        if not account_id or not contact_id:
            print("      ⏭️  Überspringe (kein Account/Kontakt)")
            return True  # Don't count as failure
        result = call_api("set_relationship", {
            "session": session,
            "module_name": "Accounts",
            "module_id": account_id,
            "link_field_name": "contacts",
            "related_ids": [contact_id],
        })
        return result.get("created", 0) > 0

    test("Relationship Account↔Kontakt", set_relationship)

    # === LOGOUT ===
    print("\n📌 LOGOUT")
    def do_logout():
        result = call_api("logout", {"session": session})
        return True  # logout returns empty on success

    test("Logout", do_logout)

    # === SUMMARY ===
    print(f"\n{'=' * 60}")
    if failures == 0:
        print(f"✅ ALLE TESTS BESTANDEN!")
    else:
        print(f"⚠️  {failures} Test(s) fehlgeschlagen")
    print(f"{'=' * 60}")


if __name__ == "__main__":
    main()