[2ff88f42] Finalize SuperOffice Integration: Enhanced Persona model with Influencer role, switched Matrix Generator to Gemini, implemented Noise Reduction for Webhooks, and added E2E test scenarios.
This commit is contained in:
100
connector-superoffice/tests/test_dynamic_change.py
Normal file
100
connector-superoffice/tests/test_dynamic_change.py
Normal file
@@ -0,0 +1,100 @@
|
||||
import os
|
||||
import requests
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
|
||||
# Configure to run from root context
|
||||
sys.path.append(os.path.join(os.getcwd(), "connector-superoffice"))
|
||||
|
||||
# Mock Config if needed, or use real one
|
||||
try:
|
||||
from config import settings
|
||||
except ImportError:
|
||||
print("Could not import settings. Ensure you are in project root.")
|
||||
sys.exit(1)
|
||||
|
||||
# FORCE CE URL for internal Docker comms if running inside container
|
||||
# If running outside, this might need localhost.
|
||||
# settings.COMPANY_EXPLORER_URL is used.
|
||||
|
||||
API_USER = os.getenv("API_USER", "admin")
|
||||
API_PASS = os.getenv("API_PASSWORD", "gemini")
|
||||
|
||||
def test_dynamic_role_change():
|
||||
print("🧪 STARTING TEST: Dynamic Role Change & Content Generation\n")
|
||||
|
||||
# Define Scenarios
|
||||
scenarios = [
|
||||
{
|
||||
"name": "Scenario A (CEO)",
|
||||
"job_title": "Geschäftsführer",
|
||||
"expect_keywords": ["Kostenreduktion", "Effizienz", "Amortisation"]
|
||||
},
|
||||
{
|
||||
"name": "Scenario B (Warehouse Mgr)",
|
||||
"job_title": "Lagerleiter",
|
||||
"expect_keywords": ["Stress", "Sauberkeit", "Entlastung"]
|
||||
}
|
||||
]
|
||||
|
||||
results = {}
|
||||
|
||||
for s in scenarios:
|
||||
print(f"--- Running {s['name']} ---")
|
||||
print(f"Role Trigger: '{s['job_title']}'")
|
||||
|
||||
payload = {
|
||||
"so_contact_id": 2, # RoboPlanet Test
|
||||
"so_person_id": 2,
|
||||
"crm_name": "RoboPlanet GmbH-SOD",
|
||||
"crm_website": "www.roboplanet.de", # Ensure we match the industry (Logistics)
|
||||
"job_title": s['job_title']
|
||||
}
|
||||
|
||||
try:
|
||||
url = f"{settings.COMPANY_EXPLORER_URL}/api/provision/superoffice-contact"
|
||||
print(f"POST {url}")
|
||||
resp = requests.post(url, json=payload, auth=(API_USER, API_PASS))
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
|
||||
# Validation
|
||||
texts = data.get("texts", {})
|
||||
subject = texts.get("subject", "")
|
||||
intro = texts.get("intro", "")
|
||||
|
||||
print(f"Received Role: {data.get('role_name')}")
|
||||
print(f"Received Subject: {subject}")
|
||||
|
||||
# Check Keywords
|
||||
full_text = (subject + " " + intro).lower()
|
||||
matches = [k for k in s['expect_keywords'] if k.lower() in full_text]
|
||||
|
||||
if len(matches) > 0:
|
||||
print(f"✅ Content Match! Found keywords: {matches}")
|
||||
results[s['name']] = "PASS"
|
||||
else:
|
||||
print(f"❌ Content Mismatch. Expected {s['expect_keywords']}, got text: {subject}...")
|
||||
results[s['name']] = "FAIL"
|
||||
|
||||
results[f"{s['name']}_Subject"] = subject # Store for comparison later
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ API Error: {e}")
|
||||
results[s['name']] = "ERROR"
|
||||
|
||||
print("")
|
||||
|
||||
# Final Comparison
|
||||
print("--- Final Result Analysis ---")
|
||||
if results["Scenario A (CEO)"] == "PASS" and results["Scenario B (Warehouse Mgr)"] == "PASS":
|
||||
if results["Scenario A (CEO)_Subject"] != results["Scenario B (Warehouse Mgr)_Subject"]:
|
||||
print("✅ SUCCESS: Different roles generated different, targeted content.")
|
||||
else:
|
||||
print("⚠️ WARNING: Content matched keywords but Subjects are identical! Check Matrix.")
|
||||
else:
|
||||
print("❌ TEST FAILED. See individual steps.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_dynamic_role_change()
|
||||
111
connector-superoffice/tests/test_full_roundtrip.py
Normal file
111
connector-superoffice/tests/test_full_roundtrip.py
Normal file
@@ -0,0 +1,111 @@
|
||||
import os
|
||||
import requests
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
|
||||
# Configure path to import modules from parent directory
|
||||
sys.path.append(os.path.join(os.getcwd(), "connector-superoffice"))
|
||||
|
||||
try:
|
||||
from config import settings
|
||||
from superoffice_client import SuperOfficeClient
|
||||
except ImportError:
|
||||
print("❌ Import Error. Ensure you are running from the project root.")
|
||||
sys.exit(1)
|
||||
|
||||
# Logging
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||||
logger = logging.getLogger("e2e-roundtrip")
|
||||
|
||||
# Config
|
||||
API_USER = os.getenv("API_USER", "admin")
|
||||
API_PASS = os.getenv("API_PASSWORD", "gemini")
|
||||
TEST_PERSON_ID = 2
|
||||
TEST_CONTACT_ID = 2
|
||||
|
||||
def run_roundtrip():
|
||||
print("🚀 STARTING FULL E2E ROUNDTRIP TEST (API -> SO Write)\n")
|
||||
|
||||
so_client = SuperOfficeClient()
|
||||
if not so_client.access_token:
|
||||
print("❌ SuperOffice Auth failed. Check .env")
|
||||
return
|
||||
|
||||
scenarios = [
|
||||
{
|
||||
"name": "Scenario A",
|
||||
"role_label": "Geschäftsführer",
|
||||
"expect_keyword": "Kosten"
|
||||
},
|
||||
{
|
||||
"name": "Scenario B",
|
||||
"role_label": "Lagerleiter",
|
||||
"expect_keyword": "Sauberkeit"
|
||||
}
|
||||
]
|
||||
|
||||
for s in scenarios:
|
||||
print(f"--- Running {s['name']}: {s['role_label']} ---")
|
||||
|
||||
# 1. Provisioning (Company Explorer)
|
||||
print(f"1. 🧠 Asking Company Explorer (Trigger: {s['role_label']})...")
|
||||
ce_url = f"{settings.COMPANY_EXPLORER_URL}/api/provision/superoffice-contact"
|
||||
payload = {
|
||||
"so_contact_id": TEST_CONTACT_ID,
|
||||
"so_person_id": TEST_PERSON_ID,
|
||||
"crm_name": "RoboPlanet GmbH-SOD",
|
||||
"crm_website": "www.roboplanet.de",
|
||||
"job_title": s['role_label'] # <-- THE TRIGGER
|
||||
}
|
||||
|
||||
try:
|
||||
resp = requests.post(ce_url, json=payload, auth=(API_USER, API_PASS))
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
|
||||
texts = data.get("texts", {})
|
||||
subject = texts.get("subject", "N/A")
|
||||
intro = texts.get("intro", "N/A")
|
||||
|
||||
print(f" -> Received Subject: '{subject}'")
|
||||
|
||||
if s['expect_keyword'].lower() not in (subject + intro).lower():
|
||||
print(f" ⚠️ WARNING: Expected keyword '{s['expect_keyword']}' not found!")
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ CE API Failed: {e}")
|
||||
continue
|
||||
|
||||
# 2. Write to SuperOffice (UDFs)
|
||||
print(f"2. ✍️ Writing Texts to SuperOffice UDFs...")
|
||||
udf_payload = {
|
||||
settings.UDF_SUBJECT: subject,
|
||||
settings.UDF_INTRO: intro,
|
||||
settings.UDF_SOCIAL_PROOF: texts.get("social_proof", "")
|
||||
}
|
||||
|
||||
if so_client.update_entity_udfs(TEST_PERSON_ID, "Person", udf_payload):
|
||||
print(" -> UDFs Updated.")
|
||||
else:
|
||||
print(" -> ❌ UDF Update Failed.")
|
||||
|
||||
# 3. Create Appointment (Proof)
|
||||
print(f"3. 📅 Creating Appointment in SuperOffice...")
|
||||
appt_subject = f"[E2E TEST] {s['role_label']}: {subject}"
|
||||
appt_desc = f"GENERATED CONTENT:\n\n{intro}\n\n{texts.get('social_proof')}"
|
||||
|
||||
appt = so_client.create_appointment(appt_subject, appt_desc, TEST_CONTACT_ID, TEST_PERSON_ID)
|
||||
if appt:
|
||||
print(f" -> ✅ Appointment Created (ID: {appt.get('AppointmentId')})")
|
||||
else:
|
||||
print(" -> ❌ Appointment Creation Failed.")
|
||||
|
||||
print("")
|
||||
time.sleep(1) # Brief pause
|
||||
|
||||
print("🏁 Test Run Complete.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_roundtrip()
|
||||
Reference in New Issue
Block a user