Spaces:
Sleeping
Sleeping
| 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()) | |