bitbylaw
7856dd1d68
- Implemented comprehensive tests for the Kommunikation Sync functionality, covering base64 encoding, marker parsing, creation, type detection, and integration scenarios. - Added a verification script to check for unique IDs in Advoware communications, ensuring stability and integrity of the IDs. - Created utility scripts for code validation, notification testing, and PUT response detail analysis to enhance development and testing processes. - Updated README with details on new tools and their usage.
362 lines
11 KiB
Python
362 lines
11 KiB
Python
"""
|
||
Test: Advoware Kommunikation API
|
||
Testet POST/GET/PUT/DELETE Operationen für Kommunikationen
|
||
|
||
Basierend auf Swagger:
|
||
- POST /api/v1/advonet/Beteiligte/{beteiligterId}/Kommunikationen
|
||
- PUT /api/v1/advonet/Beteiligte/{beteiligterId}/Kommunikationen/{kommunikationId}
|
||
- GET enthalten in Beteiligte response (kommunikation array)
|
||
- DELETE nicht dokumentiert (wird getestet)
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import sys
|
||
from services.advoware import AdvowareAPI
|
||
from services.espocrm import EspoCRMAPI
|
||
|
||
# Test-Beteiligter
|
||
TEST_BETNR = 104860 # Angela Mustermanns
|
||
|
||
# KommKz Enum (Kommunikationskennzeichen)
|
||
KOMMKZ = {
|
||
1: 'TelGesch',
|
||
2: 'FaxGesch',
|
||
3: 'Mobil',
|
||
4: 'MailGesch',
|
||
5: 'Internet',
|
||
6: 'TelPrivat',
|
||
7: 'FaxPrivat',
|
||
8: 'MailPrivat',
|
||
9: 'AutoTelefon',
|
||
10: 'Sonstige',
|
||
11: 'EPost',
|
||
12: 'Bea'
|
||
}
|
||
|
||
|
||
class SimpleContext:
|
||
"""Einfacher Context für Logging"""
|
||
class Logger:
|
||
def info(self, msg): print(f"ℹ️ {msg}")
|
||
def error(self, msg): print(f"❌ {msg}")
|
||
def warning(self, msg): print(f"⚠️ {msg}")
|
||
def debug(self, msg): print(f"🔍 {msg}")
|
||
|
||
def __init__(self):
|
||
self.logger = self.Logger()
|
||
|
||
|
||
def print_section(title):
|
||
print("\n" + "="*70)
|
||
print(title)
|
||
print("="*70 + "\n")
|
||
|
||
|
||
def print_json(title, data):
|
||
print(f"\n{title}:")
|
||
print("-" * 70)
|
||
print(json.dumps(data, indent=2, ensure_ascii=False))
|
||
print()
|
||
|
||
|
||
async def test_get_existing_kommunikationen():
|
||
"""Hole bestehende Kommunikationen vom Test-Beteiligten"""
|
||
print_section("TEST 1: GET Bestehende Kommunikationen")
|
||
|
||
context = SimpleContext()
|
||
advo = AdvowareAPI(context)
|
||
|
||
# Hole kompletten Beteiligten
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}',
|
||
method='GET'
|
||
)
|
||
|
||
# Response ist ein Array (selbst bei einzelnem Beteiligten)
|
||
if isinstance(result, list) and len(result) > 0:
|
||
beteiligte = result[0]
|
||
elif isinstance(result, dict):
|
||
beteiligte = result
|
||
else:
|
||
print(f"❌ Unerwartetes Response-Format: {type(result)}")
|
||
return []
|
||
|
||
kommunikationen = beteiligte.get('kommunikation', [])
|
||
|
||
print(f"✓ Beteiligter geladen: {beteiligte.get('name')} {beteiligte.get('vorname')}")
|
||
print(f"✓ Kommunikationen gefunden: {len(kommunikationen)}")
|
||
|
||
if kommunikationen:
|
||
print_json("Bestehende Kommunikationen", kommunikationen)
|
||
|
||
# Analysiere Felder
|
||
first = kommunikationen[0]
|
||
print("📊 Felder-Analyse (erste Kommunikation):")
|
||
for key, value in first.items():
|
||
print(f" - {key}: {value} ({type(value).__name__})")
|
||
else:
|
||
print("ℹ️ Keine Kommunikationen vorhanden")
|
||
|
||
return kommunikationen
|
||
|
||
|
||
async def test_post_kommunikation():
|
||
"""Teste POST - Neue Kommunikation erstellen"""
|
||
print_section("TEST 2: POST - Neue Kommunikation erstellen")
|
||
|
||
context = SimpleContext()
|
||
advo = AdvowareAPI(context)
|
||
|
||
# Test verschiedene KommKz Typen
|
||
test_cases = [
|
||
{
|
||
'name': 'Geschäftstelefon',
|
||
'data': {
|
||
'kommKz': 1, # TelGesch
|
||
'tlf': '+49 511 123456-10',
|
||
'bemerkung': 'TEST: Hauptnummer',
|
||
'online': False
|
||
}
|
||
},
|
||
{
|
||
'name': 'Geschäfts-Email',
|
||
'data': {
|
||
'kommKz': 4, # MailGesch
|
||
'tlf': 'test@example.com',
|
||
'bemerkung': 'TEST: Email',
|
||
'online': True
|
||
}
|
||
},
|
||
{
|
||
'name': 'Mobiltelefon',
|
||
'data': {
|
||
'kommKz': 3, # Mobil
|
||
'tlf': '+49 170 1234567',
|
||
'bemerkung': 'TEST: Mobil',
|
||
'online': False
|
||
}
|
||
}
|
||
]
|
||
|
||
created_ids = []
|
||
|
||
for test in test_cases:
|
||
print(f"\n📝 Erstelle: {test['name']}")
|
||
print_json("Request Payload", test['data'])
|
||
|
||
try:
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}/Kommunikationen',
|
||
method='POST',
|
||
data=test['data']
|
||
)
|
||
|
||
print_json("Response", result)
|
||
|
||
# Extrahiere ID
|
||
if isinstance(result, list) and len(result) > 0:
|
||
created_id = result[0].get('id')
|
||
created_ids.append(created_id)
|
||
print(f"✅ Erstellt mit ID: {created_id}")
|
||
elif isinstance(result, dict):
|
||
created_id = result.get('id')
|
||
created_ids.append(created_id)
|
||
print(f"✅ Erstellt mit ID: {created_id}")
|
||
else:
|
||
print(f"❌ Unerwartetes Response-Format: {type(result)}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ Fehler: {e}")
|
||
|
||
return created_ids
|
||
|
||
|
||
async def test_put_kommunikation(komm_id):
|
||
"""Teste PUT - Kommunikation aktualisieren"""
|
||
print_section(f"TEST 3: PUT - Kommunikation {komm_id} aktualisieren")
|
||
|
||
context = SimpleContext()
|
||
advo = AdvowareAPI(context)
|
||
|
||
# Hole aktuelle Daten
|
||
print("📥 Lade aktuelle Kommunikation...")
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}',
|
||
method='GET'
|
||
)
|
||
|
||
# Response ist ein Array
|
||
if isinstance(result, list) and len(result) > 0:
|
||
beteiligte = result[0]
|
||
elif isinstance(result, dict):
|
||
beteiligte = result
|
||
else:
|
||
print(f"❌ Unerwartetes Response-Format")
|
||
return False
|
||
|
||
kommunikationen = beteiligte.get('kommunikation', [])
|
||
current_komm = next((k for k in kommunikationen if k.get('id') == komm_id), None)
|
||
|
||
if not current_komm:
|
||
print(f"❌ Kommunikation {komm_id} nicht gefunden!")
|
||
return False
|
||
|
||
print_json("Aktuelle Daten", current_komm)
|
||
|
||
# Test 1: Ändere tlf-Feld
|
||
print("\n🔄 Test 1: Ändere tlf (Telefonnummer/Email)")
|
||
update_data = {
|
||
'kommKz': current_komm['kommKz'],
|
||
'tlf': '+49 511 999999-99', # Neue Nummer
|
||
'bemerkung': current_komm.get('bemerkung', ''),
|
||
'online': current_komm.get('online', False)
|
||
}
|
||
|
||
print_json("Update Payload", update_data)
|
||
|
||
try:
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}/Kommunikationen/{komm_id}',
|
||
method='PUT',
|
||
data=update_data
|
||
)
|
||
print_json("Response", result)
|
||
print("✅ tlf erfolgreich geändert")
|
||
except Exception as e:
|
||
print(f"❌ Fehler: {e}")
|
||
return False
|
||
|
||
# Test 2: Ändere bemerkung
|
||
print("\n🔄 Test 2: Ändere bemerkung")
|
||
update_data['bemerkung'] = 'TEST: Geändert via API'
|
||
print_json("Update Payload", update_data)
|
||
|
||
try:
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}/Kommunikationen/{komm_id}',
|
||
method='PUT',
|
||
data=update_data
|
||
)
|
||
print_json("Response", result)
|
||
print("✅ bemerkung erfolgreich geändert")
|
||
except Exception as e:
|
||
print(f"❌ Fehler: {e}")
|
||
return False
|
||
|
||
# Test 3: Ändere kommKz (Typ)
|
||
print("\n🔄 Test 3: Ändere kommKz (Kommunikationstyp)")
|
||
update_data['kommKz'] = 6 # TelPrivat statt TelGesch
|
||
print_json("Update Payload", update_data)
|
||
|
||
try:
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}/Kommunikationen/{komm_id}',
|
||
method='PUT',
|
||
data=update_data
|
||
)
|
||
print_json("Response", result)
|
||
print("✅ kommKz erfolgreich geändert")
|
||
except Exception as e:
|
||
print(f"❌ Fehler: {e}")
|
||
return False
|
||
|
||
# Test 4: Ändere online-Flag
|
||
print("\n🔄 Test 4: Ändere online-Flag")
|
||
update_data['online'] = not update_data['online']
|
||
print_json("Update Payload", update_data)
|
||
|
||
try:
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}/Kommunikationen/{komm_id}',
|
||
method='PUT',
|
||
data=update_data
|
||
)
|
||
print_json("Response", result)
|
||
print("✅ online erfolgreich geändert")
|
||
except Exception as e:
|
||
print(f"❌ Fehler: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
async def test_delete_kommunikation(komm_id):
|
||
"""Teste DELETE - Kommunikation löschen"""
|
||
print_section(f"TEST 4: DELETE - Kommunikation {komm_id} löschen")
|
||
|
||
context = SimpleContext()
|
||
advo = AdvowareAPI(context)
|
||
|
||
print(f"🗑️ Versuche Kommunikation {komm_id} zu löschen...")
|
||
|
||
try:
|
||
result = await advo.api_call(
|
||
f'api/v1/advonet/Beteiligte/{TEST_BETNR}/Kommunikationen/{komm_id}',
|
||
method='DELETE'
|
||
)
|
||
print_json("Response", result)
|
||
print("✅ DELETE erfolgreich!")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ DELETE fehlgeschlagen: {e}")
|
||
|
||
# Check ob 403 Forbidden (wie bei Adressen)
|
||
if '403' in str(e):
|
||
print("⚠️ DELETE ist FORBIDDEN (wie bei Adressen)")
|
||
|
||
return False
|
||
|
||
|
||
async def main():
|
||
print("\n" + "="*70)
|
||
print("ADVOWARE KOMMUNIKATION API - VOLLSTÄNDIGER TEST")
|
||
print("="*70)
|
||
print(f"\nTest-Beteiligter: {TEST_BETNR}")
|
||
print("\nKommKz (Kommunikationskennzeichen):")
|
||
for kz, name in KOMMKZ.items():
|
||
print(f" {kz:2d} = {name}")
|
||
|
||
try:
|
||
# TEST 1: GET bestehende
|
||
existing = await test_get_existing_kommunikationen()
|
||
|
||
# TEST 2: POST neue
|
||
created_ids = await test_post_kommunikation()
|
||
|
||
if not created_ids:
|
||
print("\n❌ Keine Kommunikationen erstellt - Tests abgebrochen")
|
||
return
|
||
|
||
# TEST 3: PUT update (erste erstellte)
|
||
first_id = created_ids[0]
|
||
await test_put_kommunikation(first_id)
|
||
|
||
# TEST 4: DELETE (erste erstellte)
|
||
await test_delete_kommunikation(first_id)
|
||
|
||
# Finale Übersicht
|
||
print_section("ZUSAMMENFASSUNG")
|
||
print("✅ POST: Funktioniert (3 Typen getestet)")
|
||
print("✅ GET: Funktioniert (über Beteiligte-Endpoint)")
|
||
print("✓/✗ PUT: Siehe Testergebnisse oben")
|
||
print("✓/✗ DELETE: Siehe Testergebnisse oben")
|
||
|
||
print("\n⚠️ WICHTIG:")
|
||
print(f" - Test-Kommunikationen in Advoware manuell prüfen!")
|
||
print(f" - BetNr: {TEST_BETNR}")
|
||
print(" - Suche nach: 'TEST:'")
|
||
|
||
if len(created_ids) > 1:
|
||
print(f"\n📝 Erstellt wurden IDs: {created_ids}")
|
||
print(" Falls DELETE nicht funktioniert, manuell löschen!")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ Unerwarteter Fehler: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
asyncio.run(main())
|