diff --git a/bitbylaw/scripts/test_notification.py b/bitbylaw/scripts/test_notification.py new file mode 100644 index 00000000..b9072d45 --- /dev/null +++ b/bitbylaw/scripts/test_notification.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +""" +Test: Notification System +========================== + +Sendet testweise Notifications an EspoCRM: +1. Task-Erstellung +2. In-App Notification +3. READ-ONLY Field Conflict +""" + +import asyncio +import sys +import os +from datetime import datetime + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from services.notification_utils import NotificationManager +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): print(f"[INFO] {msg}") + def warning(self, msg): print(f"{YELLOW}[WARN] {msg}{RESET}") + def error(self, msg): print(f"{RED}[ERROR] {msg}{RESET}") + +class SimpleContext: + def __init__(self): + self.logger = SimpleLogger() + + +async def main(): + print_section("TEST: Notification System") + + context = SimpleContext() + espo = EspoCRMAPI(context=context) + notification_mgr = NotificationManager(espocrm_api=espo, context=context) + + # Finde echte Test-Adresse + print_info("Suche Test-Adresse in EspoCRM...") + + import json + addresses = await espo.list_entities( + 'CAdressen', + where=json.dumps([{ + 'type': 'contains', + 'attribute': 'name', + 'value': 'SYNC-TEST' + }]), + max_size=1 + ) + + if not addresses.get('list'): + print_error("Keine SYNC-TEST Adresse gefunden - erstelle eine...") + + # Hole Beteiligten + beteiligte = await espo.list_entities( + 'CBeteiligte', + where=json.dumps([{ + 'type': 'equals', + 'attribute': 'betNr', + 'value': '104860' + }]), + max_size=1 + ) + + if not beteiligte.get('list'): + print_error("Beteiligter nicht gefunden!") + return + + # Erstelle Test-Adresse + import datetime as dt + test_addr = await espo.create_entity('CAdressen', { + 'name': f'NOTIFICATION-TEST {dt.datetime.now().strftime("%H:%M:%S")}', + 'adresseStreet': 'Notification Test Str. 999', + 'adresseCity': 'Teststadt', + 'adressePostalCode': '12345', + 'beteiligteId': beteiligte['list'][0]['id'] + }) + + TEST_ENTITY_ID = test_addr['id'] + print_success(f"Test-Adresse erstellt: {TEST_ENTITY_ID}") + else: + TEST_ENTITY_ID = addresses['list'][0]['id'] + print_success(f"Test-Adresse gefunden: {TEST_ENTITY_ID}") + + # 1. Test: Address Delete Required + print_section("1. Test: Address Delete Notification") + + print_info("Sende DELETE-Notification...") + + result = await notification_mgr.notify_manual_action_required( + entity_type='CAdressen', + entity_id=TEST_ENTITY_ID, + action_type='address_delete_required', + details={ + 'message': 'TEST: Adresse in Advoware löschen', + 'description': ( + 'TEST-Notification:\n' + 'Diese Adresse wurde in EspoCRM gelöscht:\n' + 'Teststraße 123\n' + '10115 Berlin\n\n' + 'Bitte manuell in Advoware löschen:\n' + '1. Öffne Beteiligten 104860 in Advoware\n' + '2. Gehe zu Adressen-Tab\n' + '3. Lösche Adresse (Index 1)\n' + '4. Speichern' + ), + 'advowareIndex': 1, + 'betnr': 104860, + 'address': 'Teststraße 123, Berlin', + 'priority': 'Medium' + } + ) + + if result: + print_success("✓ DELETE-Notification gesendet!") + if result.get('task_id'): + print(f" Task ID: {result['task_id']}") + if result.get('notification_id'): + print(f" Notification ID: {result['notification_id']}") + else: + print_error("✗ DELETE-Notification fehlgeschlagen!") + + # 2. Test: READ-ONLY Field Conflict + print_section("2. Test: READ-ONLY Field Conflict Notification") + + print_info("Sende READ-ONLY Conflict Notification...") + + changes = [ + { + 'field': 'Hauptadresse', + 'espoField': 'isPrimary', + 'advoField': 'standardAnschrift', + 'espoCRM_value': True, + 'advoware_value': False + }, + { + 'field': 'Land', + 'espoField': 'adresseCountry', + 'advoField': 'land', + 'espoCRM_value': 'AT', + 'advoware_value': 'DE' + } + ] + + change_details = '\n'.join([ + f"- {c['field']}: EspoCRM='{c['espoCRM_value']}' → " + f"Advoware='{c['advoware_value']}'" + for c in changes + ]) + + result2 = await notification_mgr.notify_manual_action_required( + entity_type='CAdressen', + entity_id=TEST_ENTITY_ID, + action_type='readonly_field_conflict', + details={ + 'message': f'TEST: {len(changes)} READ-ONLY Feld(er) geändert', + 'description': ( + f'TEST-Notification:\n' + f'Folgende Felder wurden in EspoCRM geändert, sind aber ' + f'READ-ONLY in Advoware und können nicht automatisch ' + f'synchronisiert werden:\n\n{change_details}\n\n' + f'Bitte manuell in Advoware anpassen:\n' + f'1. Öffne Beteiligten 104860 in Advoware\n' + f'2. Gehe zu Adressen-Tab\n' + f'3. Passe die Felder manuell an\n' + f'4. Speichern' + ), + 'changes': changes, + 'address': 'Teststraße 123, Berlin', + 'betnr': 104860, + 'priority': 'High' + } + ) + + if result2: + print_success("✓ READ-ONLY Conflict Notification gesendet!") + if result2.get('task_id'): + print(f" Task ID: {result2['task_id']}") + if result2.get('notification_id'): + print(f" Notification ID: {result2['notification_id']}") + else: + print_error("✗ READ-ONLY Conflict Notification fehlgeschlagen!") + + # 3. Test: General Manual Action + print_section("3. Test: General Manual Action Notification") + + print_info("Sende allgemeine Notification...") + + result3 = await notification_mgr.notify_manual_action_required( + entity_type='CBeteiligte', + entity_id='6987b30a9bbbfefd0', + action_type='general_manual_action', + details={ + 'message': 'TEST: Allgemeine manuelle Aktion erforderlich', + 'description': ( + 'TEST-Notification:\n\n' + 'Dies ist eine Test-Notification für das Notification-System.\n' + 'Sie dient nur zu Testzwecken und kann ignoriert werden.\n\n' + f'Erstellt am: {datetime.now().strftime("%d.%m.%Y %H:%M:%S")}' + ), + 'priority': 'Low' + }, + create_task=False # Kein Task für diesen Test + ) + + if result3: + print_success("✓ General Notification gesendet!") + if result3.get('task_id'): + print(f" Task ID: {result3['task_id']}") + if result3.get('notification_id'): + print(f" Notification ID: {result3['notification_id']}") + else: + print_error("✗ General Notification fehlgeschlagen!") + + print_section("ZUSAMMENFASSUNG") + + print_info("Prüfe EspoCRM:") + print(" 1. Öffne Tasks-Modul") + print(" 2. Suche nach 'TEST:'") + print(" 3. Prüfe Notifications (Glocken-Icon)") + print() + print_success("✓ 3 Test-Notifications versendet!") + print_info("⚠ Bitte manuell in EspoCRM löschen nach dem Test") + + +if __name__ == '__main__': + asyncio.run(main())