bitbylaw
7856dd1d68
- Implemented comprehensive tests for the Kommunikation Sync functionality, covering base64 encoding, marker parsing, creation, type detection, and integration scenarios. - Added a verification script to check for unique IDs in Advoware communications, ensuring stability and integrity of the IDs. - Created utility scripts for code validation, notification testing, and PUT response detail analysis to enhance development and testing processes. - Updated README with details on new tools and their usage.
305 lines
12 KiB
Python
305 lines
12 KiB
Python
#!/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())
|