128 lines
5.1 KiB
Python
128 lines
5.1 KiB
Python
import os
|
|
import requests
|
|
import json
|
|
import logging
|
|
import sys
|
|
import time
|
|
|
|
# Configure path to import modules from parent directory
|
|
# This makes the script runnable from the project root
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
parent_dir = os.path.join(script_dir, '..')
|
|
sys.path.append(parent_dir)
|
|
|
|
from dotenv import load_dotenv
|
|
# Load .env from project root
|
|
dotenv_path = os.path.join(parent_dir, '..', '.env')
|
|
load_dotenv(dotenv_path=dotenv_path)
|
|
|
|
from config import settings
|
|
from superoffice_client import SuperOfficeClient
|
|
|
|
|
|
# Logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger("e2e-roundtrip")
|
|
|
|
# Config - Use a real, enriched company for this test
|
|
API_USER = os.getenv("API_USER", "admin")
|
|
API_PASS = os.getenv("API_PASSWORD", "gemini")
|
|
TEST_PERSON_ID = 2 # This is a placeholder, a real one would be used in a live env
|
|
TEST_CONTACT_ID = 1 # Company ID for "THERME ERDING" in the CE database
|
|
|
|
def run_roundtrip():
|
|
print("🚀 STARTING E2E TEXT GENERATION TEST (CE -> SuperOffice)\n")
|
|
|
|
so_client = SuperOfficeClient()
|
|
if not so_client.access_token:
|
|
print("❌ SuperOffice Auth failed. Check .env")
|
|
return
|
|
|
|
scenarios = [
|
|
{
|
|
"name": "Scenario A: Infrastructure Role (Facility Manager)",
|
|
"job_title": "Leiter Facility Management",
|
|
"expected_opener_field": "opener",
|
|
"expected_keyword": "Sicherheit" # Keyword for Primary opener (Hygiene/Safety)
|
|
},
|
|
{
|
|
"name": "Scenario B: Operational Role (Leiter Badbetrieb)",
|
|
"job_title": "Leiter Badebetrieb",
|
|
"expected_opener_field": "opener_secondary",
|
|
"expected_keyword": "Gäste" # Keyword for Secondary opener (Guest experience/Service)
|
|
}
|
|
]
|
|
|
|
for s in scenarios:
|
|
print(f"--- Running {s['name']}: {s['job_title']} ---")
|
|
|
|
# 1. Provisioning from Company Explorer
|
|
print(f"1. 🧠 Asking Company Explorer for texts...")
|
|
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": "THERME ERDING Service GmbH", # Real data
|
|
"crm_website": "https://www.therme-erding.de/",
|
|
"job_title": s['job_title']
|
|
}
|
|
|
|
try:
|
|
resp = requests.post(ce_url, json=payload, auth=(API_USER, API_PASS))
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
# --- ASSERTIONS ---
|
|
print("2. 🧐 Verifying API Response...")
|
|
|
|
# Check if opener fields exist
|
|
assert "opener" in data, "❌ FAILED: 'opener' field is missing in response!"
|
|
assert "opener_secondary" in data, "❌ FAILED: 'opener_secondary' field is missing in response!"
|
|
print(" ✅ 'opener' and 'opener_secondary' fields are present.")
|
|
|
|
# Check if the specific opener for the role is not empty
|
|
opener_text = data.get(s['expected_opener_field'])
|
|
assert opener_text, f"❌ FAILED: Expected opener '{s['expected_opener_field']}' is empty!"
|
|
print(f" ✅ Expected opener '{s['expected_opener_field']}' is not empty.")
|
|
print(f" -> Content: '{opener_text}'")
|
|
|
|
# Check for keyword
|
|
assert s['expected_keyword'].lower() in opener_text.lower(), f"❌ FAILED: Keyword '{s['expected_keyword']}' not in opener text!"
|
|
print(f" ✅ Keyword '{s['expected_keyword']}' found in opener.")
|
|
|
|
# --- Write to SuperOffice ---
|
|
print(f"3. ✍️ Writing verified texts to SuperOffice UDFs...")
|
|
texts = data.get("texts", {})
|
|
udf_payload = {
|
|
settings.UDF_SUBJECT: texts.get("subject", ""),
|
|
settings.UDF_INTRO: texts.get("intro", ""),
|
|
settings.UDF_SOCIAL_PROOF: texts.get("social_proof", ""),
|
|
"x_opener_primary": data.get("opener", ""), # Assuming UDF names
|
|
"x_opener_secondary": data.get("opener_secondary", "") # Assuming UDF names
|
|
}
|
|
|
|
# This part is a simulation of the write; in a real test we'd need the real ProgIDs
|
|
# For now, we confirm the logic works up to this point.
|
|
if so_client.update_entity_udfs(TEST_PERSON_ID, "Person", {"String10": "E2E Test OK"}):
|
|
print(" -> ✅ Successfully wrote test confirmation to SuperOffice.")
|
|
else:
|
|
print(" -> ❌ Failed to write to SuperOffice.")
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f" ❌ CE API HTTP Error: {e.response.status_code} - {e.response.text}")
|
|
continue
|
|
except AssertionError as e:
|
|
print(f" {e}")
|
|
continue
|
|
except Exception as e:
|
|
print(f" ❌ An unexpected error occurred: {e}")
|
|
continue
|
|
|
|
print(f"--- PASSED: {s['name']} ---\n")
|
|
time.sleep(1)
|
|
|
|
print("🏁 Test Run Complete.")
|
|
|
|
if __name__ == "__main__":
|
|
run_roundtrip()
|