179 lines
6.5 KiB
Python
179 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Test-Skript zum Überprüfen der Many-to-Many-Beziehung mit additionalColumns
|
||
zwischen CDokumente und CAICollections
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
# API-Konfiguration (aus e2e_tests.py übernommen)
|
||
BASE_URL = "https://crm.bitbylaw.com"
|
||
API_KEY = "2b0747ca34d15032aa233ae043cc61bc"
|
||
HEADERS = {
|
||
"X-Api-Key": API_KEY,
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
def api_request(method, endpoint, data=None):
|
||
"""Führt einen API-Request aus"""
|
||
url = f"{BASE_URL}/api/v1/{endpoint}"
|
||
try:
|
||
if method == "GET":
|
||
response = requests.get(url, headers=HEADERS)
|
||
elif method == "POST":
|
||
response = requests.post(url, headers=HEADERS, json=data)
|
||
elif method == "PUT":
|
||
response = requests.put(url, headers=HEADERS, json=data)
|
||
elif method == "DELETE":
|
||
response = requests.delete(url, headers=HEADERS)
|
||
else:
|
||
raise ValueError(f"Unknown method: {method}")
|
||
|
||
response.raise_for_status()
|
||
return response.json() if response.text else {}
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ API Error: {e}")
|
||
if hasattr(e.response, 'text'):
|
||
print(f" Response: {e.response.text}")
|
||
return None
|
||
|
||
def main():
|
||
print("="*80)
|
||
print(" "*20 + "Many-to-Many Junction Test")
|
||
print("="*80)
|
||
print()
|
||
|
||
# Test-Daten
|
||
test_sync_id = f"TEST-SYNC-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
|
||
doc_id = None
|
||
collection_id = None
|
||
|
||
try:
|
||
# 1. CDokumente-Eintrag erstellen
|
||
print("1️⃣ Erstelle Test-Dokument...")
|
||
doc_data = {
|
||
"name": f"Test-Dokument für Junction-Test {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
|
||
"description": "Test-Dokument für Many-to-Many mit syncId"
|
||
}
|
||
doc_result = api_request("POST", "CDokumente", doc_data)
|
||
if not doc_result or 'id' not in doc_result:
|
||
print("❌ Fehler beim Erstellen des Dokuments")
|
||
return False
|
||
doc_id = doc_result['id']
|
||
print(f"✓ Dokument erstellt: {doc_id}")
|
||
print()
|
||
|
||
# 2. CAICollections-Eintrag erstellen
|
||
print("2️⃣ Erstelle Test-Collection...")
|
||
collection_data = {
|
||
"name": f"Test-Collection für Junction-Test {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
|
||
"description": "Test-Collection für Many-to-Many mit syncId"
|
||
}
|
||
collection_result = api_request("POST", "CAICollections", collection_data)
|
||
if not collection_result or 'id' not in collection_result:
|
||
print("❌ Fehler beim Erstellen der Collection")
|
||
return False
|
||
collection_id = collection_result['id']
|
||
print(f"✓ Collection erstellt: {collection_id}")
|
||
print()
|
||
|
||
# 3. Verknüpfung erstellen mit syncId
|
||
print("3️⃣ Verknüpfe Dokument mit Collection (mit syncId)...")
|
||
print(f" syncId: {test_sync_id}")
|
||
link_data = {
|
||
"id": collection_id,
|
||
"columns": {
|
||
"syncId": test_sync_id
|
||
}
|
||
}
|
||
link_result = api_request("POST", f"CDokumente/{doc_id}/cAICollections", link_data)
|
||
if link_result is None:
|
||
print("❌ Fehler beim Verknüpfen")
|
||
return False
|
||
print(f"✓ Verknüpfung erstellt")
|
||
print()
|
||
|
||
# 4. Beziehung über API abrufen
|
||
print("4️⃣ Rufe Beziehung über API ab...")
|
||
relations = api_request("GET", f"CDokumente/{doc_id}/cAICollections")
|
||
if not relations:
|
||
print("❌ Fehler beim Abrufen der Beziehung")
|
||
return False
|
||
|
||
print(f"✓ Beziehung abgerufen")
|
||
print("\nAPI-Response:")
|
||
print("-" * 80)
|
||
print(json.dumps(relations, indent=2, ensure_ascii=False))
|
||
print("-" * 80)
|
||
print()
|
||
|
||
# 5. Prüfe ob syncId vorhanden ist
|
||
print("5️⃣ Prüfe ob syncId in der Response vorhanden ist...")
|
||
if 'list' in relations and len(relations['list']) > 0:
|
||
first_relation = relations['list'][0]
|
||
if 'syncId' in first_relation:
|
||
returned_sync_id = first_relation['syncId']
|
||
if returned_sync_id == test_sync_id:
|
||
print(f"✅ syncId korrekt zurückgegeben: {returned_sync_id}")
|
||
success = True
|
||
else:
|
||
print(f"⚠️ syncId zurückgegeben, aber Wert stimmt nicht überein:")
|
||
print(f" Erwartet: {test_sync_id}")
|
||
print(f" Erhalten: {returned_sync_id}")
|
||
success = False
|
||
else:
|
||
print("❌ syncId ist NICHT in der API-Response vorhanden")
|
||
print(f" Vorhandene Felder: {list(first_relation.keys())}")
|
||
success = False
|
||
else:
|
||
print("❌ Keine Beziehungen in der Response gefunden")
|
||
success = False
|
||
print()
|
||
|
||
# 6. Direkter Datenbankcheck (optional)
|
||
print("6️⃣ Prüfe Datenbank direkt...")
|
||
import subprocess
|
||
db_check = subprocess.run([
|
||
"docker", "exec", "espocrm-db", "mariadb",
|
||
"-u", "espocrm", "-pdatabase_password", "espocrm",
|
||
"-e", f"SELECT * FROM c_a_i_collection_c_dokumente WHERE c_dokumente_id='{doc_id}' AND deleted=0;"
|
||
], capture_output=True, text=True)
|
||
|
||
if db_check.returncode == 0:
|
||
print("Datenbank-Inhalt:")
|
||
print("-" * 80)
|
||
print(db_check.stdout)
|
||
print("-" * 80)
|
||
else:
|
||
print(f"⚠️ Konnte Datenbank nicht direkt abfragen: {db_check.stderr}")
|
||
print()
|
||
|
||
return success
|
||
|
||
finally:
|
||
# Cleanup
|
||
print("7️⃣ Räume Test-Daten auf...")
|
||
if doc_id:
|
||
result = api_request("DELETE", f"CDokumente/{doc_id}")
|
||
if result is not None:
|
||
print(f"✓ Dokument gelöscht: {doc_id}")
|
||
if collection_id:
|
||
result = api_request("DELETE", f"CAICollections/{collection_id}")
|
||
if result is not None:
|
||
print(f"✓ Collection gelöscht: {collection_id}")
|
||
print()
|
||
|
||
if __name__ == "__main__":
|
||
print()
|
||
result = main()
|
||
print("="*80)
|
||
if result:
|
||
print("✅ TEST ERFOLGREICH - Many-to-Many mit additionalColumns funktioniert!")
|
||
else:
|
||
print("❌ TEST FEHLGESCHLAGEN - syncId nicht in API-Response")
|
||
print("="*80)
|
||
sys.exit(0 if result else 1)
|