introducing junction table extension
This commit is contained in:
178
custom/scripts/junctiontabletests/test_junction_api.py
Normal file
178
custom/scripts/junctiontabletests/test_junction_api.py
Normal file
@@ -0,0 +1,178 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user