133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Teste verschiedene API-Syntaxen für additionalColumns
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
from datetime import datetime
|
|
|
|
BASE_URL = "https://crm.bitbylaw.com"
|
|
API_KEY = "e53def10eea27b92a6cd00f40a3e09a4"
|
|
HEADERS = {
|
|
"X-Api-Key": API_KEY,
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
def api_request(method, endpoint, data=None):
|
|
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 == "DELETE":
|
|
response = requests.delete(url, headers=HEADERS)
|
|
|
|
return response
|
|
except Exception as e:
|
|
print(f" Exception: {e}")
|
|
return None
|
|
|
|
print("\n" + "="*80)
|
|
print(" "*20 + "API Syntax Variations Test")
|
|
print("="*80 + "\n")
|
|
|
|
doc_id = None
|
|
collection_id = None
|
|
|
|
try:
|
|
# Erstelle Test-Entities
|
|
print("Erstelle Test-Entities...")
|
|
doc = api_request("POST", "CDokumente", {
|
|
"name": f"Test Doc {datetime.now().strftime('%H:%M:%S')}"
|
|
}).json()
|
|
doc_id = doc['id']
|
|
print(f" Doc: {doc_id}")
|
|
|
|
collection = api_request("POST", "CAICollections", {
|
|
"name": f"Test Col {datetime.now().strftime('%H:%M:%S')}"
|
|
}).json()
|
|
collection_id = collection['id']
|
|
print(f" Col: {collection_id}\n")
|
|
|
|
# Teste verschiedene Syntaxen
|
|
variations = [
|
|
{
|
|
"name": "Variante 1: columns mit syncId",
|
|
"data": {
|
|
"id": collection_id,
|
|
"columns": {
|
|
"syncId": "TEST-SYNC-001"
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "Variante 2: Direkt syncId im Body",
|
|
"data": {
|
|
"id": collection_id,
|
|
"syncId": "TEST-SYNC-002"
|
|
}
|
|
},
|
|
{
|
|
"name": "Variante 3: columns mit sync_id (Snake-Case)",
|
|
"data": {
|
|
"id": collection_id,
|
|
"columns": {
|
|
"sync_id": "TEST-SYNC-003"
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "Variante 4: additionalColumns",
|
|
"data": {
|
|
"id": collection_id,
|
|
"additionalColumns": {
|
|
"syncId": "TEST-SYNC-004"
|
|
}
|
|
}
|
|
}
|
|
]
|
|
|
|
for i, variant in enumerate(variations, 1):
|
|
print(f"{i}. {variant['name']}")
|
|
|
|
# Lösche vorherige Verknüpfung
|
|
api_request("DELETE", f"CDokumente/{doc_id}/cAICollections", {"id": collection_id})
|
|
|
|
# Verknüpfe mit Variante
|
|
response = api_request("POST", f"CDokumente/{doc_id}/cAICollections", variant['data'])
|
|
print(f" Status: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
# Prüfe Datenbank
|
|
import subprocess
|
|
db_check = subprocess.run([
|
|
"docker", "exec", "espocrm-db", "mariadb",
|
|
"-u", "espocrm", "-pdatabase_password", "espocrm",
|
|
"-e", f"SELECT sync_id FROM c_a_i_collection_c_dokumente WHERE c_dokumente_id='{doc_id}' AND deleted=0;"
|
|
], capture_output=True, text=True)
|
|
|
|
lines = db_check.stdout.strip().split('\n')
|
|
if len(lines) > 1:
|
|
sync_id_value = lines[1].strip()
|
|
if sync_id_value and sync_id_value != "NULL":
|
|
print(f" ✅ syncId in DB: {sync_id_value}")
|
|
else:
|
|
print(f" ❌ syncId ist NULL")
|
|
else:
|
|
print(f" ⚠️ Keine Zeile in DB")
|
|
else:
|
|
print(f" ❌ Request fehlgeschlagen: {response.text[:100]}")
|
|
|
|
print()
|
|
|
|
finally:
|
|
print("Cleanup...")
|
|
if doc_id:
|
|
api_request("DELETE", f"CDokumente/{doc_id}")
|
|
if collection_id:
|
|
api_request("DELETE", f"CAICollections/{collection_id}")
|
|
|
|
print("="*80 + "\n")
|