#!/usr/bin/env python3 """ Test: DELETE + bemerkung-basiertes Matching für Adressen ========================================================== Ziele: 1. Teste ob DELETE funktioniert 2. Teste ob reihenfolgeIndex nach DELETE neu sortiert wird 3. Teste bemerkung als Matching-Field mit EspoCRM-ID 4. Validiere ob bemerkung stabil bleibt bei PUT """ import asyncio import sys import os from datetime import datetime # Add parent directory to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from services.advoware import AdvowareAPI # Test-Konfiguration TEST_BETNR = 104860 # Test Beteiligte ESPOCRM_TEST_IDS = ["espo-001", "espo-002", "espo-003"] # ANSI Color codes BOLD = '\033[1m' RED = '\033[91m' GREEN = '\033[92m' YELLOW = '\033[93m' BLUE = '\033[94m' RESET = '\033[0m' def print_header(text): print(f"\n{BOLD}{'='*80}{RESET}") print(f"{BOLD}{text}{RESET}") print(f"{BOLD}{'='*80}{RESET}\n") def print_success(text): print(f"{GREEN}✓ {text}{RESET}") def print_error(text): print(f"{RED}✗ {text}{RESET}") def print_warning(text): print(f"{YELLOW}⚠ {text}{RESET}") def print_info(text): print(f"{BLUE}ℹ {text}{RESET}") class SimpleLogger: """Minimal logger für AdvowareAPI""" def info(self, msg): pass def error(self, msg): print_error(msg) def debug(self, msg): pass def warning(self, msg): pass class SimpleContext: """Minimal context für AdvowareAPI""" def __init__(self): self.logger = SimpleLogger() def log_info(self, msg): pass def log_error(self, msg): print_error(msg) def log_debug(self, msg): pass async def test_1_create_addresses_with_espocrm_ids(): """Test 1: Erstelle 3 Adressen mit EspoCRM-IDs im bemerkung-Feld""" print_header("TEST 1: Erstelle Adressen mit EspoCRM-IDs im bemerkung-Feld") context = SimpleContext() advo = AdvowareAPI(context=context) created_addresses = [] for i, espo_id in enumerate(ESPOCRM_TEST_IDS, 1): print_info(f"\nErstelle Adresse {i} mit EspoCRM-ID: {espo_id}") address_data = { "strasse": f"Teststraße {i*10}", "plz": f"3015{i}", "ort": f"Testort-{i}", "land": "DE", "bemerkung": f"EspoCRM-ID: {espo_id}", # ← Unsere Sync-ID! "gueltigVon": f"2026-02-0{i}T00:00:00" } try: result = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='POST', json_data=address_data ) if result and len(result) > 0: addr = result[0] created_addresses.append({ 'espo_id': espo_id, 'rowId': addr.get('rowId'), 'reihenfolgeIndex': addr.get('reihenfolgeIndex'), 'bemerkung': addr.get('bemerkung') }) print_success(f"✓ Erstellt: rowId={addr.get('rowId')}, Index={addr.get('reihenfolgeIndex')}") print_info(f" bemerkung: {addr.get('bemerkung')}") else: print_error("POST lieferte leere Response") except Exception as e: print_error(f"Fehler beim Erstellen: {e}") import traceback traceback.print_exc() return None print_success(f"\n✓ {len(created_addresses)} Adressen erfolgreich erstellt") return created_addresses async def test_2_find_addresses_by_espocrm_id(): """Test 2: Finde Adressen via EspoCRM-ID im bemerkung-Feld""" print_header("TEST 2: Finde Adressen via EspoCRM-ID (bemerkung-Matching)") context = SimpleContext() advo = AdvowareAPI(context=context) try: all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) print_info(f"Gesamtanzahl Adressen: {len(all_addresses)}") # Parse bemerkung und finde unsere IDs found_mapping = {} for addr in all_addresses: bemerkung = addr.get('bemerkung', '') if bemerkung and 'EspoCRM-ID:' in bemerkung: # Parse: "EspoCRM-ID: espo-001" → "espo-001" espo_id = bemerkung.split('EspoCRM-ID:')[1].strip() found_mapping[espo_id] = { 'reihenfolgeIndex': addr.get('reihenfolgeIndex'), 'rowId': addr.get('rowId'), 'strasse': addr.get('strasse'), 'bemerkung': bemerkung } print_success(f"\n✓ {len(found_mapping)} Adressen mit EspoCRM-ID gefunden:") for espo_id, data in found_mapping.items(): print(f" {espo_id}:") print(f" - Index: {data['reihenfolgeIndex']}") print(f" - Straße: {data['strasse']}") print(f" - rowId: {data['rowId']}") # Validierung for test_id in ESPOCRM_TEST_IDS: if test_id in found_mapping: print_success(f"✓ {test_id} gefunden!") else: print_error(f"✗ {test_id} NICHT gefunden!") return found_mapping except Exception as e: print_error(f"Fehler beim Abrufen: {e}") import traceback traceback.print_exc() return None async def test_3_update_address_check_bemerkung_stability(): """Test 3: Versuche bemerkung zu ändern und prüfe Stabilität""" print_header("TEST 3: Teste ob bemerkung bei PUT stabil bleibt") context = SimpleContext() advo = AdvowareAPI(context=context) try: # Hole Adressen all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) # Finde erste Test-Adresse test_addr = None for addr in all_addresses: bemerkung = addr.get('bemerkung') or '' if bemerkung and 'EspoCRM-ID: espo-001' in bemerkung: test_addr = addr break if not test_addr: print_error("Test-Adresse mit espo-001 nicht gefunden") return False original_bemerkung = test_addr.get('bemerkung') reihenfolge_index = test_addr.get('reihenfolgeIndex') print_info(f"Test-Adresse Index: {reihenfolge_index}") print_info(f"Original bemerkung: {original_bemerkung}") # Versuche Update mit ANDERER bemerkung print_info("\nVersuche bemerkung zu ändern via PUT...") update_data = { "strasse": test_addr.get('strasse'), "plz": test_addr.get('plz'), "ort": "GEÄNDERT-ORT", # Ändere ort "land": test_addr.get('land'), "bemerkung": "GEÄNDERT: Diese bemerkung sollte NICHT überschrieben werden!" } await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{reihenfolge_index}', method='PUT', json_data=update_data ) # Hole erneut und prüfe print_info("\nHole Adresse erneut und prüfe bemerkung...") all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) updated_addr = next((a for a in all_addresses if a.get('reihenfolgeIndex') == reihenfolge_index), None) if updated_addr: updated_bemerkung = updated_addr.get('bemerkung') updated_ort = updated_addr.get('ort') print_info(f"Nach PUT bemerkung: {updated_bemerkung}") print_info(f"Nach PUT ort: {updated_ort}") if updated_bemerkung == original_bemerkung: print_success("\n✓✓✓ PERFEKT: bemerkung ist READ-ONLY bei PUT!") print_success("✓ EspoCRM-ID bleibt stabil → Perfekt für Matching!") return True else: print_warning("\n⚠ bemerkung wurde geändert - nicht stabil!") print_error(f" Original: {original_bemerkung}") print_error(f" Neu: {updated_bemerkung}") return False else: print_error("Adresse nach PUT nicht gefunden") return False except Exception as e: print_error(f"Fehler: {e}") import traceback traceback.print_exc() return False async def test_4_delete_middle_address_check_reindex(): """Test 4: Lösche mittlere Adresse und prüfe ob Indices neu sortiert werden""" print_header("TEST 4: DELETE - Werden reihenfolgeIndex neu sortiert?") context = SimpleContext() advo = AdvowareAPI(context=context) try: # Hole aktuelle Adressen print_info("VOR DELETE:") all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) # Zeige nur unsere Test-Adressen test_addresses_before = [] for addr in all_addresses: bemerkung = addr.get('bemerkung') or '' if bemerkung and 'EspoCRM-ID:' in bemerkung: test_addresses_before.append({ 'index': addr.get('reihenfolgeIndex'), 'espo_id': bemerkung.split('EspoCRM-ID:')[1].strip(), 'strasse': addr.get('strasse') }) print(f" Index {addr.get('reihenfolgeIndex')}: {bemerkung}") # Finde mittlere Adresse (espo-002) middle_addr = None for addr in all_addresses: bemerkung = addr.get('bemerkung') or '' if bemerkung and 'EspoCRM-ID: espo-002' in bemerkung: middle_addr = addr break if not middle_addr: print_error("Mittlere Test-Adresse (espo-002) nicht gefunden") return False delete_index = middle_addr.get('reihenfolgeIndex') print_warning(f"\nLösche Adresse mit Index: {delete_index} (espo-002)") # DELETE try: await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{delete_index}', method='DELETE' ) print_success("✓ DELETE erfolgreich") except Exception as e: print_error(f"DELETE fehlgeschlagen: {e}") # Versuche mit anderen Index-Werten print_info("Versuche DELETE mit rowId...") # Note: Swagger zeigt nur reihenfolgeIndex, aber vielleicht geht rowId? return None # Hole erneut und vergleiche print_info("\nNACH DELETE:") all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) test_addresses_after = [] for addr in all_addresses: bemerkung = addr.get('bemerkung') or '' if bemerkung and 'EspoCRM-ID:' in bemerkung: test_addresses_after.append({ 'index': addr.get('reihenfolgeIndex'), 'espo_id': bemerkung.split('EspoCRM-ID:')[1].strip(), 'strasse': addr.get('strasse') }) print(f" Index {addr.get('reihenfolgeIndex')}: {bemerkung}") # Analyse print_info("\n=== Index-Analyse ===") print(f"Anzahl vorher: {len(test_addresses_before)}") print(f"Anzahl nachher: {len(test_addresses_after)}") if len(test_addresses_after) == len(test_addresses_before) - 1: print_success("✓ Eine Adresse wurde gelöscht") # Prüfe ob Indices lückenlos sind indices_after = sorted([a['index'] for a in test_addresses_after]) print_info(f"Indices nachher: {indices_after}") # Erwartung: Lückenlos von 1 aufsteigend expected_indices = list(range(1, len(all_addresses) + 1)) all_indices = sorted([a.get('reihenfolgeIndex') for a in all_addresses]) if all_indices == expected_indices: print_success("✓✓✓ WICHTIG: Indices wurden NEU SORTIERT (lückenlos)!") print_warning("⚠ Das bedeutet: reihenfolgeIndex ist NICHT stabil nach DELETE!") print_success("✓ ABER: bemerkung-Matching funktioniert unabhängig davon!") else: print_info(f"Indices haben Lücken: {all_indices}") return True else: print_error("Unerwartete Anzahl Adressen nach DELETE") return False except Exception as e: print_error(f"Fehler: {e}") import traceback traceback.print_exc() return None async def test_5_restore_deleted_address(): """Test 5: Stelle gelöschte Adresse wieder her""" print_header("TEST 5: Stelle gelöschte Adresse wieder her (espo-002)") context = SimpleContext() advo = AdvowareAPI(context=context) address_data = { "strasse": "Teststraße 20", "plz": "30152", "ort": "Testort-2", "land": "DE", "bemerkung": "EspoCRM-ID: espo-002", "gueltigVon": "2026-02-02T00:00:00" } try: result = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='POST', json_data=address_data ) if result and len(result) > 0: addr = result[0] print_success(f"✓ Adresse wiederhergestellt: Index={addr.get('reihenfolgeIndex')}") return True else: print_error("POST fehlgeschlagen") return False except Exception as e: print_error(f"Fehler: {e}") return False async def main(): print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}") print(f"{BOLD}║ DELETE + bemerkung-Matching Tests für Adressen-Sync ║{RESET}") print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n") print(f"Test-Konfiguration:") print(f" BetNr: {TEST_BETNR}") print(f" Test-IDs: {ESPOCRM_TEST_IDS}") print(f" Datum: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # Test 1: Erstelle Adressen mit EspoCRM-IDs created = await test_1_create_addresses_with_espocrm_ids() if not created: print_error("\nTest abgebrochen: Konnte Adressen nicht erstellen") return # Test 2: Finde via bemerkung found = await test_2_find_addresses_by_espocrm_id() if not found or len(found) != len(ESPOCRM_TEST_IDS): print_error("\nTest abgebrochen: Matching fehlgeschlagen") return # Test 3: bemerkung Stabilität is_stable = await test_3_update_address_check_bemerkung_stability() # Test 4: DELETE und Re-Index await test_4_delete_middle_address_check_reindex() # Test 5: Restore await test_5_restore_deleted_address() # Finale Übersicht print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}") print(f"{BOLD}║ FINALE ERKENNTNISSE ║{RESET}") print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n") if is_stable: print_success("✓✓✓ bemerkung-Feld ist PERFEKT für Sync-Matching:") print_success(" 1. Kann bei POST gesetzt werden") print_success(" 2. Ist READ-ONLY bei PUT (bleibt stabil)") print_success(" 3. Überlebt Index-Änderungen durch DELETE") print_success(" 4. Format: 'EspoCRM-ID: {uuid}' ist eindeutig parsebar") print() print_info("💡 Empfohlene Sync-Strategie:") print_info(" - Beim Erstellen: bemerkung = 'EspoCRM-ID: {espo_address_id}'") print_info(" - Beim Sync: GET alle Adressen, parse bemerkung, match via ID") print_info(" - Bei DELETE in Advoware: EspoCRM-Adresse als 'deleted' markieren") print_info(" - Bei Konflikt: bemerkung hat Vorrang vor reihenfolgeIndex") else: print_warning("⚠ bemerkung-Matching hat Einschränkungen - siehe Details oben") print(f"\n{YELLOW}⚠️ ACHTUNG: Test-Adressen mit 'EspoCRM-ID:' im bemerkung-Feld{RESET}") print(f"{YELLOW} sollten manuell bereinigt werden.{RESET}\n") if __name__ == "__main__": asyncio.run(main())