81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
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()
|