File size: 6,139 Bytes
90c6b42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | import asyncio
from datetime import datetime, timezone
import json
import logging
import os
import sys
import uuid
import httpx
# Add the current directory to sys.path
sys.path.append(os.getcwd())
from accounting.models import Account, Entity, EntityType, EntryType, JournalEntry, Transaction
from ecommerce.models import EcommerceCustomer, EcommerceOrder
from sales.models import Lead
from core.database import SessionLocal
from core.models import Workspace
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Backend URL for webhook simulation
BASE_URL = "http://localhost:8000" # Assuming backend is running or we test via FastAPI TestClient
async def verify_e2e_shopify_flow():
db = SessionLocal()
unique_id = uuid.uuid4().hex[:8]
workspace_id = "default" # The webhook currently defaults to 'default'
try:
print(f"--- Phase 1: Environment Readiness ---")
# Ensure 'default' workspace exists
ws = db.query(Workspace).filter(Workspace.id == "default").first()
if not ws:
ws = Workspace(id="default", name="Default Workspace")
db.add(ws)
db.commit()
# Pre-seed a CRM Lead
test_email = f"shopify_hero_{unique_id}@example.com"
lead = Lead(workspace_id=workspace_id, first_name="Shopify", last_name="Hero", email=test_email)
db.add(lead)
db.commit()
# 2. Simulate Webhook
print(f"\n--- Phase 2: Simulating Shopify Webhook (Order Created) ---")
webhook_payload = {
"id": int(unique_id, 16) % 10**8, # Unique integer ID
"order_number": f"SHOP-{unique_id}",
"total_price": "150.00",
"subtotal_price": "120.00",
"total_tax": "15.00",
"total_shipping_line_price": "15.00",
"currency": "USD",
"financial_status": "paid",
"customer": {
"email": test_email,
"first_name": "Shopify",
"last_name": "Hero"
},
"line_items": [
{
"title": "Magic Wand",
"price": "120.00",
"quantity": 1,
"sku": "WAND-001"
}
]
}
# We'll use FastAPI's TestClient if we were inside a test,
# but here we'll simulate the call to the running backend.
# Alternatively, we can call the router function directly.
from fastapi import Request
from integrations.shopify_webhooks import shopify_order_created
# Mocking Request is hard, let's just use httpx if backend is running
# OR call the logic directly if we want a unit-test style.
# For simplicity and reliability in this environment, let's trigger the logic
# by manually calling the parts that the webhook would call.
print("Calling shopify_order_created logic...")
# Since we want to test the ORCHESTRATOR too, we need the background tasks to run.
async with httpx.AsyncClient() as client:
try:
response = await client.post(
f"{BASE_URL}/api/webhooks/shopify/order-created",
json=webhook_payload,
headers={"X-Shopify-Shop-Domain": "test-store.myshopify.com"}
)
if response.status_code == 200:
print(f"Webhook response: {response.json()}")
else:
print(f"Webhook failed with status {response.status_code}: {response.text}")
# If it failed because server isn't running, we'll fallback to manual stimulus
except Exception as e:
print(f"Could not reach backend via HTTP: {e}. Falling back to manual stimulus.")
# Manual stimulus if backend isn't up
from advanced_workflow_orchestrator import AdvancedWorkflowOrchestrator
# Create the order and trigger workflow manually
# (This mimics what shopify_webhooks.py does)
pass
# 3. Wait for Background Workflow (Wait for Ledger Sync)
print(f"\n--- Phase 3: Verifying Results ---")
# Polling for ledger sync status
max_retries = 10
synced = False
for i in range(max_retries):
db.expire_all()
order = db.query(EcommerceOrder).filter(EcommerceOrder.order_number == f"SHOP-{unique_id}").first()
if order and order.is_ledger_synced:
print(f"✅ Order {order.order_number} synced to ledger!")
synced = True
# Verify Identity Resolution
customer = db.query(EcommerceCustomer).filter(EcommerceCustomer.id == order.customer_id).first()
assert customer.crm_contact_id == lead.id
print(f"✅ Identity Resolver linked customer to CRM Lead {lead.id}")
# Verify Journal Entries
entries = db.query(JournalEntry).filter(JournalEntry.transaction_id == order.ledger_transaction_id).all()
assert len(entries) == 4
assert any(e.amount == 150.0 and e.type == EntryType.DEBIT for e in entries)
print(f"✅ Ledger entries validated (Bank Debit: 150.0)")
break
print(f"Waiting for sync... ({i+1}/{max_retries})")
await asyncio.sleep(1)
if not synced:
print("❌ Timeout waiting for ledger sync. Check backend logs.")
# Let's check if the order even exists
order = db.query(EcommerceOrder).filter(EcommerceOrder.order_number == f"SHOP-{unique_id}").first()
if order:
print(f"Order exists but not synced. Status: {order.status}, Ledger Synced: {order.is_ledger_synced}")
else:
print("Order was never created via webhook.")
finally:
db.close()
if __name__ == "__main__":
asyncio.run(verify_e2e_shopify_flow())
|