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.
253 lines
7.9 KiB
Python
253 lines
7.9 KiB
Python
#!/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())
|