""" Advoware Adressen-API Tester Testet die Advoware Adressen-API umfassend, um herauszufinden: 1. Welche IDs für Mapping nutzbar sind 2. Welche Felder wirklich beschreibbar/änderbar sind 3. Wie sich die API bei mehreren Adressen verhält Basierend auf Erfahrungen mit Beteiligte-API, wo nur 8 von vielen Feldern funktionierten. Usage: python scripts/test_adressen_api.py """ import asyncio import sys import json from datetime import datetime from typing import Dict, Any, List sys.path.insert(0, '/opt/motia-app/bitbylaw') from services.advoware import AdvowareAPI # Test-Konfiguration TEST_BETNR = 104860 # Beteiligten-Nr für Tests # ANSI Color Codes GREEN = '\033[92m' RED = '\033[91m' YELLOW = '\033[93m' BLUE = '\033[94m' CYAN = '\033[96m' RESET = '\033[0m' BOLD = '\033[1m' class SimpleContext: """Mock context for logging""" class Logger: def info(self, msg): print(f"[INFO] {msg}") def error(self, msg): print(f"[ERROR] {msg}") def debug(self, msg): print(f"[DEBUG] {msg}") def warning(self, msg): print(f"[WARNING] {msg}") def __init__(self): self.logger = self.Logger() def print_header(title: str): """Print formatted section header""" print(f"\n{'='*80}") print(f"{BOLD}{CYAN}{title}{RESET}") print(f"{'='*80}\n") def print_success(msg: str): """Print success message""" print(f"{GREEN}✓ {msg}{RESET}") def print_error(msg: str): """Print error message""" print(f"{RED}✗ {msg}{RESET}") def print_warning(msg: str): """Print warning message""" print(f"{YELLOW}⚠ {msg}{RESET}") def print_info(msg: str): """Print info message""" print(f"{BLUE}ℹ {msg}{RESET}") async def test_1_get_existing_addresses(): """Test 1: Hole bestehende Adressen und analysiere Struktur""" print_header("TEST 1: GET Adressen - Struktur analysieren") context = SimpleContext() advo = AdvowareAPI(context=context) try: endpoint = f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen' print_info(f"GET {endpoint}") addresses = await advo.api_call(endpoint, method='GET') if not addresses: print_warning("Keine Adressen gefunden - wird in Test 2 erstellen") return [] print_success(f"Erfolgreich {len(addresses)} Adressen abgerufen") # Analysiere Struktur print(f"\n{BOLD}Anzahl Adressen:{RESET} {len(addresses)}") for i, addr in enumerate(addresses, 1): print(f"\n{BOLD}--- Adresse {i} ---{RESET}") print(f" id: {addr.get('id')}") print(f" beteiligterId: {addr.get('beteiligterId')}") print(f" reihenfolgeIndex: {addr.get('reihenfolgeIndex')}") print(f" rowId: {addr.get('rowId')}") print(f" strasse: {addr.get('strasse')}") print(f" plz: {addr.get('plz')}") print(f" ort: {addr.get('ort')}") print(f" land: {addr.get('land')}") print(f" postfach: {addr.get('postfach')}") print(f" postfachPLZ: {addr.get('postfachPLZ')}") print(f" anschrift: {addr.get('anschrift')}") print(f" standardAnschrift: {addr.get('standardAnschrift')}") print(f" bemerkung: {addr.get('bemerkung')}") print(f" gueltigVon: {addr.get('gueltigVon')}") print(f" gueltigBis: {addr.get('gueltigBis')}") # ID-Analyse print(f"\n{BOLD}ID-Analyse für Mapping:{RESET}") print(f" - 'id' vorhanden: {all('id' in a for a in addresses)}") print(f" - 'id' Typ: {type(addresses[0].get('id')) if addresses else 'N/A'}") print(f" - 'id' eindeutig: {len(set(a.get('id') for a in addresses)) == len(addresses)}") print(f" - 'rowId' vorhanden: {all('rowId' in a for a in addresses)}") print(f" - 'rowId' eindeutig: {len(set(a.get('rowId') for a in addresses)) == len(addresses)}") print_success("✓ ID-Felder 'id' und 'rowId' sind nutzbar für Mapping") return addresses except Exception as e: print_error(f"GET fehlgeschlagen: {e}") import traceback traceback.print_exc() return [] async def test_2_create_test_address(): """Test 2: Erstelle Test-Adresse mit allen Feldern""" print_header("TEST 2: POST - Neue Adresse erstellen") context = SimpleContext() advo = AdvowareAPI(context=context) # Vollständige Test-Daten mit allen Feldern test_address = { 'strasse': 'Teststraße 123', 'plz': '30159', 'ort': 'Hannover', 'land': 'DE', 'postfach': 'PF 10 20 30', 'postfachPLZ': '30001', 'anschrift': 'Teststraße 123\n30159 Hannover\nDeutschland', 'standardAnschrift': False, 'bemerkung': f'TEST-Adresse erstellt am {datetime.now().isoformat()}', 'gueltigVon': '2026-02-08T00:00:00', 'gueltigBis': '2027-12-31T23:59:59' } print_info("Erstelle Adresse mit allen Feldern:") print(json.dumps(test_address, indent=2, ensure_ascii=False)) try: endpoint = f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen' print_info(f"\nPOST {endpoint}") result = await advo.api_call(endpoint, method='POST', json_data=test_address) print_success("POST erfolgreich!") print(f"\n{BOLD}Response:{RESET}") # Advoware gibt Array zurück if isinstance(result, list): print_info(f"Response ist Array mit {len(result)} Elementen") if result: created_addr = result[0] print(json.dumps(created_addr, indent=2, ensure_ascii=False)) return created_addr else: print(json.dumps(result, indent=2, ensure_ascii=False)) return result except Exception as e: print_error(f"POST fehlgeschlagen: {e}") import traceback traceback.print_exc() return None async def test_3_verify_created_fields(created_addr: Dict): """Test 3: Vergleiche gesendete vs. zurückgegebene Daten""" print_header("TEST 3: Feld-Verifikation - Was wurde wirklich gespeichert?") if not created_addr: print_error("Keine Adresse zum Verifizieren") return # Erwartete vs. tatsächliche Werte expected = { 'strasse': 'Teststraße 123', 'plz': '30159', 'ort': 'Hannover', 'land': 'DE', 'postfach': 'PF 10 20 30', 'postfachPLZ': '30001', 'anschrift': 'Teststraße 123\n30159 Hannover\nDeutschland', 'standardAnschrift': False, 'bemerkung': 'TEST-Adresse', # Partial match 'gueltigVon': '2026-02-08', # Nur Datum-Teil 'gueltigBis': '2027-12-31' } working_fields = [] broken_fields = [] print(f"\n{BOLD}Feld-für-Feld-Vergleich:{RESET}\n") for field, expected_val in expected.items(): actual_val = created_addr.get(field) # Vergleich if field in ['bemerkung']: # Partial match für Felder mit Timestamps matches = expected_val in str(actual_val) if actual_val else False elif field in ['gueltigVon', 'gueltigBis']: # Datum-Vergleich (nur YYYY-MM-DD Teil) actual_date = str(actual_val).split('T')[0] if actual_val else None matches = actual_date == expected_val else: matches = actual_val == expected_val if matches: print_success(f"{field:20} : {actual_val}") working_fields.append(field) else: print_error(f"{field:20} : Expected '{expected_val}', Got '{actual_val}'") broken_fields.append(field) # Zusätzliche Felder prüfen print(f"\n{BOLD}Zusätzliche Felder:{RESET}") extra_fields = ['id', 'beteiligterId', 'reihenfolgeIndex', 'rowId'] for field in extra_fields: val = created_addr.get(field) if val is not None: print_success(f"{field:20} : {val}") # Zusammenfassung print(f"\n{BOLD}{'='*60}{RESET}") print(f"{GREEN}✓ Funktionierende Felder ({len(working_fields)}):{RESET}") for field in working_fields: print(f" - {field}") if broken_fields: print(f"\n{RED}✗ Nicht funktionierende Felder ({len(broken_fields)}):{RESET}") for field in broken_fields: print(f" - {field}") return created_addr async def test_4_update_address_full(row_id: str): """Test 4: Update mit allen Feldern (Read-Modify-Write Pattern)""" print_header("TEST 4: PUT - Adresse mit allen Feldern ändern") context = SimpleContext() advo = AdvowareAPI(context=context) try: # 1. Lese aktuelle Adresse print_info("Schritt 1: Lese aktuelle Adresse...") all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) # Finde via rowId current_addr = next((a for a in all_addresses if a.get('rowId') == row_id), None) if not current_addr: print_error(f"Adresse mit rowId {row_id} nicht gefunden") return None addr_id = current_addr.get('reihenfolgeIndex') print_success(f"Aktuelle Adresse geladen: {current_addr.get('strasse')} (Index: {addr_id})") # 2. Ändere ALLE Felder print_info("\nSchritt 2: Ändere alle Felder...") modified_addr = { 'strasse': 'GEÄNDERT Neue Straße 999', 'plz': '10115', 'ort': 'Berlin', 'land': 'DE', 'postfach': 'PF 99 88 77', 'postfachPLZ': '10001', 'anschrift': 'GEÄNDERT Neue Straße 999\n10115 Berlin\nDeutschland', 'standardAnschrift': True, # Toggle 'bemerkung': f'GEÄNDERT am {datetime.now().isoformat()}', 'gueltigVon': '2026-03-01T00:00:00', 'gueltigBis': '2028-12-31T23:59:59' } print(json.dumps(modified_addr, indent=2, ensure_ascii=False)) # 3. Update print_info(f"\nSchritt 3: PUT zu Advoware (Index: {addr_id})...") endpoint = f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{addr_id}' result = await advo.api_call(endpoint, method='PUT', json_data=modified_addr) print_success("PUT erfolgreich!") print(f"\n{BOLD}Response:{RESET}") print(json.dumps(result, indent=2, ensure_ascii=False)) return result except Exception as e: print_error(f"PUT fehlgeschlagen: {e}") import traceback traceback.print_exc() return None async def test_5_verify_update(row_id: str): """Test 5: Hole Adresse erneut und prüfe was wirklich geändert wurde""" print_header("TEST 5: Update-Verifikation - Was wurde wirklich geändert?") context = SimpleContext() advo = AdvowareAPI(context=context) try: all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) # Finde via rowId updated_addr = next((a for a in all_addresses if a.get('rowId') == row_id), None) if not updated_addr: print_error(f"Adresse mit rowId {row_id} nicht gefunden") return None print_success("Adresse neu geladen") # Erwartete geänderte Werte expected_changes = { 'strasse': 'GEÄNDERT Neue Straße 999', 'plz': '10115', 'ort': 'Berlin', 'land': 'DE', 'postfach': 'PF 99 88 77', 'postfachPLZ': '10001', 'standardAnschrift': True, 'bemerkung': 'GEÄNDERT am', 'gueltigVon': '2026-03-01', 'gueltigBis': '2028-12-31' } updatable_fields = [] readonly_fields = [] print(f"\n{BOLD}Änderungs-Verifikation:{RESET}\n") for field, expected_val in expected_changes.items(): actual_val = updated_addr.get(field) # Vergleich if field == 'bemerkung': changed = expected_val in str(actual_val) if actual_val else False elif field in ['gueltigVon', 'gueltigBis']: actual_date = str(actual_val).split('T')[0] if actual_val else None changed = actual_date == expected_val else: changed = actual_val == expected_val if changed: print_success(f"{field:20} : ✓ GEÄNDERT → {actual_val}") updatable_fields.append(field) else: print_error(f"{field:20} : ✗ NICHT GEÄNDERT (ist: {actual_val})") readonly_fields.append(field) # Zusammenfassung print(f"\n{BOLD}{'='*60}{RESET}") print(f"{GREEN}✓ Änderbare Felder ({len(updatable_fields)}):{RESET}") for field in updatable_fields: print(f" - {field}") if readonly_fields: print(f"\n{RED}✗ Nicht änderbare Felder ({len(readonly_fields)}):{RESET}") for field in readonly_fields: print(f" - {field}") return updated_addr except Exception as e: print_error(f"Verifikation fehlgeschlagen: {e}") import traceback traceback.print_exc() return None async def test_6_multiple_addresses_behavior(): """Test 6: Verhalten bei mehreren Adressen""" print_header("TEST 6: Mehrere Adressen - Verhalten testen") context = SimpleContext() advo = AdvowareAPI(context=context) try: # Hole alle Adressen all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) print_info(f"Aktuelle Anzahl Adressen: {len(all_addresses)}") # Erstelle 2. Test-Adresse print_info("\nErstelle 2. Test-Adresse...") test_addr_2 = { 'strasse': 'Zweite Straße 456', 'plz': '20095', 'ort': 'Hamburg', 'land': 'DE', 'standardAnschrift': False, 'bemerkung': 'TEST-Adresse 2' } result = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='POST', json_data=test_addr_2 ) if isinstance(result, list) and result: addr_2 = result[0] print_success(f"2. Adresse erstellt: ID {addr_2.get('id')}") # Hole erneut alle Adressen all_addresses_after = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) print_success(f"Neue Anzahl Adressen: {len(all_addresses_after)}") # Analysiere reihenfolgeIndex print(f"\n{BOLD}Reihenfolge-Analyse:{RESET}") for addr in all_addresses_after: print(f" ID {addr.get('id'):5} | Index: {addr.get('reihenfolgeIndex'):3} | " f"Standard: {addr.get('standardAnschrift')} | {addr.get('ort')}") # Prüfe standardAnschrift Logik standard_addrs = [a for a in all_addresses_after if a.get('standardAnschrift')] print(f"\n{BOLD}standardAnschrift-Logik:{RESET}") if len(standard_addrs) == 0: print_warning("Keine Adresse als Standard markiert") elif len(standard_addrs) == 1: print_success(f"Genau 1 Standard-Adresse (ID: {standard_addrs[0].get('id')})") else: print_error(f"MEHRERE Standard-Adressen: {len(standard_addrs)}") return all_addresses_after except Exception as e: print_error(f"Test fehlgeschlagen: {e}") import traceback traceback.print_exc() return [] async def test_7_field_by_field_update(row_id: str): """Test 7: Teste jedes Feld einzeln (einzelne Updates)""" print_header("TEST 7: Feld-für-Feld Update-Test") context = SimpleContext() advo = AdvowareAPI(context=context) # Hole Index für PUT all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) test_addr = next((a for a in all_addresses if a.get('rowId') == row_id), None) if not test_addr: print_error("Test-Adresse nicht gefunden") return {} addr_index = test_addr.get('reihenfolgeIndex') print_info(f"Verwende Adresse mit Index: {addr_index}") # Test-Felder mit Werten test_fields = { 'strasse': 'Einzeltest Straße', 'plz': '80331', 'ort': 'München', 'land': 'AT', 'postfach': 'PF 11 22', 'postfachPLZ': '80001', 'anschrift': 'Formatierte Anschrift\nTest', 'standardAnschrift': True, 'bemerkung': 'Einzelfeld-Test', 'gueltigVon': '2026-04-01T00:00:00', 'gueltigBis': '2026-12-31T23:59:59' } results = {} for field_name, test_value in test_fields.items(): print(f"\n{BOLD}Test Feld: {field_name}{RESET}") print_info(f"Setze auf: {test_value}") try: # 1. Lese aktuelle Adresse all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) current = next((a for a in all_addresses if a.get('rowId') == row_id), None) if not current: print_error(f"Adresse nicht gefunden") results[field_name] = 'FAILED' continue # 2. Update nur dieses eine Feld update_data = { 'strasse': current.get('strasse'), 'plz': current.get('plz'), 'ort': current.get('ort'), 'land': current.get('land'), 'standardAnschrift': current.get('standardAnschrift', False) } update_data[field_name] = test_value await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{addr_index}', method='PUT', json_data=update_data ) # 3. Verifiziere all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) updated = next((a for a in all_addresses if a.get('rowId') == row_id), None) actual_value = updated.get(field_name) # Vergleich (mit Toleranz für Datumsfelder) if field_name in ['gueltigVon', 'gueltigBis']: expected_date = test_value.split('T')[0] actual_date = str(actual_value).split('T')[0] if actual_value else None success = actual_date == expected_date else: success = actual_value == test_value if success: print_success(f"✓ FUNKTIONIERT: {actual_value}") results[field_name] = 'WORKING' else: print_error(f"✗ FUNKTIONIERT NICHT: Expected '{test_value}', Got '{actual_value}'") results[field_name] = 'BROKEN' except Exception as e: print_error(f"Fehler: {e}") results[field_name] = 'ERROR' await asyncio.sleep(0.5) # Rate limiting # Zusammenfassung print(f"\n{BOLD}{'='*60}{RESET}") print(f"{BOLD}FINAL RESULTS - Feld-für-Feld Test:{RESET}\n") working = [f for f, r in results.items() if r == 'WORKING'] broken = [f for f, r in results.items() if r == 'BROKEN'] errors = [f for f, r in results.items() if r == 'ERROR'] print(f"{GREEN}✓ WORKING ({len(working)}):{RESET}") for f in working: print(f" - {f}") if broken: print(f"\n{RED}✗ BROKEN ({len(broken)}):{RESET}") for f in broken: print(f" - {f}") if errors: print(f"\n{YELLOW}⚠ ERRORS ({len(errors)}):{RESET}") for f in errors: print(f" - {f}") return results async def main(): """Haupt-Test-Ablauf""" print(f"\n{BOLD}{CYAN}╔══════════════════════════════════════════════════════════════╗{RESET}") print(f"{BOLD}{CYAN}║ ADVOWARE ADRESSEN-API - UMFASSENDER FUNKTIONS-TEST ║{RESET}") print(f"{BOLD}{CYAN}╚══════════════════════════════════════════════════════════════╝{RESET}") print(f"\n{BOLD}Test-Konfiguration:{RESET}") print(f" BetNr: {TEST_BETNR}") print(f" Datum: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # Test 1: GET existing existing_addresses = await test_1_get_existing_addresses() # Test 2: POST new created_addr = await test_2_create_test_address() if not created_addr: print_error("\nTest abgebrochen: Konnte keine Adresse erstellen") return row_id = created_addr.get('rowId') initial_id = created_addr.get('id') if not row_id: print_error("\nTest abgebrochen: Keine rowId zurückgegeben") return print_warning(f"\n⚠️ KRITISCH: POST gibt id={initial_id} zurück") print_info(f"rowId: {row_id}") # Hole Adressen erneut, um echte ID zu finden print_info("\nHole Adressen erneut, um zu prüfen...") context = SimpleContext() advo = AdvowareAPI(context=context) try: all_addresses = await advo.api_call( f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen', method='GET' ) # Finde via rowId found_addr = next((a for a in all_addresses if a.get('rowId') == row_id), None) if found_addr: actual_id = found_addr.get('id') actual_index = found_addr.get('reihenfolgeIndex') print_success(f"✓ Adresse via rowId gefunden:") print(f" - id: {actual_id}") print(f" - reihenfolgeIndex: {actual_index}") print(f" - rowId: {row_id}") # KRITISCHE ERKENNTNIS if actual_id == 0: print_error("\n❌ KRITISCH: 'id' ist immer 0 - NICHT NUTZBAR für Mapping!") print_success(f"✓ Nur 'rowId' ist eindeutig → MUSS für Mapping verwendet werden") print_warning(f"⚠️ 'reihenfolgeIndex' könnte als Alternative dienen: {actual_index}") # Verwende reihenfolgeIndex als "ID" addr_id = actual_index print_info(f"\n>>> Verwende reihenfolgeIndex={addr_id} für weitere Tests") else: addr_id = actual_id print_info(f"\n>>> Test-Adressen-ID: {addr_id}") else: print_error("Konnte Adresse nicht via rowId finden") return except Exception as e: print_error(f"Fehler beim Abrufen: {e}") import traceback traceback.print_exc() return # Test 3: Verify created fields await test_3_verify_created_fields(created_addr) # Test 4: Update full await test_4_update_address_full(row_id) # Test 5: Verify update await test_5_verify_update(row_id) # Test 6: Multiple addresses await test_6_multiple_addresses_behavior() # Test 7: Field-by-field (most important!) await test_7_field_by_field_update(row_id) # Final Summary print(f"\n{BOLD}{CYAN}╔══════════════════════════════════════════════════════════════╗{RESET}") print(f"{BOLD}{CYAN}║ TEST ABGESCHLOSSEN ║{RESET}") print(f"{BOLD}{CYAN}╚══════════════════════════════════════════════════════════════╝{RESET}") print(f"\n{BOLD}Wichtigste Erkenntnisse:{RESET}") print(f" - Test-Adresse rowId: {row_id}") print(f" - ❌ KRITISCH: 'id' ist immer 0 - nicht nutzbar!") print(f" - ✓ 'rowId' ist eindeutig → MUSS für Mapping verwendet werden") print(f" - Siehe Feld-für-Feld Ergebnisse oben") print(f" - Dokumentation wird in ADRESSEN_SYNC_ANALYSE.md aktualisiert") print(f"\n{YELLOW}⚠️ ACHTUNG:{RESET} Test-Adressen wurden in Advoware erstellt!") print(f" Diese sollten manuell gelöscht oder via Support entfernt werden.") print(f" Test-Adressen enthalten 'TEST' oder 'GEÄNDERT' im Text.\n") if __name__ == '__main__': asyncio.run(main())