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.
697 lines
25 KiB
Python
697 lines
25 KiB
Python
"""
|
||
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())
|