Autoparts-DB/pos/tests/test_fase3.py

#!/usr/bin/env python3
"""Test FASE 3 improvements:
- Multi-sucursal sync (#1)
- Reorder alerts (#7)
- Warranty / RMA (#10)
"""

import os
import sys

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

os.environ.setdefault('MASTER_DB_URL', 'postgresql://postgres@/nexus_autoparts')
os.environ.setdefault('TENANT_DB_URL_TEMPLATE', 'postgresql://postgres@/{db_name}')
os.environ.setdefault('POS_JWT_SECRET', 'test-secret-12345678901234567890123456789012')

from services.inventory_engine import get_stock, record_transfer
from services.reorder_engine import generate_alerts, list_alerts, suggest_po_from_alerts
from services.warranty_engine import (
    register_warranty, create_claim, resolve_claim, get_warranty, get_claim, list_claims
)
from tenant_db import get_tenant_conn_by_dbname

RED = '\033[91m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
RESET = '\033[0m'


def print_result(name, passed, detail=""):
    status = f"{GREEN}PASS{RESET}" if passed else f"{RED}FAIL{RESET}"
    print(f"  [{status}] {name}" + (f" — {detail}" if detail else ""))


def get_test_inventory(conn):
    cur = conn.cursor()
    cur.execute("""
        SELECT id, part_number, name, branch_id, min_stock, max_stock
        FROM inventory WHERE is_active = true LIMIT 1
    """)
    row = cur.fetchone()
    cur.close()
    return row


def get_branches(conn):
    cur = conn.cursor()
    cur.execute("SELECT id, name FROM branches WHERE is_active = true LIMIT 2")
    rows = cur.fetchall()
    cur.close()
    return rows


def main():
    print("=" * 60)
    print("FASE 3: Multi-sucursal + Alertas + Garantías — VALIDATION")
    print("=" * 60)

    passed = 0
    failed = 0
    conn = get_tenant_conn_by_dbname('tenant_acct_test')
    inv = get_test_inventory(conn)
    branches = get_branches(conn)

    if not inv:
        print(f"\n{RED}No inventory items — aborting{RESET}")
        return passed, failed
    if len(branches) < 2:
        print(f"\n{YELLOW}Only 1 branch — multi-branch tests limited{RESET}")

    inv_id, part_num, inv_name, branch_id, min_stock, max_stock = inv
    branch_a = branches[0][0] if branches else branch_id
    branch_b = branches[1][0] if len(branches) > 1 else branch_a

    # Get or create a valid customer_id
    cur = conn.cursor()
    cur.execute("SELECT id FROM customers LIMIT 1")
    cust_row = cur.fetchone()
    if not cust_row:
        cur.execute("""
            INSERT INTO customers (name, phone, branch_id, is_active)
            VALUES ('Cliente Test', '555-0000', %s, true)
            RETURNING id
        """, (branch_id,))
        customer_id = cur.fetchone()[0]
        conn.commit()
    else:
        customer_id = cust_row[0]
    cur.close()

    # ═══════════════════════════════════════════════════════════════════════
    # MULTI-SUCURSAL
    # ═══════════════════════════════════════════════════════════════════════
    print("\n[MULTI-SUCURSAL]")

    # Test 1: Stock by branch query
    try:
        cur = conn.cursor()
        cur.execute("""
            SELECT b.id, b.name, COALESCE(SUM(io.quantity), 0)
            FROM branches b
            LEFT JOIN inventory_operations io ON io.branch_id = b.id AND io.inventory_id = %s
            WHERE b.is_active = true GROUP BY b.id ORDER BY b.name
        """, (inv_id,))
        rows = cur.fetchall()
        cur.close()
        if rows:
            print_result("Stock by branch", True, f"{len(rows)} branches")
            passed += 1
        else:
            print_result("Stock by branch", False, "no data")
            failed += 1
    except Exception as e:
        print_result("Stock by branch", False, str(e))
        failed += 1

    # Test 2: Transfer between branches
    try:
        if branch_a != branch_b:
            stock_before_a = get_stock(conn, inv_id, branch_a)
            stock_before_b = get_stock(conn, inv_id, branch_b)
            record_transfer(conn, inv_id, branch_a, branch_b, 2, notes="Test transfer")
            conn.commit()
            stock_after_a = get_stock(conn, inv_id, branch_a)
            stock_after_b = get_stock(conn, inv_id, branch_b)
            if stock_after_a == stock_before_a - 2 and stock_after_b == stock_before_b + 2:
                print_result("Transfer", True, f"2 uds de {branch_a} a {branch_b}")
                passed += 1
            else:
                print_result("Transfer", False, f"stock mismatch A:{stock_before_a}->{stock_after_a} B:{stock_before_b}->{stock_after_b}")
                failed += 1
        else:
            print_result("Transfer", True, "SKIP (solo 1 sucursal)")
            passed += 1
    except Exception as e:
        conn.rollback()
        print_result("Transfer", False, str(e))
        failed += 1

    # Test 3: Price sync
    try:
        cur = conn.cursor()
        cur.execute("UPDATE inventory SET price_1 = 999.99 WHERE id = %s", (inv_id,))
        conn.commit()
        cur.execute("""
            UPDATE inventory SET price_1 = 999.99, price_2 = 888.88, price_3 = 777.77
            WHERE part_number = (SELECT part_number FROM inventory WHERE id = %s)
              AND id != %s
        """, (inv_id, inv_id))
        synced = cur.rowcount
        conn.commit()
        cur.close()
        print_result("Price sync", True, f"{synced} items actualizados")
        passed += 1
    except Exception as e:
        conn.rollback()
        print_result("Price sync", False, str(e))
        failed += 1

    # ═══════════════════════════════════════════════════════════════════════
    # ALERTAS DE REORDER
    # ═══════════════════════════════════════════════════════════════════════
    print("\n[ALERTAS DE REORDER]")

    # Test 4: Generate alerts
    try:
        # Ensure min_stock is set for our test item so it generates an alert
        cur = conn.cursor()
        cur.execute("UPDATE inventory SET min_stock = 1000, reorder_point = 500 WHERE id = %s", (inv_id,))
        conn.commit()
        cur.close()

        result = generate_alerts(conn, branch_id=branch_id, auto_notify=False)
        conn.commit()

        # Accept either new alerts created or valid empty result (deduplication)
        if isinstance(result, dict) and 'created' in result:
            print_result("Generate alerts", True, f"{result['created']} alertas ({result['by_type']})")
            passed += 1
        else:
            print_result("Generate alerts", False, "resultado inesperado")
            failed += 1
    except Exception as e:
        conn.rollback()
        print_result("Generate alerts", False, str(e))
        failed += 1

    # Test 5: List alerts
    try:
        alerts = list_alerts(conn, status='open', branch_id=branch_id, limit=10)
        if alerts:
            print_result("List alerts", True, f"{len(alerts)} alertas abiertas")
            passed += 1
        else:
            print_result("List alerts", False, "no hay alertas")
            failed += 1
    except Exception as e:
        print_result("List alerts", False, str(e))
        failed += 1

    # Test 6: Suggest PO
    try:
        suggestion = suggest_po_from_alerts(conn, branch_id=branch_id)
        if suggestion and suggestion.get('items'):
            print_result("Suggest PO", True, f"{len(suggestion['items'])} items sugeridos")
            passed += 1
        else:
            print_result("Suggest PO", False, "sin sugerencias")
            failed += 1
    except Exception as e:
        print_result("Suggest PO", False, str(e))
        failed += 1

    # ═══════════════════════════════════════════════════════════════════════
    # GARANTÍAS / RMA
    # ═══════════════════════════════════════════════════════════════════════
    print("\n[GARANTÍAS / RMA]")

    # Test 7: Register warranty
    try:
        if not customer_id:
            print_result("Register warranty", True, "SKIP (no customers)")
            passed += 1
            w_id = None
        else:
            w_id = register_warranty(
                conn, sale_id=1, sale_item_id=1, inventory_id=inv_id,
                customer_id=customer_id, warranty_months=12, part_number=part_num, name=inv_name
            )
        conn.commit()
        print_result("Register warranty", True, f"id={w_id}")
        passed += 1
    except Exception as e:
        conn.rollback()
        print_result("Register warranty", False, str(e))
        failed += 1
        w_id = None
        customer_id = None

    # Test 8: Get warranty
    try:
        if w_id:
            w = get_warranty(conn, w_id)
            if w and w['status'] == 'active':
                print_result("Get warranty", True, f"status={w['status']}, meses={w['warranty_months']}")
                passed += 1
            else:
                print_result("Get warranty", False, "datos incorrectos")
                failed += 1
        else:
            print_result("Get warranty", False, "no warranty id")
            failed += 1
    except Exception as e:
        print_result("Get warranty", False, str(e))
        failed += 1

    # Test 9: Create claim
    try:
        if w_id:
            claim_id = create_claim(conn, w_id, "El articulo presenta falla de fabrica", notes="Test claim")
            conn.commit()
            print_result("Create claim", True, f"id={claim_id}")
            passed += 1
        else:
            print_result("Create claim", False, "no warranty")
            failed += 1
    except Exception as e:
        conn.rollback()
        print_result("Create claim", False, str(e))
        failed += 1

    # Test 10: Resolve claim
    try:
        if w_id and claim_id:
            ok = resolve_claim(conn, claim_id, 'replaced', replacement_inventory_id=inv_id, notes="Reemplazado")
            conn.commit()
            if ok:
                c = get_claim(conn, claim_id)
                if c and c['resolution'] == 'replaced':
                    print_result("Resolve claim", True, f"resolution={c['resolution']}")
                    passed += 1
                else:
                    print_result("Resolve claim", False, "no se actualizo")
                    failed += 1
            else:
                print_result("Resolve claim", False, "resolve fallo")
                failed += 1
        else:
            print_result("Resolve claim", False, "no claim id")
            failed += 1
    except Exception as e:
        conn.rollback()
        print_result("Resolve claim", False, str(e))
        failed += 1

    # Test 11: List claims
    try:
        claims = list_claims(conn, status='resolved', limit=10)
        print_result("List claims", True, f"{len(claims)} claims")
        passed += 1
    except Exception as e:
        print_result("List claims", False, str(e))
        failed += 1

    conn.close()

    # ── Summary ─────────────────────────────────────────────────
    print("\n" + "=" * 60)
    print(f"RESULTS: {GREEN}{passed} passed{RESET}, {RED}{failed} failed{RESET}")
    print("=" * 60)
    return passed, failed


if __name__ == '__main__':
    passed, failed = main()
    sys.exit(0 if failed == 0 else 1)