bitbylaw
c770f2c8ee
- 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.
467 lines
17 KiB
Python
467 lines
17 KiB
Python
#!/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())
|