Autoparts-DB/pos/tests/test_fase5.py

#!/usr/bin/env python3
"""FASE 5 — VALIDATION SUITE

Tests:
  1. CRM: activities, tags, loyalty, analytics
  2. Service Orders: create, status transitions, items, labor, kanban
  3. Images: upload info delete
"""

import os, sys, io, time
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from PIL import Image

from tenant_db import get_master_conn, get_tenant_conn_by_dbname
from services.crm_engine import (
    log_activity, get_activities, create_tag, list_tags,
    assign_tag, get_customer_tags, add_loyalty_points,
    redeem_points, get_loyalty_history, get_customer_analytics,
)
from services.service_order_engine import (
    create_service_order, get_service_order, list_service_orders,
    update_status, add_item, add_labor, get_kanban_summary,
)
from services.image_service import save_image, delete_image, get_image_info

MASTER_DB_URL = os.environ.get('MASTER_DB_URL', 'postgresql://postgres@/nexus_autoparts')
TENANT_TEMPLATE = os.environ.get('TENANT_DB_URL_TEMPLATE', 'postgresql://postgres@/{db_name}')

PASS = '\033[92mPASS\033[0m'
FAIL = '\033[91mFAIL\033[0m'
passed = 0
failed = 0


def ok(label, condition, detail=''):
    global passed, failed
    if condition:
        print(f"  [{PASS}] {label}")
        passed += 1
    else:
        print(f"  [{FAIL}] {label} {detail}")
        failed += 1


# Get a test tenant
master = get_master_conn()
cur = master.cursor()
cur.execute("SELECT db_name FROM tenants WHERE is_active = true ORDER BY id LIMIT 1")
row = cur.fetchone()
cur.close(); master.close()

if not row:
    print("No active tenant found!")
    sys.exit(1)

db_name = row[0]
conn = get_tenant_conn_by_dbname(db_name)

print("=" * 60)
print("FASE 5: CRM + Service Orders + Images — VALIDATION")
print("=" * 60)

# ─── Find test customer ─────────────────────────────
cur = conn.cursor()
cur.execute("SELECT id FROM customers WHERE is_active = true LIMIT 1")
cust_row = cur.fetchone()
customer_id = cust_row[0] if cust_row else None
cur.close()

if not customer_id:
    print("No customer found — creating one")
    cur = conn.cursor()
    cur.execute("INSERT INTO customers (name, phone, is_active) VALUES ('Test CRM', '5550000000', true) RETURNING id")
    customer_id = cur.fetchone()[0]
    conn.commit()
    cur.close()

print(f"\nTest customer_id: {customer_id}")

# ─── CRM: Activities ─────────────────────────────
print("\n[CRM ACTIVITIES]")
try:
    act_id = log_activity(conn, customer_id, 'note', title='Test note', description='Hello CRM')
    ok("Log activity", act_id is not None)

    acts = get_activities(conn, customer_id)
    ok("Get activities", len(acts) >= 1, f"count={len(acts)}")
except Exception as e:
    conn.rollback()
    ok("CRM activities", False, str(e))

# ─── CRM: Tags ─────────────────────────────
print("\n[CRM TAGS]")
try:
    tag_name = f"VIP-{int(time.time())}"
    tag_id = create_tag(conn, 1, tag_name, color='#FFD700', description='Very Important Customer')
    ok("Create tag", tag_id is not None)

    tags = list_tags(conn, 1)
    ok("List tags", any(t['name'] == tag_name for t in tags))

    assign_tag(conn, customer_id, tag_id)
    cust_tags = get_customer_tags(conn, customer_id)
    ok("Assign tag", any(t['name'] == tag_name for t in cust_tags))
except Exception as e:
    conn.rollback()
    ok("CRM tags", False, str(e))

# ─── CRM: Loyalty ─────────────────────────────
print("\n[CRM LOYALTY]")
try:
    # Add enough points for gold tier
    pid = add_loyalty_points(conn, customer_id, 2500, points_type='earned',
                             source_type='sale', description='Test points')
    ok("Add loyalty points", pid is not None)

    history = get_loyalty_history(conn, customer_id)
    ok("Loyalty history", len(history) >= 1)

    # Check balance updated (may have points from previous runs)
    cur = conn.cursor()
    cur.execute("SELECT loyalty_points_balance, loyalty_tier FROM customers WHERE id = %s", (customer_id,))
    bal, tier = cur.fetchone()
    cur.close()
    ok("Balance updated", bal >= 2500, f"balance={bal}")
    ok("Tier updated", tier in ('gold', 'platinum'), f"tier={tier}")  # 2500+ pts = gold or platinum
except Exception as e:
    conn.rollback()
    ok("CRM loyalty", False, str(e))

# ─── CRM: Analytics ─────────────────────────────
print("\n[CRM ANALYTICS]")
try:
    analytics = get_customer_analytics(conn, customer_id)
    ok("Analytics computed", isinstance(analytics, dict))
    ok("Analytics has LTV", 'ltv' in analytics)
    ok("Analytics has churn_risk", analytics.get('churn_risk') in ['low', 'medium', 'high'])
except Exception as e:
    conn.rollback()
    ok("CRM analytics", False, str(e))

# ─── Service Orders ─────────────────────────────
print("\n[SERVICE ORDERS]")
try:
    so = create_service_order(conn, {
        'tenant_id': 1, 'branch_id': None, 'customer_id': customer_id,
        'priority': 'high', 'reception_notes': 'Test order',
        'estimated_cost': 1500.00, 'mileage_in': 50000,
    })
    ok("Create SO", so.get('service_order_id') is not None, f"id={so.get('service_order_id')}")
    so_id = so['service_order_id']

    so_detail = get_service_order(conn, so_id)
    ok("Get SO detail", so_detail is not None and so_detail['status'] == 'received')
    ok("SO has order_number", so_detail.get('order_number', '').startswith('SO-'))

    # Status transition: received -> diagnosis
    result = update_status(conn, so_id, 'diagnosis', changed_by=1, notes='Starting diagnosis')
    ok("Status received->diagnosis", result['new_status'] == 'diagnosis')

    # Invalid transition
    try:
        update_status(conn, so_id, 'delivered')
        ok("Invalid transition blocked", False, "Should have raised ValueError")
    except ValueError:
        ok("Invalid transition blocked", True)

    # Add item
    item_id = add_item(conn, so_id, {
        'part_number': 'BP-123', 'name': 'Brake Pads', 'quantity': 2,
        'unit_price': 450.00, 'status': 'pending',
    })
    ok("Add item", item_id is not None)

    # Add labor
    labor_id = add_labor(conn, so_id, {
        'description': 'Replace brake pads', 'hours': 2.0, 'hourly_rate': 350.00,
    })
    ok("Add labor", labor_id is not None)

    # Refresh detail
    so_detail2 = get_service_order(conn, so_id)
    ok("SO has items", len(so_detail2.get('items', [])) >= 1)
    ok("SO has labor", len(so_detail2.get('labor', [])) >= 1)

    # List orders
    orders = list_service_orders(conn, status='diagnosis')
    ok("List orders", orders.get('data') is not None)

    # Kanban summary
    summary = get_kanban_summary(conn)
    ok("Kanban summary", 'received' in summary or 'diagnosis' in summary)

except Exception as e:
    conn.rollback()
    import traceback
    ok("Service Orders", False, f"{e}\n{traceback.format_exc()}")

# ─── Images ─────────────────────────────
print("\n[IMAGES]")
try:
    # Create a test image in memory
    img = Image.new('RGB', (800, 600), color='red')
    buf = io.BytesIO()
    img.save(buf, format='PNG')
    buf.seek(0)

    # Need a tenant_id and item_id
    tenant_id = 1
    cur = conn.cursor()
    cur.execute("SELECT id FROM inventory WHERE is_active = true LIMIT 1")
    inv_row = cur.fetchone()
    if inv_row:
        item_id = inv_row[0]
    else:
        cur.execute("INSERT INTO inventory (part_number, name, is_active) VALUES ('IMG-TEST', 'Image Test', true) RETURNING id")
        item_id = cur.fetchone()[0]
        conn.commit()
    cur.close()

    result = save_image(tenant_id, item_id, file_obj=buf, filename_hint='test.png')
    ok("Save image", result.get('image_url') is not None)

    info = get_image_info(tenant_id, item_id)
    ok("Get image info", info['has_image'] is True)

    delete_image(tenant_id, item_id)
    info2 = get_image_info(tenant_id, item_id)
    ok("Delete image", info2['has_image'] is False)
except Exception as e:
    conn.rollback()
    import traceback
    ok("Images", False, f"{e}\n{traceback.format_exc()}")

conn.close()

print("\n" + "=" * 60)
print(f"RESULTS: {PASS} {passed} passed, {FAIL} {failed} failed")
print("=" * 60)

sys.exit(0 if failed == 0 else 1)