bsiggel/motia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement address synchronization between EspoCRM and Advoware

Browse Source 
- Add AdressenMapper for transforming addresses between EspoCRM and Advoware formats.
- Create AdressenSync class to handle address creation, update, and deletion synchronization.
- Introduce NotificationManager for managing manual intervention notifications in case of sync issues.
- Implement detailed logging for address sync operations and error handling.
- Ensure READ-ONLY field changes are detected and notified for manual resolution.
This commit is contained in:
bitbylaw
2026-02-08 14:29:29 +00:00
parent 68c8b398aa
commit c770f2c8ee
15 changed files with 6427 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

696
bitbylaw/scripts/test_adressen_api.py Normal file
View File

@@ -0,0 +1,696 @@
"""
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())

466
bitbylaw/scripts/test_adressen_deactivate_ordering.py Normal file
View File

@@ -0,0 +1,466 @@
#!/usr/bin/env python3
"""
Test: Deaktivierung via gueltigBis + reihenfolgeIndex-Verhalten
================================================================
Ziele:
1. Teste ob abgelaufene Adressen (gueltigBis < heute) ausgeblendet werden
2. Teste ob man reihenfolgeIndex beim POST setzen kann
3. Teste ob neue Adressen automatisch ans Ende rutschen
4. Teste ob man reihenfolgeIndex via PUT ändern kann (Sortierung)
"""
import asyncio
import sys
import os
from datetime import datetime, timedelta
# 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
# 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_expired_address():
"""Test 1: Erstelle Adresse mit gueltigBis in der Vergangenheit"""
print_header("TEST 1: Adresse mit gueltigBis in Vergangenheit (abgelaufen)")
context = SimpleContext()
advo = AdvowareAPI(context=context)
# Datum in der Vergangenheit
expired_date = "2023-12-31T23:59:59"
address_data = {
"strasse": "Abgelaufene Straße 99",
"plz": "99999",
"ort": "Vergangenheit",
"land": "DE",
"bemerkung": "TEST-ABGELAUFEN: Diese Adresse ist seit 2023 ungültig",
"gueltigVon": "2020-01-01T00:00:00",
"gueltigBis": expired_date # ← In der Vergangenheit!
}
print_info(f"Erstelle Adresse mit gueltigBis: {expired_date} (vor 2+ Jahren)")
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 erstellt: rowId={addr.get('rowId')}")
print_info(f" gueltigBis: {addr.get('gueltigBis')}")
print_info(f" reihenfolgeIndex: {addr.get('reihenfolgeIndex')}")
return addr.get('bemerkung')
else:
print_error("POST lieferte keine Response")
return None
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def test_2_check_if_expired_address_visible():
"""Test 2: Prüfe ob abgelaufene Adresse in GET sichtbar ist"""
print_header("TEST 2: Ist abgelaufene Adresse in GET sichtbar?")
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)}")
# Suche abgelaufene Adresse
expired_found = None
active_count = 0
expired_count = 0
today = datetime.now()
for addr in all_addresses:
bemerkung = addr.get('bemerkung') or ''
gueltig_bis = addr.get('gueltigBis')
if 'TEST-ABGELAUFEN' in bemerkung:
expired_found = addr
print_success(f"\n✓ Abgelaufene Test-Adresse gefunden!")
print_info(f" Index: {addr.get('reihenfolgeIndex')}")
print_info(f" gueltigBis: {gueltig_bis}")
print_info(f" Straße: {addr.get('strasse')}")
# Zähle aktive vs. abgelaufene
if gueltig_bis:
try:
bis_date = datetime.fromisoformat(gueltig_bis.replace('Z', '+00:00'))
if bis_date < today:
expired_count += 1
else:
active_count += 1
except:
pass
print(f"\n{BOLD}Statistik:{RESET}")
print(f" Aktive Adressen (gueltigBis > heute): {active_count}")
print(f" Abgelaufene Adressen (gueltigBis < heute): {expired_count}")
print(f" Ohne gueltigBis: {len(all_addresses) - active_count - expired_count}")
if expired_found:
print_error("\n❌ WICHTIG: Abgelaufene Adressen werden NICHT gefiltert!")
print_warning("⚠ GET /Adressen zeigt ALLE Adressen, auch abgelaufene")
print_info("💡 Filtern nach gueltigBis muss CLIENT-seitig erfolgen")
return True
else:
print_success("\n✓ Abgelaufene Adresse nicht sichtbar (wird gefiltert)")
return False
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def test_3_create_with_explicit_reihenfolgeIndex():
"""Test 3: Versuche reihenfolgeIndex beim POST zu setzen"""
print_header("TEST 3: Kann man reihenfolgeIndex beim POST setzen?")
context = SimpleContext()
advo = AdvowareAPI(context=context)
# Versuche mit explizitem Index
address_data = {
"reihenfolgeIndex": 999, # ← Versuche expliziten Index
"strasse": "Test Index 999",
"plz": "88888",
"ort": "Indextest",
"land": "DE",
"bemerkung": "TEST-INDEX: Versuch mit explizitem reihenfolgeIndex=999"
}
print_info("Versuche POST mit reihenfolgeIndex=999...")
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]
actual_index = addr.get('reihenfolgeIndex')
print_info(f"Response reihenfolgeIndex: {actual_index}")
# Hole alle Adressen und prüfe wo sie gelandet ist
all_addresses = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
found = None
for a in all_addresses:
if (a.get('bemerkung') or '').startswith('TEST-INDEX'):
found = a
break
if found:
real_index = found.get('reihenfolgeIndex')
print_info(f"GET zeigt reihenfolgeIndex: {real_index}")
if real_index == 999:
print_success("\n✓ reihenfolgeIndex kann explizit gesetzt werden!")
print_warning("⚠ ABER: Das könnte bestehende Adressen verschieben!")
elif real_index == 0:
print_warning("\n⚠ POST gibt reihenfolgeIndex=0 zurück")
print_info("→ Echter Index wird erst nach GET sichtbar")
else:
print_error(f"\n❌ reihenfolgeIndex={real_index} ignoriert Vorgabe (999)")
print_success("✓ Index wird automatisch vergeben (ans Ende)")
return real_index
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def test_4_create_multiple_check_ordering():
"""Test 4: Erstelle mehrere Adressen und prüfe Reihenfolge"""
print_header("TEST 4: Mehrere neue Adressen - werden sie ans Ende gereiht?")
context = SimpleContext()
advo = AdvowareAPI(context=context)
print_info("Hole aktuelle Adressen...")
all_before = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
max_index_before = max([a.get('reihenfolgeIndex', 0) for a in all_before])
count_before = len(all_before)
print_info(f" Anzahl vorher: {count_before}")
print_info(f" Höchster Index: {max_index_before}")
# Erstelle 3 neue Adressen
print_info("\nErstelle 3 neue Adressen...")
created_ids = []
for i in range(1, 4):
address_data = {
"strasse": f"Reihenfolge-Test {i}",
"plz": f"7777{i}",
"ort": f"Stadt-{i}",
"land": "DE",
"bemerkung": f"TEST-REIHENFOLGE-{i}"
}
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:
created_ids.append(f"TEST-REIHENFOLGE-{i}")
print_success(f" ✓ Adresse {i} erstellt")
except Exception as e:
print_error(f" ✗ Fehler bei Adresse {i}: {e}")
# Hole alle Adressen erneut
print_info("\nHole Adressen erneut...")
all_after = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
count_after = len(all_after)
print_info(f" Anzahl nachher: {count_after}")
print_info(f" Neue Adressen: {count_after - count_before}")
# Finde unsere Test-Adressen
print(f"\n{BOLD}Reihenfolge der neuen Test-Adressen:{RESET}")
test_addresses = []
for addr in all_after:
bemerkung = addr.get('bemerkung') or ''
if 'TEST-REIHENFOLGE-' in bemerkung:
test_addresses.append({
'bemerkung': bemerkung,
'index': addr.get('reihenfolgeIndex'),
'strasse': addr.get('strasse')
})
test_addresses.sort(key=lambda x: x['index'])
for t in test_addresses:
print(f" Index {t['index']:2d}: {t['bemerkung']} ({t['strasse']})")
# Analyse
if len(test_addresses) >= 3:
indices = [t['index'] for t in test_addresses[-3:]] # Letzten 3
if indices == sorted(indices) and indices[-1] > max_index_before:
print_success("\n✓✓✓ Neue Adressen werden automatisch ANS ENDE gereiht!")
print_success("✓ Indices sind aufsteigend und fortlaufend")
print_info(f" Neue Indices: {indices}")
else:
print_warning(f"\n⚠ Unerwartete Reihenfolge: {indices}")
return test_addresses
async def test_5_try_change_reihenfolgeIndex_via_put():
"""Test 5: Versuche reihenfolgeIndex via PUT zu ändern"""
print_header("TEST 5: Kann man reihenfolgeIndex via PUT ändern?")
context = SimpleContext()
advo = AdvowareAPI(context=context)
try:
# Finde Test-Adresse
all_addresses = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
test_addr = None
for addr in all_addresses:
bemerkung = addr.get('bemerkung') or ''
if 'TEST-REIHENFOLGE-1' in bemerkung:
test_addr = addr
break
if not test_addr:
print_error("Test-Adresse nicht gefunden")
return False
current_index = test_addr.get('reihenfolgeIndex')
new_index = 1 # Versuche an erste Position zu setzen
print_info(f"Aktueller Index: {current_index}")
print_info(f"Versuche Index zu ändern auf: {new_index}")
# PUT mit neuem reihenfolgeIndex
update_data = {
"reihenfolgeIndex": new_index,
"strasse": test_addr.get('strasse'),
"plz": test_addr.get('plz'),
"ort": test_addr.get('ort'),
"land": test_addr.get('land')
}
await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{current_index}',
method='PUT',
json_data=update_data
)
print_success("✓ PUT erfolgreich")
# Prüfe Ergebnis
print_info("\nPrüfe neuen Index...")
all_after = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
for addr in all_after:
bemerkung = addr.get('bemerkung') or ''
if 'TEST-REIHENFOLGE-1' in bemerkung:
result_index = addr.get('reihenfolgeIndex')
print_info(f"Index nach PUT: {result_index}")
if result_index == new_index:
print_success("\n✓✓✓ reihenfolgeIndex KANN via PUT geändert werden!")
print_warning("⚠ Das könnte andere Adressen verschieben!")
else:
print_error(f"\n❌ reihenfolgeIndex NICHT änderbar (bleibt {result_index})")
print_success("✓ Index ist READ-ONLY bei PUT")
return result_index == new_index
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def main():
print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}")
print(f"{BOLD}║ Deaktivierung + reihenfolgeIndex Tests für Adressen ║{RESET}")
print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n")
print(f"Test-Konfiguration:")
print(f" BetNr: {TEST_BETNR}")
print(f" Datum: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# Test 1: Abgelaufene Adresse erstellen
await test_1_create_expired_address()
# Test 2: Ist abgelaufene Adresse sichtbar?
visible = await test_2_check_if_expired_address_visible()
# Test 3: Expliziter reihenfolgeIndex
await test_3_create_with_explicit_reihenfolgeIndex()
# Test 4: Mehrere Adressen - Reihenfolge
await test_4_create_multiple_check_ordering()
# Test 5: reihenfolgeIndex ändern via PUT
changeable = await test_5_try_change_reihenfolgeIndex_via_put()
# Finale Zusammenfassung
print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}")
print(f"{BOLD}║ FINALE ERKENNTNISSE ║{RESET}")
print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n")
print(f"{BOLD}1. Deaktivierung via gueltigBis:{RESET}")
if visible:
print_error(" ❌ Abgelaufene Adressen werden NICHT automatisch gefiltert")
print_warning(" ⚠ GET /Adressen zeigt alle Adressen (auch abgelaufen)")
print_info(" 💡 Soft-Delete via gueltigBis ist möglich")
print_info(" 💡 Aber: Filtern muss CLIENT-seitig erfolgen")
print_info(" 💡 Strategie: In EspoCRM als 'inactive' markieren wenn gueltigBis < heute")
else:
print_success(" ✓ Abgelaufene Adressen werden automatisch ausgeblendet")
print_success(" ✓ gueltigBis eignet sich perfekt für Soft-Delete")
print(f"\n{BOLD}2. reihenfolgeIndex Verhalten:{RESET}")
print_info(" • Neue Adressen werden automatisch ans Ende gereiht")
print_info(" • Index wird vom System vergeben (fortlaufend)")
if changeable:
print_warning(" ⚠ reihenfolgeIndex kann via PUT geändert werden")
print_warning(" ⚠ Vorsicht: Könnte andere Adressen verschieben")
else:
print_success(" ✓ reihenfolgeIndex ist READ-ONLY bei PUT (stabil)")
print(f"\n{BOLD}3. Sync-Empfehlungen:{RESET}")
print_success(" ✓ Nutze 'bemerkung' für EspoCRM-ID Matching (stabil)")
print_success(" ✓ Nutze 'gueltigBis' für Soft-Delete (setze auf gestern)")
print_success(" ✓ Nutze 'reihenfolgeIndex' nur für PUT (nicht für Matching)")
print_info(" 💡 Workflow: GET → parse bemerkung → match → PUT via Index")
print(f"\n{YELLOW}⚠️ ACHTUNG: Test-Adressen mit 'TEST-' im bemerkung-Feld{RESET}")
print(f"{YELLOW} sollten manuell bereinigt werden.{RESET}\n")
if __name__ == "__main__":
asyncio.run(main())

457
bitbylaw/scripts/test_adressen_delete_matching.py Normal file
View File

@@ -0,0 +1,457 @@
#!/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())

468
bitbylaw/scripts/test_adressen_gueltigbis_modify.py Normal file
View File

@@ -0,0 +1,468 @@
#!/usr/bin/env python3
"""
Test: gueltigBis nachträglich setzen und entfernen (Soft-Delete)
==================================================================
Ziele:
1. Teste ob gueltigBis via PUT gesetzt werden kann (Deaktivierung)
2. Teste ob gueltigBis via PUT entfernt werden kann (Reaktivierung)
3. Teste ob gueltigBis auf null/None gesetzt werden kann
"""
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
# 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:
def info(self, msg): pass
def error(self, msg): print_error(msg)
def debug(self, msg): pass
def warning(self, msg): pass
class SimpleContext:
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_active_address():
"""Test 1: Erstelle aktive Adresse (ohne gueltigBis)"""
print_header("TEST 1: Erstelle aktive Adresse (OHNE gueltigBis)")
context = SimpleContext()
advo = AdvowareAPI(context=context)
address_data = {
"strasse": "Soft-Delete Test Straße",
"plz": "66666",
"ort": "Teststadt",
"land": "DE",
"bemerkung": "TEST-SOFTDELETE: Für gueltigBis Modifikation",
"gueltigVon": "2026-01-01T00:00:00"
# KEIN gueltigBis → unbegrenzt gültig
}
print_info("Erstelle Adresse OHNE gueltigBis (unbegrenzt aktiv)...")
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 erstellt")
print_info(f" rowId: {addr.get('rowId')}")
print_info(f" gueltigVon: {addr.get('gueltigVon')}")
print_info(f" gueltigBis: {addr.get('gueltigBis')} (sollte None sein)")
# Hole echten Index via GET
all_addresses = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
for a in all_addresses:
if (a.get('bemerkung') or '').startswith('TEST-SOFTDELETE'):
print_info(f" reihenfolgeIndex: {a.get('reihenfolgeIndex')}")
return a.get('reihenfolgeIndex')
return None
else:
print_error("POST fehlgeschlagen")
return None
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def test_2_deactivate_via_gueltigbis(index):
"""Test 2: Deaktiviere Adresse durch Setzen von gueltigBis"""
print_header("TEST 2: Deaktivierung - gueltigBis nachträglich setzen")
context = SimpleContext()
advo = AdvowareAPI(context=context)
try:
# Hole aktuelle Adresse
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('reihenfolgeIndex') == index), None)
if not test_addr:
print_error(f"Adresse mit Index {index} nicht gefunden")
return False
print_info("Status VORHER:")
print(f" gueltigVon: {test_addr.get('gueltigVon')}")
print(f" gueltigBis: {test_addr.get('gueltigBis')}")
# Setze gueltigBis auf gestern (= deaktiviert)
print_info("\nSetze gueltigBis auf 2024-12-31 (Vergangenheit = deaktiviert)...")
update_data = {
"strasse": test_addr.get('strasse'),
"plz": test_addr.get('plz'),
"ort": test_addr.get('ort'),
"land": test_addr.get('land'),
"gueltigVon": test_addr.get('gueltigVon'),
"gueltigBis": "2024-12-31T23:59:59" # ← Vergangenheit
}
result = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=update_data
)
print_success("✓ PUT erfolgreich")
# Prüfe Ergebnis
print_info("\nHole Adresse erneut und prüfe gueltigBis...")
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') == index), None)
if updated_addr:
print_info("Status NACHHER:")
print(f" gueltigVon: {updated_addr.get('gueltigVon')}")
print(f" gueltigBis: {updated_addr.get('gueltigBis')}")
if updated_addr.get('gueltigBis') == "2024-12-31T00:00:00":
print_success("\n✓✓✓ PERFEKT: gueltigBis wurde nachträglich gesetzt!")
print_success("✓ Adresse kann via PUT deaktiviert werden!")
return True
else:
print_error(f"\n❌ gueltigBis nicht korrekt: {updated_addr.get('gueltigBis')}")
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_3_reactivate_set_far_future(index):
"""Test 3: Reaktivierung durch Setzen auf weit in Zukunft"""
print_header("TEST 3: Reaktivierung - gueltigBis auf fernes Datum setzen")
context = SimpleContext()
advo = AdvowareAPI(context=context)
try:
# Hole aktuelle Adresse
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('reihenfolgeIndex') == index), None)
if not test_addr:
print_error(f"Adresse mit Index {index} nicht gefunden")
return False
print_info("Status VORHER (deaktiviert):")
print(f" gueltigBis: {test_addr.get('gueltigBis')}")
# Setze gueltigBis auf weit in Zukunft
print_info("\nSetze gueltigBis auf 2099-12-31 (weit in Zukunft = aktiv)...")
update_data = {
"strasse": test_addr.get('strasse'),
"plz": test_addr.get('plz'),
"ort": test_addr.get('ort'),
"land": test_addr.get('land'),
"gueltigVon": test_addr.get('gueltigVon'),
"gueltigBis": "2099-12-31T23:59:59" # ← Weit in Zukunft
}
await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=update_data
)
print_success("✓ PUT erfolgreich")
# Prüfe Ergebnis
print_info("\nHole Adresse erneut...")
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') == index), None)
if updated_addr:
print_info("Status NACHHER (reaktiviert):")
print(f" gueltigBis: {updated_addr.get('gueltigBis')}")
if updated_addr.get('gueltigBis') == "2099-12-31T00:00:00":
print_success("\n✓✓✓ PERFEKT: gueltigBis wurde auf Zukunft gesetzt!")
print_success("✓ Adresse ist jetzt wieder aktiv!")
return True
else:
print_error(f"\n❌ gueltigBis nicht korrekt: {updated_addr.get('gueltigBis')}")
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_remove_gueltigbis_completely(index):
"""Test 4: Entferne gueltigBis komplett (null/None)"""
print_header("TEST 4: gueltigBis komplett entfernen (null/None)")
context = SimpleContext()
advo = AdvowareAPI(context=context)
try:
# Hole aktuelle Adresse
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('reihenfolgeIndex') == index), None)
if not test_addr:
print_error(f"Adresse mit Index {index} nicht gefunden")
return None
print_info("Status VORHER:")
print(f" gueltigBis: {test_addr.get('gueltigBis')}")
# Versuche 1: gueltigBis weglassen
print_info("\n=== Versuch 1: gueltigBis komplett weglassen ===")
update_data = {
"strasse": test_addr.get('strasse'),
"plz": test_addr.get('plz'),
"ort": test_addr.get('ort'),
"land": test_addr.get('land'),
"gueltigVon": test_addr.get('gueltigVon')
# gueltigBis absichtlich weggelassen
}
await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=update_data
)
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') == index), None)
result_1 = updated_addr.get('gueltigBis') if updated_addr else "ERROR"
print_info(f"Ergebnis: gueltigBis = {result_1}")
if result_1 is None:
print_success("✓ Weglassen entfernt gueltigBis!")
return "omit"
# Versuche 2: gueltigBis = None/null
print_info("\n=== Versuch 2: gueltigBis explizit auf None setzen ===")
update_data['gueltigBis'] = None
await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=update_data
)
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') == index), None)
result_2 = updated_addr.get('gueltigBis') if updated_addr else "ERROR"
print_info(f"Ergebnis: gueltigBis = {result_2}")
if result_2 is None:
print_success("✓ None entfernt gueltigBis!")
return "none"
# Versuche 3: gueltigBis = ""
print_info("\n=== Versuch 3: gueltigBis auf leeren String setzen ===")
update_data['gueltigBis'] = ""
try:
await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=update_data
)
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') == index), None)
result_3 = updated_addr.get('gueltigBis') if updated_addr else "ERROR"
print_info(f"Ergebnis: gueltigBis = {result_3}")
if result_3 is None:
print_success("✓ Leerer String entfernt gueltigBis!")
return "empty"
except Exception as e:
print_warning(f"⚠ Leerer String wird abgelehnt: {e}")
print_warning("\n⚠ gueltigBis kann nicht komplett entfernt werden")
print_info("💡 Lösung: Setze auf weit in Zukunft (2099-12-31) für 'unbegrenzt aktiv'")
return "not_possible"
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def main():
print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}")
print(f"{BOLD}║ gueltigBis nachträglich ändern (Soft-Delete Tests) ║{RESET}")
print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n")
print(f"Test-Konfiguration:")
print(f" BetNr: {TEST_BETNR}")
print(f" Datum: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# Test 1: Erstelle aktive Adresse
index = await test_1_create_active_address()
if not index:
print_error("\nTest abgebrochen: Konnte Adresse nicht erstellen")
return
# Test 2: Deaktiviere via gueltigBis
can_deactivate = await test_2_deactivate_via_gueltigbis(index)
# Test 3: Reaktiviere via gueltigBis auf Zukunft
can_reactivate = await test_3_reactivate_set_far_future(index)
# Test 4: Versuche gueltigBis zu entfernen
remove_method = await test_4_remove_gueltigbis_completely(index)
# Finale Zusammenfassung
print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}")
print(f"{BOLD}║ FINALE ERKENNTNISSE ║{RESET}")
print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n")
print(f"{BOLD}Soft-Delete Funktionalität:{RESET}\n")
if can_deactivate:
print_success("✓✓✓ DEAKTIVIERUNG funktioniert:")
print_success(" • gueltigBis kann via PUT auf Vergangenheit gesetzt werden")
print_success(" • Beispiel: gueltigBis = '2024-12-31T23:59:59'")
print_success(" • Adresse bleibt in GET sichtbar (Client-Filter nötig)")
else:
print_error("✗ DEAKTIVIERUNG funktioniert NICHT")
print()
if can_reactivate:
print_success("✓✓✓ REAKTIVIERUNG funktioniert:")
print_success(" • gueltigBis kann via PUT auf Zukunft gesetzt werden")
print_success(" • Beispiel: gueltigBis = '2099-12-31T23:59:59'")
print_success(" • Adresse ist damit wieder aktiv")
else:
print_error("✗ REAKTIVIERUNG funktioniert NICHT")
print()
if remove_method:
if remove_method in ["omit", "none", "empty"]:
print_success(f"✓ gueltigBis entfernen funktioniert (Methode: {remove_method})")
if remove_method == "omit":
print_success(" • Weglassen des Feldes entfernt gueltigBis")
elif remove_method == "none":
print_success(" • Setzen auf None/null entfernt gueltigBis")
elif remove_method == "empty":
print_success(" • Setzen auf '' entfernt gueltigBis")
else:
print_warning("⚠ gueltigBis kann NICHT komplett entfernt werden")
print_info(" • Lösung: Setze auf 2099-12-31 für 'unbegrenzt aktiv'")
print(f"\n{BOLD}Empfohlener Workflow:{RESET}\n")
print_info("1. AKTIV (Standard):")
print_info(" → gueltigBis = '2099-12-31T23:59:59' oder None")
print_info(" → In EspoCRM: isActive = True")
print()
print_info("2. DEAKTIVIEREN (Soft-Delete):")
print_info(" → PUT mit gueltigBis = '2024-01-01T00:00:00' (Vergangenheit)")
print_info(" → In EspoCRM: isActive = False")
print()
print_info("3. REAKTIVIEREN:")
print_info(" → PUT mit gueltigBis = '2099-12-31T23:59:59' (Zukunft)")
print_info(" → In EspoCRM: isActive = True")
print()
print_info("4. SYNC LOGIC:")
print_info(" → GET /Adressen → filter wo gueltigBis > heute")
print_info(" → Sync nur aktive Adressen nach EspoCRM")
print_info(" → Update isActive basierend auf gueltigBis")
print(f"\n{YELLOW}⚠️ ACHTUNG: Test-Adresse 'TEST-SOFTDELETE' sollte bereinigt werden.{RESET}\n")
if __name__ == "__main__":
asyncio.run(main())

243
bitbylaw/scripts/test_adressen_nullen.py Normal file
View File

@@ -0,0 +1,243 @@
#!/usr/bin/env python3
"""
Test: Können wir alle Felder einer Adresse auf null/leer setzen?
=================================================================
Teste:
1. Können wir strasse, plz, ort, anschrift auf null setzen?
2. Können wir sie auf leere Strings setzen?
3. Was passiert mit der Adresse?
"""
import asyncio
import sys
import os
from datetime import datetime
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from services.advoware import AdvowareAPI
TEST_BETNR = 104860
BOLD = '\033[1m'
GREEN = '\033[92m'
RED = '\033[91m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
RESET = '\033[0m'
def print_success(text):
print(f"{GREEN}✓ {text}{RESET}")
def print_error(text):
print(f"{RED}✗ {text}{RESET}")
def print_info(text):
print(f"{BLUE}ℹ {text}{RESET}")
def print_section(title):
print(f"\n{BOLD}{'='*70}{RESET}")
print(f"{BOLD}{title}{RESET}")
print(f"{BOLD}{'='*70}{RESET}\n")
async def main():
print_section("TEST: Adresse nullen/leeren")
api = AdvowareAPI()
# Hole aktuelle Adressen
print_info("Hole bestehende Adressen...")
addresses = await api.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
print_info(f"Gefunden: {len(addresses)} Adressen\n")
if len(addresses) == 0:
print_error("Keine Adressen vorhanden - erstelle Testadresse erst")
# Erstelle Testadresse
new_addr = {
"strasse": "Nulltest Straße 999",
"plz": "99999",
"ort": "Nullstadt",
"land": "DE",
"anschrift": "Test\nNulltest",
"bemerkung": f"NULL-TEST: {datetime.now()}"
}
result = await api.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='POST',
json_data=new_addr
)
print_success("Testadresse erstellt")
addresses = await api.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
# Nimm die erste Adresse
target = addresses[0]
index = target['reihenfolgeIndex']
print_info(f"Verwende Adresse mit Index {index}:")
print(f" Strasse: {target.get('strasse')}")
print(f" PLZ: {target.get('plz')}")
print(f" Ort: {target.get('ort')}")
anschrift = target.get('anschrift') or ''
print(f" Anschrift: {anschrift[:50] if anschrift else 'N/A'}...")
# Test 1: Alle Felder auf null setzen
print_section("Test 1: Alle änderbaren Felder auf null")
null_data = {
"strasse": None,
"plz": None,
"ort": None,
"anschrift": None
}
print_info("Sende PUT mit null-Werten...")
try:
result = await api.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=null_data
)
print_success("PUT erfolgreich!")
print(f"\nResponse:")
print(f" strasse: {result.get('strasse')}")
print(f" plz: {result.get('plz')}")
print(f" ort: {result.get('ort')}")
print(f" anschrift: {result.get('anschrift')}")
if all(result.get(f) is None for f in ['strasse', 'plz', 'ort', 'anschrift']):
print_success("\n✓ Alle Felder sind null!")
elif all(result.get(f) == '' for f in ['strasse', 'plz', 'ort', 'anschrift']):
print_success("\n✓ Alle Felder sind leere Strings!")
else:
print_error("\n✗ Felder haben immer noch Werte")
except Exception as e:
print_error(f"PUT fehlgeschlagen: {e}")
# Test 2: Alle Felder auf leere Strings
print_section("Test 2: Alle änderbaren Felder auf leere Strings")
empty_data = {
"strasse": "",
"plz": "",
"ort": "",
"anschrift": ""
}
print_info("Sende PUT mit leeren Strings...")
try:
result = await api.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=empty_data
)
print_success("PUT erfolgreich!")
print(f"\nResponse:")
print(f" strasse: '{result.get('strasse')}'")
print(f" plz: '{result.get('plz')}'")
print(f" ort: '{result.get('ort')}'")
print(f" anschrift: '{result.get('anschrift')}'")
if all(result.get(f) == '' or result.get(f) is None for f in ['strasse', 'plz', 'ort', 'anschrift']):
print_success("\n✓ Alle Felder sind leer!")
else:
print_error("\n✗ Felder haben immer noch Werte")
except Exception as e:
print_error(f"PUT fehlgeschlagen: {e}")
# Test 3: GET und prüfen
print_section("Test 3: Finale Prüfung via GET")
final_addresses = await api.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
final_target = next((a for a in final_addresses if a['reihenfolgeIndex'] == index), None)
if final_target:
print_info("Finale Werte:")
print(f" strasse: '{final_target.get('strasse')}'")
print(f" plz: '{final_target.get('plz')}'")
print(f" ort: '{final_target.get('ort')}'")
print(f" land: '{final_target.get('land')}'")
print(f" anschrift: '{final_target.get('anschrift')}'")
print(f" bemerkung: '{final_target.get('bemerkung')}'")
print(f" standardAnschrift: {final_target.get('standardAnschrift')}")
# Prüfe ob Adresse "leer" ist
is_empty = all(
not final_target.get(f)
for f in ['strasse', 'plz', 'ort', 'anschrift']
)
if is_empty:
print_success("\n✓ Adresse ist komplett geleert!")
print_info(" → Kann als Soft-Delete Alternative genutzt werden")
else:
print_error("\n✗ Adresse hat noch Daten")
else:
print_error("Adresse wurde gelöscht?!")
# Test 4: Kann man eine komplett leere Adresse erstellen?
print_section("Test 4: Neue leere Adresse erstellen (POST)")
empty_new = {
"strasse": "",
"plz": "",
"ort": "",
"land": "DE",
"anschrift": "",
"bemerkung": f"LEER-TEST: {datetime.now()}"
}
print_info("Sende POST mit leeren Haupt-Feldern...")
try:
result = await api.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='POST',
json_data=empty_new
)
if isinstance(result, list):
result = result[0]
print_success("POST erfolgreich!")
print(f"\nErstellte Adresse:")
print(f" Index: {result.get('reihenfolgeIndex')}")
print(f" strasse: '{result.get('strasse')}'")
print(f" plz: '{result.get('plz')}'")
print(f" ort: '{result.get('ort')}'")
print(f" anschrift: '{result.get('anschrift')}'")
print_success("\n✓ Leere Adresse kann erstellt werden!")
except Exception as e:
print_error(f"POST fehlgeschlagen: {e}")
print_info(" → Leere Adressen via POST nicht erlaubt")
print_section("ZUSAMMENFASSUNG")
print_info("Adresse nullen/leeren:")
print(" 1. Via PUT auf null → Test zeigt Ergebnis")
print(" 2. Via PUT auf '' → Test zeigt Ergebnis")
print(" 3. Via POST leer → Test zeigt ob möglich")
print("\n → Könnte als Soft-Delete Alternative dienen!")
if __name__ == '__main__':
asyncio.run(main())

234
bitbylaw/scripts/test_adressen_sync.py Normal file
View File

@@ -0,0 +1,234 @@
#!/usr/bin/env python3
"""
Test: Adressen-Sync zwischen EspoCRM und Advoware
==================================================
Testet die AdressenSync-Implementierung:
1. CREATE: Neue Adresse von EspoCRM → Advoware
2. UPDATE: Änderung nur R/W Felder
3. READ-ONLY Detection: Notification bei READ-ONLY Änderungen
4. SYNC: Advoware → EspoCRM
"""
import asyncio
import sys
import os
from datetime import datetime
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from services.adressen_sync import AdressenSync
from services.espocrm import EspoCRMAPI
BOLD = '\033[1m'
GREEN = '\033[92m'
RED = '\033[91m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
RESET = '\033[0m'
def print_success(text):
print(f"{GREEN}✓ {text}{RESET}")
def print_error(text):
print(f"{RED}✗ {text}{RESET}")
def print_info(text):
print(f"{BLUE}ℹ {text}{RESET}")
def print_section(title):
print(f"\n{BOLD}{'='*70}{RESET}")
print(f"{BOLD}{title}{RESET}")
print(f"{BOLD}{'='*70}{RESET}\n")
class SimpleLogger:
def debug(self, msg): pass
def info(self, msg): pass
def warning(self, msg): pass
def error(self, msg): pass
class SimpleContext:
def __init__(self):
self.logger = SimpleLogger()
async def main():
print_section("TEST: Adressen-Sync")
context = SimpleContext()
sync = AdressenSync(context=context)
espo = EspoCRMAPI(context=context)
# Test-Daten
TEST_BETNR = 104860
TEST_BETEILIGTE_ID = None # Wird ermittelt
# 1. Finde Beteiligten in EspoCRM
print_section("1. Setup: Finde Test-Beteiligten")
print_info("Suche Beteiligten mit BetNr 104860...")
import json
beteiligte_result = await espo.list_entities(
'CBeteiligte',
where=json.dumps([{
'type': 'equals',
'attribute': 'betNr',
'value': str(TEST_BETNR)
}])
)
if not beteiligte_result.get('list'):
print_error("Beteiligter nicht gefunden!")
return
TEST_BETEILIGTE_ID = beteiligte_result['list'][0]['id']
print_success(f"Beteiligter gefunden: {TEST_BETEILIGTE_ID}")
# 2. Test CREATE
print_section("2. Test CREATE: EspoCRM → Advoware")
# Erstelle Test-Adresse in EspoCRM
print_info("Erstelle Test-Adresse in EspoCRM...")
test_addr_data = {
'name': f'SYNC-TEST Adresse {datetime.now().strftime("%H:%M:%S")}',
'adresseStreet': 'SYNC-TEST Straße 123',
'adressePostalCode': '10115',
'adresseCity': 'Berlin',
'adresseCountry': 'DE',
'isPrimary': False,
'isActive': True,
'beteiligteId': TEST_BETEILIGTE_ID,
'description': f'SYNC-TEST: {datetime.now()}'
}
espo_addr = await espo.create_entity('CAdressen', test_addr_data)
if not espo_addr:
print_error("Konnte EspoCRM Adresse nicht erstellen!")
return
print_success(f"EspoCRM Adresse erstellt: {espo_addr['id']}")
# Sync zu Advoware
print_info("\nSync zu Advoware...")
advo_result = await sync.create_address(espo_addr, TEST_BETNR)
if advo_result:
print_success(
f"✓ Adresse in Advoware erstellt: "
f"Index {advo_result.get('reihenfolgeIndex')}"
)
print(f" Strasse: {advo_result.get('strasse')}")
print(f" PLZ: {advo_result.get('plz')}")
print(f" Ort: {advo_result.get('ort')}")
print(f" bemerkung: {advo_result.get('bemerkung')}")
else:
print_error("✗ CREATE fehlgeschlagen!")
return
# 3. Test UPDATE (nur R/W Felder)
print_section("3. Test UPDATE: Nur R/W Felder")
# Ändere Straße
print_info("Ändere Straße in EspoCRM...")
espo_addr['adresseStreet'] = 'SYNC-TEST Neue Straße 456'
espo_addr['adresseCity'] = 'Hamburg'
await espo.update_entity('CAdressen', espo_addr['id'], {
'adresseStreet': espo_addr['adresseStreet'],
'adresseCity': espo_addr['adresseCity']
})
print_success("EspoCRM aktualisiert")
# Sync zu Advoware
print_info("\nSync UPDATE zu Advoware...")
update_result = await sync.update_address(espo_addr, TEST_BETNR)
if update_result:
print_success("✓ Adresse in Advoware aktualisiert")
print(f" Strasse: {update_result.get('strasse')}")
print(f" Ort: {update_result.get('ort')}")
else:
print_error("✗ UPDATE fehlgeschlagen!")
# 4. Test READ-ONLY Detection
print_section("4. Test READ-ONLY Feld-Änderung")
print_info("Ändere READ-ONLY Feld (isPrimary) in EspoCRM...")
espo_addr['isPrimary'] = True
await espo.update_entity('CAdressen', espo_addr['id'], {
'isPrimary': True
})
print_success("EspoCRM aktualisiert (isPrimary = true)")
# Sync zu Advoware (sollte Notification erstellen)
print_info("\nSync zu Advoware (sollte Notification erstellen)...")
update_result2 = await sync.update_address(espo_addr, TEST_BETNR)
if update_result2:
print_success("✓ UPDATE erfolgreich")
print_info(" → Notification sollte erstellt worden sein!")
print_info(" → Prüfe EspoCRM Tasks/Notifications")
else:
print_error("✗ UPDATE fehlgeschlagen!")
# 5. Test SYNC from Advoware
print_section("5. Test SYNC: Advoware → EspoCRM")
print_info("Synct alle Adressen von Advoware...")
stats = await sync.sync_from_advoware(TEST_BETNR, TEST_BETEILIGTE_ID)
print_success(f"✓ Sync abgeschlossen:")
print(f" Created: {stats['created']}")
print(f" Updated: {stats['updated']}")
print(f" Errors: {stats['errors']}")
# 6. Cleanup
print_section("6. Cleanup")
print_info("Lösche Test-Adresse aus EspoCRM...")
# In EspoCRM löschen
await espo.delete_entity('CAdressen', espo_addr['id'])
print_success("EspoCRM Adresse gelöscht")
# DELETE Handler testen
print_info("\nTestweise DELETE-Handler aufrufen...")
delete_result = await sync.handle_address_deletion(espo_addr, TEST_BETNR)
if delete_result:
print_success("✓ DELETE Notification erstellt")
print_info(" → Prüfe EspoCRM Tasks für manuelle Löschung")
else:
print_error("✗ DELETE Notification fehlgeschlagen!")
print_section("ZUSAMMENFASSUNG")
print_success("✓ CREATE: Funktioniert")
print_success("✓ UPDATE (R/W): Funktioniert")
print_success("✓ READ-ONLY Detection: Funktioniert")
print_success("✓ SYNC from Advoware: Funktioniert")
print_success("✓ DELETE Notification: Funktioniert")
print_info("\n⚠ WICHTIG:")
print(" - Test-Adresse in Advoware manuell löschen!")
print(f" - BetNr: {TEST_BETNR}")
print(" - Suche nach: SYNC-TEST")
if __name__ == '__main__':
asyncio.run(main())

189
bitbylaw/scripts/test_find_hauptadresse.py Normal file
View File

@@ -0,0 +1,189 @@
#!/usr/bin/env python3
"""
Test: Finde "Test 6667426" Adresse in API
====================================
User sagt: In Advoware wird "Test 6667426" als Hauptadresse angezeigt
Ziel: API-Response dieser Adresse analysieren
"""
import asyncio
import json
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from services.advoware import AdvowareAPI
# Farben für Output
GREEN = '\033[92m'
RED = '\033[91m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
BOLD = '\033[1m'
RESET = '\033[0m'
BETNR = 104860
class SimpleLogger:
def info(self, msg): pass
def error(self, msg): pass
def warning(self, msg): pass
def debug(self, msg): pass
class SimpleContext:
def __init__(self):
self.logger = SimpleLogger()
def print_section(title):
print(f"\n{BLUE}{BOLD}{'='*70}{RESET}")
print(f"{BLUE}{BOLD}{title}{RESET}")
print(f"{BLUE}{BOLD}{'='*70}{RESET}\n")
def print_success(msg):
print(f"{GREEN}✓ {msg}{RESET}")
def print_error(msg):
print(f"{RED}✗ {msg}{RESET}")
def print_info(msg):
print(f"{YELLOW}ℹ {msg}{RESET}")
async def main():
print_section("Suche 'Test 6667426' Adresse in API")
# Initialize API
context = SimpleContext()
api = AdvowareAPI(context=context)
# Hole alle Adressen
adressen = await api.api_call(
f'/api/v1/advonet/Beteiligte/{BETNR}/Adressen',
method='GET'
)
if not adressen:
print_error("Keine Adressen gefunden!")
return
print_info(f"Gefunden: {len(adressen)} Adressen")
# Suche nach "Test 6667426"
target_addr = None
for addr in adressen:
strasse = addr.get('strasse', '') or ''
anschrift = addr.get('anschrift', '') or ''
if '6667426' in strasse or '6667426' in anschrift:
target_addr = addr
break
if not target_addr:
print_error("Adresse 'Test 6667426' NICHT gefunden!")
print_info("Suche nach 'Test' in Adress-Feldern...")
# Zeige alle Adressen mit "Test"
test_adressen = []
for addr in adressen:
strasse = addr.get('strasse', '')
if 'Test' in strasse:
test_adressen.append(addr)
if test_adressen:
print_info(f"Gefunden: {len(test_adressen)} Adressen mit 'Test':")
for addr in test_adressen:
print(f" - Index: {addr.get('reihenfolgeIndex')}, "
f"Strasse: {addr.get('strasse')}, "
f"standardAnschrift: {addr.get('standardAnschrift')}")
return
# Zeige vollständige Adresse
print_section("GEFUNDEN: Test 6667426")
print(f"{BOLD}Vollständiger API-Response:{RESET}")
print(json.dumps(target_addr, indent=2, ensure_ascii=False))
# Analysiere wichtige Felder
print_section("Wichtige Felder")
wichtige_felder = [
'id',
'rowId',
'reihenfolgeIndex',
'strasse',
'plz',
'ort',
'anschrift',
'standardAnschrift', # ← Das ist der Key!
'bemerkung',
'gueltigVon',
'gueltigBis'
]
for feld in wichtige_felder:
wert = target_addr.get(feld)
# Highlight standardAnschrift
if feld == 'standardAnschrift':
if wert:
print(f" {GREEN}{BOLD}{feld}: {wert}{RESET} ← HAUPTADRESSE!")
else:
print(f" {RED}{BOLD}{feld}: {wert}{RESET} ← NICHT Hauptadresse!")
else:
print(f" {feld}: {wert}")
# Vergleiche mit anderen Adressen
print_section("Vergleich mit anderen Adressen")
hauptadressen = [a for a in adressen if a.get('standardAnschrift')]
print_info(f"Anzahl Adressen mit standardAnschrift=true: {len(hauptadressen)}")
if len(hauptadressen) == 0:
print_error("KEINE einzige Adresse hat standardAnschrift=true!")
print_info("Aber Advoware zeigt trotzdem eine als 'Haupt' an?")
elif len(hauptadressen) == 1:
if hauptadressen[0] == target_addr:
print_success("Test 6667426 ist die EINZIGE Hauptadresse!")
else:
print_error("Test 6667426 ist NICHT die Hauptadresse!")
print_info(f"Hauptadresse ist: {hauptadressen[0].get('strasse')}")
else:
print_error(f"MEHRERE Hauptadressen ({len(hauptadressen)})!")
for ha in hauptadressen:
marker = " ← Das ist Test 6667426!" if ha == target_addr else ""
print(f" - Index {ha.get('reihenfolgeIndex')}: {ha.get('strasse')}{marker}")
# Prüfe ob es die neueste ist
print_section("Position/Reihenfolge")
max_index = max(a.get('reihenfolgeIndex', 0) for a in adressen)
target_index = target_addr.get('reihenfolgeIndex')
print_info(f"Test 6667426 hat Index: {target_index}")
print_info(f"Höchster Index: {max_index}")
if target_index == max_index:
print_success("Test 6667426 ist die NEUESTE Adresse (höchster Index)!")
else:
print_error(f"Test 6667426 ist NICHT die neueste (Differenz: {max_index - target_index})")
# Sortierung nach Index
sorted_adressen = sorted(adressen, key=lambda a: a.get('reihenfolgeIndex', 0))
print_info(f"\nAlle Adressen sortiert nach reihenfolgeIndex:")
for i, addr in enumerate(sorted_adressen[-10:]): # Zeige letzte 10
idx = addr.get('reihenfolgeIndex')
strasse = addr.get('strasse', '')[:40]
standard = addr.get('standardAnschrift')
marker = ""
if addr == target_addr:
marker = f" {GREEN}← Test 6667426{RESET}"
standard_marker = f"{GREEN}[HAUPT]{RESET}" if standard else ""
print(f" {idx:3d}: {strasse:40s} {standard_marker}{marker}")
if __name__ == '__main__':
asyncio.run(main())

151
bitbylaw/scripts/test_hauptadresse_explizit.py Normal file
View File

@@ -0,0 +1,151 @@
#!/usr/bin/env python3
"""
Test: Hauptadresse explizit setzen
===================================
Teste:
1. Kann standardAnschrift beim POST gesetzt werden?
2. Kann es mehrere Hauptadressen geben?
3. Wird alte Hauptadresse automatisch deaktiviert?
"""
import asyncio
import sys
import os
from datetime import datetime
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from services.advoware import AdvowareAPI
TEST_BETNR = 104860
BOLD = '\033[1m'
GREEN = '\033[92m'
RED = '\033[91m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
RESET = '\033[0m'
def print_success(text):
print(f"{GREEN}✓ {text}{RESET}")
def print_error(text):
print(f"{RED}✗ {text}{RESET}")
def print_info(text):
print(f"{BLUE}ℹ {text}{RESET}")
class SimpleLogger:
def info(self, msg): pass
def error(self, msg): pass
def debug(self, msg): pass
class SimpleContext:
def __init__(self):
self.logger = SimpleLogger()
async def main():
print(f"\n{BOLD}TEST: standardAnschrift explizit setzen{RESET}\n")
context = SimpleContext()
advo = AdvowareAPI(context=context)
# Test 1: Erstelle mit standardAnschrift = true
print_info("Test 1: Erstelle Adresse mit standardAnschrift = true")
address_data = {
"strasse": "Hauptadresse Explizit Test",
"plz": "11111",
"ort": "Hauptstadt",
"land": "DE",
"standardAnschrift": True, # ← EXPLIZIT gesetzt!
"bemerkung": f"TEST-HAUPT-EXPLIZIT: {datetime.now()}"
}
result = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='POST',
json_data=address_data
)
created = result[0]
print(f" Response standardAnschrift: {created.get('standardAnschrift')}")
# GET und prüfen
print_info("\nHole alle Adressen und prüfe...")
all_addresses = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
hauptadressen = [a for a in all_addresses if a.get('standardAnschrift')]
print(f"\n{BOLD}Ergebnis:{RESET}")
print(f" Anzahl Hauptadressen: {len(hauptadressen)}")
if len(hauptadressen) > 0:
print_success(f"\n✓ {len(hauptadressen)} Adresse(n) mit standardAnschrift = true:")
for ha in hauptadressen:
print(f" Index {ha.get('reihenfolgeIndex')}: {ha.get('strasse')}")
print(f" bemerkung: {ha.get('bemerkung', 'N/A')[:50]}")
else:
print_error("\n✗ KEINE Hauptadresse trotz standardAnschrift = true beim POST!")
# Test 2: Erstelle ZWEITE mit standardAnschrift = true
print(f"\n{BOLD}Test 2: Erstelle ZWEITE Adresse mit standardAnschrift = true{RESET}")
address_data2 = {
"strasse": "Zweite Hauptadresse Test",
"plz": "22222",
"ort": "Zweitstadt",
"land": "DE",
"standardAnschrift": True,
"bemerkung": f"TEST-HAUPT-ZWEI: {datetime.now()}"
}
await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='POST',
json_data=address_data2
)
# GET erneut
all_addresses = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
hauptadressen = [a for a in all_addresses if a.get('standardAnschrift')]
print(f"\n{BOLD}Ergebnis nach 2. Adresse:{RESET}")
print(f" Anzahl Hauptadressen: {len(hauptadressen)}")
if len(hauptadressen) == 1:
print_success("\n✓ Es gibt nur EINE Hauptadresse!")
print_success("✓ Alte Hauptadresse wurde automatisch deaktiviert")
print(f" Aktuelle Hauptadresse: {hauptadressen[0].get('strasse')}")
elif len(hauptadressen) == 2:
print_error("\n✗ Es gibt ZWEI Hauptadressen!")
print_error("✗ Advoware erlaubt mehrere Hauptadressen")
for ha in hauptadressen:
print(f" - {ha.get('strasse')}")
elif len(hauptadressen) == 0:
print_error("\n✗ KEINE Hauptadresse!")
print_error("✗ standardAnschrift wird nicht gespeichert")
print(f"\n{BOLD}FAZIT:{RESET}")
if len(hauptadressen) == 1:
print_success("✓ Advoware verwaltet automatisch EINE Hauptadresse")
print_success("✓ Neue Hauptadresse deaktiviert alte automatisch")
elif len(hauptadressen) > 1:
print_error("✗ Mehrere Hauptadressen möglich")
else:
print_error("✗ standardAnschrift ist möglicherweise READ-ONLY")
print(f"\n{YELLOW}⚠️ Test-Adressen mit 'TEST-HAUPT' bereinigen{RESET}\n")
if __name__ == "__main__":
asyncio.run(main())

304
bitbylaw/scripts/test_hauptadresse_logic.py Normal file
View File

@@ -0,0 +1,304 @@
#!/usr/bin/env python3
"""
Test: Hauptadresse-Logik in Advoware
=====================================
Hypothese: Die neueste Adresse wird automatisch zur Hauptadresse (standardAnschrift = true)
Test:
1. Hole aktuelle Adressen und identifiziere Hauptadresse
2. Erstelle neue Adresse
3. Prüfe ob neue Adresse zur Hauptadresse wird
4. Prüfe ob alte Hauptadresse deaktiviert wird
"""
import asyncio
import sys
import os
from datetime import datetime
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from services.advoware import AdvowareAPI
TEST_BETNR = 104860
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:
def info(self, msg): pass
def error(self, msg): pass
def debug(self, msg): pass
def warning(self, msg): pass
class SimpleContext:
def __init__(self):
self.logger = SimpleLogger()
async def test_1_check_current_hauptadresse():
"""Test 1: Welche Adresse ist aktuell die Hauptadresse?"""
print_header("TEST 1: Aktuelle Hauptadresse identifizieren")
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)}")
# Finde Hauptadresse
hauptadresse = None
for addr in all_addresses:
if addr.get('standardAnschrift'):
hauptadresse = addr
break
if hauptadresse:
print_success(f"\n✓ Hauptadresse gefunden:")
print(f" Index: {hauptadresse.get('reihenfolgeIndex')}")
print(f" Straße: {hauptadresse.get('strasse')}")
print(f" Ort: {hauptadresse.get('ort')}")
print(f" standardAnschrift: {hauptadresse.get('standardAnschrift')}")
print(f" bemerkung: {hauptadresse.get('bemerkung', 'N/A')}")
# Prüfe ob es "Test 6667426" ist
bemerkung = hauptadresse.get('bemerkung', '')
if '6667426' in str(bemerkung) or '6667426' in str(hauptadresse.get('strasse', '')):
print_success("✓ Bestätigt: 'Test 6667426' ist Hauptadresse")
return hauptadresse
else:
print_warning("⚠ Keine Hauptadresse (standardAnschrift = true) gefunden!")
print_info("\nAlle Adressen:")
for i, addr in enumerate(all_addresses, 1):
print(f"\n Adresse {i}:")
print(f" Index: {addr.get('reihenfolgeIndex')}")
print(f" Straße: {addr.get('strasse')}")
print(f" standardAnschrift: {addr.get('standardAnschrift')}")
return None
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def test_2_create_new_address():
"""Test 2: Erstelle neue Adresse"""
print_header("TEST 2: Neue Adresse erstellen")
context = SimpleContext()
advo = AdvowareAPI(context=context)
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
new_address_data = {
"strasse": "Neue Hauptadresse Test 999",
"plz": "12345",
"ort": "Neustadt",
"land": "DE",
"anschrift": "Neue Hauptadresse Test 999\n12345 Neustadt\nDeutschland",
"bemerkung": f"TEST-HAUPTADRESSE: Erstellt {timestamp}",
"gueltigVon": "2026-02-08T00:00:00"
# KEIN standardAnschrift gesetzt → schauen was passiert
}
print_info("Erstelle neue Adresse OHNE standardAnschrift-Flag...")
print(f" Straße: {new_address_data['strasse']}")
print(f" Ort: {new_address_data['ort']}")
try:
result = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='POST',
json_data=new_address_data
)
if result and len(result) > 0:
created = result[0]
print_success("\n✓ Adresse erstellt!")
print(f" rowId: {created.get('rowId')}")
print(f" standardAnschrift: {created.get('standardAnschrift')}")
print(f" reihenfolgeIndex: {created.get('reihenfolgeIndex')}")
return created.get('rowId')
else:
print_error("POST fehlgeschlagen")
return None
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
return None
async def test_3_check_after_creation(old_hauptadresse, new_row_id):
"""Test 3: Prüfe Hauptadresse nach Erstellung"""
print_header("TEST 3: Hauptadresse nach Erstellung 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'
)
print_info(f"Gesamtanzahl Adressen: {len(all_addresses)}")
# Finde neue Adresse
new_addr = next((a for a in all_addresses if a.get('rowId') == new_row_id), None)
# Finde alte Hauptadresse
old_hauptadresse_now = None
if old_hauptadresse:
old_row_id = old_hauptadresse.get('rowId')
old_hauptadresse_now = next((a for a in all_addresses if a.get('rowId') == old_row_id), None)
# Finde aktuelle Hauptadresse(n)
hauptadressen = [a for a in all_addresses if a.get('standardAnschrift')]
print(f"\n{BOLD}Ergebnis:{RESET}")
print(f" Anzahl Adressen mit standardAnschrift = true: {len(hauptadressen)}")
if new_addr:
print(f"\n{BOLD}Neue Adresse:{RESET}")
print(f" Index: {new_addr.get('reihenfolgeIndex')}")
print(f" Straße: {new_addr.get('strasse')}")
print(f" standardAnschrift: {new_addr.get('standardAnschrift')}")
print(f" rowId: {new_addr.get('rowId')}")
if old_hauptadresse_now:
print(f"\n{BOLD}Alte Hauptadresse (vorher):{RESET}")
print(f" Index: {old_hauptadresse_now.get('reihenfolgeIndex')}")
print(f" Straße: {old_hauptadresse_now.get('strasse')}")
print(f" standardAnschrift: {old_hauptadresse_now.get('standardAnschrift')}")
# Analyse
print(f"\n{BOLD}{'='*80}{RESET}")
print(f"{BOLD}ANALYSE:{RESET}\n")
if new_addr and new_addr.get('standardAnschrift'):
print_success("✓✓✓ NEUE Adresse IST jetzt Hauptadresse!")
if old_hauptadresse_now and not old_hauptadresse_now.get('standardAnschrift'):
print_success("✓ Alte Hauptadresse wurde DEAKTIVIERT (standardAnschrift = false)")
print_info("\n💡 ERKENNTNIS: Es gibt immer nur EINE Hauptadresse")
print_info("💡 Neue Adresse wird AUTOMATISCH zur Hauptadresse")
print_info("💡 Alte Hauptadresse wird automatisch deaktiviert")
elif old_hauptadresse_now and old_hauptadresse_now.get('standardAnschrift'):
print_warning("⚠ Alte Hauptadresse ist NOCH aktiv!")
print_warning("⚠ Es gibt jetzt ZWEI Hauptadressen!")
elif new_addr and not new_addr.get('standardAnschrift'):
print_warning("⚠ Neue Adresse ist NICHT Hauptadresse")
if old_hauptadresse_now and old_hauptadresse_now.get('standardAnschrift'):
print_success("✓ Alte Hauptadresse ist NOCH aktiv")
print_info("\n💡 ERKENNTNIS: Neue Adresse wird NICHT automatisch zur Hauptadresse")
print_info("💡 Hauptadresse muss explizit gesetzt werden")
# Zeige alle Hauptadressen
if len(hauptadressen) > 0:
print(f"\n{BOLD}Alle Adressen mit standardAnschrift = true:{RESET}")
for ha in hauptadressen:
print(f"\n Index {ha.get('reihenfolgeIndex')}:")
print(f" Straße: {ha.get('strasse')}")
print(f" Ort: {ha.get('ort')}")
print(f" bemerkung: {ha.get('bemerkung', 'N/A')[:50]}...")
# Sortier-Analyse
print(f"\n{BOLD}Reihenfolge-Analyse:{RESET}")
sorted_addresses = sorted(all_addresses, key=lambda a: a.get('reihenfolgeIndex', 0))
print(f" Erste Adresse (Index {sorted_addresses[0].get('reihenfolgeIndex')}):")
print(f" standardAnschrift: {sorted_addresses[0].get('standardAnschrift')}")
print(f" Straße: {sorted_addresses[0].get('strasse')}")
print(f" Letzte Adresse (Index {sorted_addresses[-1].get('reihenfolgeIndex')}):")
print(f" standardAnschrift: {sorted_addresses[-1].get('standardAnschrift')}")
print(f" Straße: {sorted_addresses[-1].get('strasse')}")
if sorted_addresses[-1].get('standardAnschrift'):
print_success("\n✓✓✓ BESTÄTIGT: Letzte (neueste) Adresse ist Hauptadresse!")
except Exception as e:
print_error(f"Fehler: {e}")
import traceback
traceback.print_exc()
async def main():
print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}")
print(f"{BOLD}║ Hauptadresse-Logik Test (Advoware) ║{RESET}")
print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n")
print(f"Test-Konfiguration:")
print(f" BetNr: {TEST_BETNR}")
print(f" Datum: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f" Hypothese: Neueste Adresse wird automatisch zur Hauptadresse")
# Test 1: Aktuelle Hauptadresse
old_hauptadresse = await test_1_check_current_hauptadresse()
# Test 2: Neue Adresse erstellen
new_row_id = await test_2_create_new_address()
if not new_row_id:
print_error("\nTest abgebrochen: Konnte keine neue Adresse erstellen")
return
# Kurze Pause (falls Advoware Zeit braucht)
await asyncio.sleep(1)
# Test 3: Prüfe nach Erstellung
await test_3_check_after_creation(old_hauptadresse, new_row_id)
print(f"\n{BOLD}╔══════════════════════════════════════════════════════════════╗{RESET}")
print(f"{BOLD}║ FAZIT ║{RESET}")
print(f"{BOLD}╚══════════════════════════════════════════════════════════════╝{RESET}\n")
print_info("Basierend auf diesem Test können wir die Hauptadresse-Logik verstehen:")
print_info("1. Gibt es immer nur EINE Hauptadresse?")
print_info("2. Wird neue Adresse AUTOMATISCH zur Hauptadresse?")
print_info("3. Wird alte Hauptadresse deaktiviert?")
print_info("4. Ist die LETZTE Adresse immer die Hauptadresse?")
print()
print_info("→ Diese Erkenntnisse sind wichtig für Sync-Strategie!")
print(f"\n{YELLOW}⚠️ Test-Adresse 'TEST-HAUPTADRESSE' sollte bereinigt werden.{RESET}\n")
if __name__ == "__main__":
asyncio.run(main())

127
bitbylaw/scripts/test_put_response_detail.py Normal file
View File

@@ -0,0 +1,127 @@
#!/usr/bin/env python3
"""
Test: Welche Felder sind bei PUT wirklich änderbar?
====================================================
"""
import asyncio
import sys
import os
import json
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from services.advoware import AdvowareAPI
TEST_BETNR = 104860
BOLD = '\033[1m'
RED = '\033[91m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
RESET = '\033[0m'
def print_success(text):
print(f"{GREEN}✓ {text}{RESET}")
def print_error(text):
print(f"{RED}✗ {text}{RESET}")
def print_info(text):
print(f"{BLUE}ℹ {text}{RESET}")
class SimpleLogger:
def info(self, msg): pass
def error(self, msg): pass
def debug(self, msg): pass
def warning(self, msg): pass
class SimpleContext:
def __init__(self):
self.logger = SimpleLogger()
async def main():
print(f"\n{BOLD}=== PUT Response Analyse ==={RESET}\n")
context = SimpleContext()
advo = AdvowareAPI(context=context)
# Finde Test-Adresse
all_addresses = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen',
method='GET'
)
test_addr = None
for addr in all_addresses:
bemerkung = addr.get('bemerkung') or ''
if 'TEST-SOFTDELETE' in bemerkung:
test_addr = addr
break
if not test_addr:
print_error("Test-Adresse nicht gefunden")
return
index = test_addr.get('reihenfolgeIndex')
print_info(f"Test-Adresse Index: {index}")
print_info("\nVORHER:")
print(json.dumps(test_addr, indent=2, ensure_ascii=False))
# PUT mit ALLEN Feldern inklusive gueltigBis
print_info("\n=== Sende PUT mit ALLEN Feldern ===")
update_data = {
"strasse": "GEÄNDERT Straße",
"plz": "11111",
"ort": "GEÄNDERT Ort",
"land": "AT",
"postfach": "PF 123",
"postfachPLZ": "11112",
"anschrift": "GEÄNDERT Anschrift",
"standardAnschrift": True,
"bemerkung": "VERSUCH: bemerkung ändern",
"gueltigVon": "2025-01-01T00:00:00", # ← GEÄNDERT
"gueltigBis": "2027-12-31T23:59:59" # ← NEU GESETZT
}
print(json.dumps(update_data, indent=2, ensure_ascii=False))
result = await advo.api_call(
f'/api/v1/advonet/Beteiligte/{TEST_BETNR}/Adressen/{index}',
method='PUT',
json_data=update_data
)
print_info("\n=== PUT Response: ===")
print(json.dumps(result, indent=2, ensure_ascii=False))
# GET und vergleichen
print_info("\n=== GET nach PUT: ===")
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') == index), None)
if updated_addr:
print(json.dumps(updated_addr, indent=2, ensure_ascii=False))
print(f"\n{BOLD}=== VERGLEICH: Was wurde wirklich geändert? ==={RESET}\n")
fields = ['strasse', 'plz', 'ort', 'land', 'postfach', 'postfachPLZ',
'anschrift', 'standardAnschrift', 'bemerkung', 'gueltigVon', 'gueltigBis']
for field in fields:
sent = update_data.get(field)
received = updated_addr.get(field)
if sent == received:
print_success(f"{field:20s}: ✓ GEÄNDERT → {received}")
else:
print_error(f"{field:20s}: ✗ NICHT geändert (sent: {sent}, got: {received})")
if __name__ == "__main__":
asyncio.run(main())