[2ff88f42] Finalize SuperOffice Connector: Centralized Config, Added Position/Role Mapping Logic, and Discovery Tools

connector-superoffice/test_e2e_local.py

@@ -0,0 +1,80 @@
+import time
+import json
+import logging
+from queue_manager import JobQueue
+from worker import process_job
+from superoffice_client import SuperOfficeClient
+from config import settings
+from unittest.mock import MagicMock
+
+# Setup Logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("e2e-test")
+
+def test_e2e():
+    print("🚀 Starting End-to-End Simulation...")
+
+    # 1. Mock the SuperOffice Client
+    # We don't want to hit the real API in this test script unless we are sure.
+    # But wait, the user asked for "Finaler End-to-End Systemtest".
+    # Usually this implies hitting the real systems. 
+    # Let's try to use the REAL client if credentials are present, otherwise Mock.
+    
+    real_client = False
+    if settings.SO_CLIENT_ID and settings.SO_REFRESH_TOKEN:
+        print("✅ Real Credentials found. Attempting real connection...")
+        try:
+            so_client = SuperOfficeClient()
+            if so_client.access_token:
+                real_client = True
+        except:
+            print("⚠️ Real connection failed. Falling back to Mock.")
+    
+    if not real_client:
+        print("⚠️ Using MOCKED SuperOffice Client.")
+        so_client = MagicMock()
+        so_client.get_contact.return_value = {"ContactId": 123, "Name": "Test Company", "UserDefinedFields": {}}
+        so_client.get_person.return_value = {"PersonId": 456, "Contact": {"ContactId": 123}, "UserDefinedFields": {}}
+        so_client.update_entity_udfs.return_value = True
+    else:
+        # Use a SAFE contact ID for testing if possible. 
+        # CAUTION: This writes to the real system.
+        # Verify with user? Use a known "Gemini Test" contact?
+        # Let's use the ID 2 (which was mentioned in status updates as test data)
+        TEST_CONTACT_ID = 2
+        # Verify it exists
+        c = so_client.get_contact(TEST_CONTACT_ID)
+        if not c:
+            print(f"❌ Test Contact {TEST_CONTACT_ID} not found. Aborting real write-test.")
+            return
+
+        print(f"ℹ️ Using Real Contact: {c.get('Name')} (ID: {TEST_CONTACT_ID})")
+    
+    # 2. Create a Fake Job
+    fake_job = {
+        "id": "test-job-001",
+        "event_type": "contact.changed",
+        "payload": {
+            "PrimaryKey": 2, # Use the real ID
+            "ContactId": 2,
+            "JobTitle": "Geschäftsführer" # Trigger mapping
+        },
+        "created_at": time.time()
+    }
+
+    # 3. Process the Job (using worker logic)
+    # NOTE: This assumes COMPANY_EXPLORER_URL is reachable. 
+    # If running in CLI container, it might need to be 'localhost' or the docker DNS name.
+    # Let's override config for this test run.
+    # settings.COMPANY_EXPLORER_URL = "http://localhost:8000" # Try localhost first if running on host/mapped
+    
+    print(f"\n⚙️ Processing Job with CE URL: {settings.COMPANY_EXPLORER_URL}...")
+    
+    try:
+        result = process_job(fake_job, so_client)
+        print(f"\n✅ Job Result: {result}")
+    except Exception as e:
+        print(f"\n❌ Job Failed: {e}")
+
+if __name__ == "__main__":
+    test_e2e()