feat: API test scripts + quickstart
Some checks failed
Docker Build & Push SugarCRM 6.5 CE / build-and-push (push) Has been cancelled
Some checks failed
Docker Build & Push SugarCRM 6.5 CE / build-and-push (push) Has been cancelled
- start.sh: All-in-One Launcher (compose up + wait + API test) - test_api_extended.py: Full CRUD, search, relationships - test_seed.py: Mass data generator (accounts/contacts/leads) - Updated README with test script documentation
This commit is contained in:
root
321
test_api_extended.py
Executable file
321
test_api_extended.py
Executable file
@@ -0,0 +1,321 @@
|
||||
#!/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_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()
|
||||
Reference in New Issue
Block a user