Spaces:
Sleeping
Sleeping
File size: 4,221 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 | import asyncio
from datetime import datetime, timezone
import logging
import os
import sys
import uuid
from sqlalchemy.orm import Session
# Add the current directory to sys.path
sys.path.append(os.getcwd())
from accounting.models import (
Account,
AccountType,
Entity,
EntityType,
EntryType,
JournalEntry,
Transaction,
)
from ecommerce.ledger_mapper import OrderToLedgerMapper
from ecommerce.models import EcommerceCustomer, EcommerceOrder, EcommerceOrderItem
from sales.models import Lead
from core.database import SessionLocal, engine
from core.identity_resolver import CustomerResolutionEngine
from core.models import Workspace
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def verify_ecommerce_flow():
db = SessionLocal()
unique_id = uuid.uuid4().hex[:8]
workspace_id = f"test-ws-ecommerce-{unique_id}"
try:
# 1. Setup Environment
print(f"--- Phase 1: Setting up Test Environment (WS: {workspace_id}) ---")
ws = Workspace(id=workspace_id, name=f"Ecommerce Test {unique_id}")
db.add(ws)
db.commit() # MUST commit workspace first for FKs to work
# Pre-seed a CRM Lead to test resolution
test_email = f"customer_{unique_id}@example.com"
lead = Lead(workspace_id=workspace_id, first_name="John", last_name="Doe", email=test_email)
db.add(lead)
# Pre-seed an Accounting Entity to test resolution
entity = Entity(workspace_id=workspace_id, name="John Doe", type=EntityType.CUSTOMER)
db.add(entity)
db.commit()
# 2. Simulate Order Arrival
print(f"\n--- Phase 2: Simulating Shopify Order Arrival ---")
customer = EcommerceCustomer(
workspace_id=workspace_id,
email=test_email,
first_name="John",
last_name="Doe",
external_id=f"sh_cust_{unique_id}"
)
db.add(customer)
db.flush()
order = EcommerceOrder(
workspace_id=workspace_id,
customer_id=customer.id,
external_id=f"sh_ord_{unique_id}",
order_number="1001",
total_price=120.0,
subtotal_price=100.0,
total_tax=10.0,
total_shipping=10.0,
currency="USD",
status="paid"
)
db.add(order)
db.commit()
db.refresh(order)
# 3. Resolve Identity
print(f"\n--- Phase 3: Resolving Identity ---")
resolver = CustomerResolutionEngine(db)
resolved_cust = resolver.resolve_customer(workspace_id, test_email, "John", "Doe")
assert resolved_cust.crm_contact_id == lead.id
assert resolved_cust.accounting_entity_id == entity.id
print("✅ Identity Resolution Verified!")
# 4. Map to Ledger
print(f"\n--- Phase 4: Mapping to Ledger ---")
mapper = OrderToLedgerMapper(db)
tx_id = mapper.process_order(order.id)
assert tx_id is not None
# Verify Journal Entries
tx = db.query(Transaction).filter(Transaction.id == tx_id).first()
entries = db.query(JournalEntry).filter(JournalEntry.transaction_id == tx_id).all()
print(f"Transaction: {tx.description}, Amount: {tx.amount}")
for je in entries:
acc = db.query(Account).filter(Account.id == je.account_id).first()
print(f" Entry: {je.type} {je.amount} -> {acc.name} ({acc.type})")
# Expectations:
# DEBIT Bank 120.0
# CREDIT Service Revenue 100.0
# CREDIT Sales Tax Payable 10.0
# CREDIT Shipping Income 10.0
assert len(entries) == 4
assert any(e.amount == 120.0 and e.type == EntryType.DEBIT for e in entries)
assert any(e.amount == 100.0 and e.type == EntryType.CREDIT for e in entries)
print("✅ Ledger Mapping Verified!")
except Exception as e:
logger.error(f"Verification failed: {e}")
import traceback
traceback.print_exc()
raise e
finally:
db.close()
if __name__ == "__main__":
asyncio.run(verify_ecommerce_flow())
|