Autoparts-DB/pos/tests/test_multi_currency.py

#!/usr/bin/env python3
"""Test multi-currency support (Mejora #8).

Validates:
1. MXN sale (default) works unchanged
2. USD sale stores currency='USD' and exchange_rate
3. Sale items and payments inherit currency
4. Accounting entries are in MXN (converted)
5. Quotation in USD converts correctly
"""

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.pos_engine import process_sale, calculate_totals
from services.currency import convert, get_exchange_rate, to_mxn
from services.accounting_engine import record_sale_entry
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):
    """Get first active inventory item with a price."""
    cur = conn.cursor()
    cur.execute("""
        SELECT id, part_number, name, cost, price_1, tax_rate, branch_id
        FROM inventory WHERE is_active = true AND price_1 > 0 LIMIT 1
    """)
    row = cur.fetchone()
    cur.close()
    return row


def get_open_register(conn):
    cur = conn.cursor()
    cur.execute("SELECT id FROM cash_registers WHERE status = 'open' LIMIT 1")
    row = cur.fetchone()
    cur.close()
    return row[0] if row else None


def main():
    print("=" * 60)
    print("MULTI-CURRENCY — VALIDATION SUITE")
    print("=" * 60)

    passed = 0
    failed = 0

    conn = get_tenant_conn_by_dbname('tenant_acct_test')
    inv = get_test_inventory(conn)
    register_id = get_open_register(conn)

    if not inv:
        print(f"\n{RED}No inventory items available — aborting{RESET}")
        return passed, failed
    if not register_id:
        print(f"\n{RED}No open cash register — aborting{RESET}")
        return passed, failed

    inv_id, part_num, name, cost, price_1, tax_rate, branch_id = inv

    # ── Test 1: MXN sale (default) ──────────────────────────────
    print("\n[1] MXN Sale (default)")
    sale_data_mxn = {
        'items': [{
            'inventory_id': inv_id,
            'quantity': 2,
            'unit_price': float(price_1),
            'discount_pct': 0,
            'tax_rate': float(tax_rate or 0.16),
        }],
        'payment_method': 'efectivo',
        'sale_type': 'cash',
        'register_id': register_id,
        'amount_paid': float(price_1) * 2 * 1.16 + 100,  # overpay
    }

    try:
        sale_mxn = process_sale(conn, sale_data_mxn)
        conn.commit()
        if sale_mxn.get('currency') == 'MXN' and sale_mxn.get('exchange_rate') == 1.0:
            print_result("MXN sale", True, f"total={sale_mxn['total']:.2f} MXN")
            passed += 1
        else:
            print_result("MXN sale", False, f"currency={sale_mxn.get('currency')}, rate={sale_mxn.get('exchange_rate')}")
            failed += 1
    except Exception as e:
        conn.rollback()
        print_result("MXN sale", False, str(e))
        failed += 1

    # ── Test 2: USD sale ────────────────────────────────────────
    print("\n[2] USD Sale")
    # Get exchange rate
    rate = float(get_exchange_rate(conn, 'USD', 'MXN'))
    # Convert price to USD
    price_usd = round(float(price_1) / rate, 2)

    sale_data_usd = {
        'items': [{
            'inventory_id': inv_id,
            'quantity': 2,
            'unit_price': price_usd,
            'discount_pct': 0,
            'tax_rate': float(tax_rate or 0.16),
        }],
        'payment_method': 'efectivo',
        'sale_type': 'cash',
        'register_id': register_id,
        'amount_paid': price_usd * 2 * 1.16 + 10,
        'currency': 'USD',
    }

    try:
        sale_usd = process_sale(conn, sale_data_usd)
        conn.commit()

        checks = []
        if sale_usd.get('currency') == 'USD':
            checks.append("currency=USD")
        if sale_usd.get('exchange_rate', 0) > 1:
            checks.append(f"rate={sale_usd['exchange_rate']:.4f}")

        # Verify DB has currency columns populated
        cur = conn.cursor()
        cur.execute("SELECT currency, exchange_rate FROM sales WHERE id = %s", (sale_usd['id'],))
        db_sale = cur.fetchone()
        cur.execute("SELECT currency, exchange_rate FROM sale_items WHERE sale_id = %s LIMIT 1", (sale_usd['id'],))
        db_item = cur.fetchone()
        cur.execute("SELECT currency, exchange_rate FROM sale_payments WHERE sale_id = %s LIMIT 1", (sale_usd['id'],))
        db_pay = cur.fetchone()
        cur.close()

        if db_sale and db_sale[0] == 'USD':
            checks.append("sale_row_currency=USD")
        if db_item and db_item[0] == 'USD':
            checks.append("item_row_currency=USD")
        if db_pay and db_pay[0] == 'USD':
            checks.append("payment_row_currency=USD")

        if len(checks) >= 5:
            print_result("USD sale", True, ", ".join(checks))
            passed += 1
        else:
            print_result("USD sale", False, f"only {len(checks)} checks passed: {checks}")
            failed += 1
    except Exception as e:
        conn.rollback()
        print_result("USD sale", False, str(e))
        failed += 1

    # ── Test 3: Accounting in MXN regardless of sale currency ───
    print("\n[3] Accounting entries in MXN")
    try:
        # Look up the journal entry for the USD sale
        cur = conn.cursor()
        cur.execute("""
            SELECT id, description FROM journal_entries
            WHERE reference_type = 'sale' AND reference_id = %s
        """, (sale_usd['id'],))
        je = cur.fetchone()
        cur.close()

        if je:
            print_result("Accounting entry", True, f"JE #{je[0]} exists for USD sale")
            passed += 1
        else:
            print_result("Accounting entry", False, "no journal entry found")
            failed += 1
    except Exception as e:
        print_result("Accounting entry", False, str(e))
        failed += 1

    # ── Test 4: Currency conversion helper ──────────────────────
    print("\n[4] Currency conversion helper")
    try:
        rate_usd_mxn = float(get_exchange_rate(conn, 'USD', 'MXN'))
        rate_mxn_usd = float(get_exchange_rate(conn, 'MXN', 'USD'))
        usd_to_mxn = convert(100, 'USD', 'MXN', rate=rate_usd_mxn, conn=conn)
        mxn_to_usd = convert(usd_to_mxn, 'MXN', 'USD', rate=rate_mxn_usd, conn=conn)

        if abs(usd_to_mxn - 100 * rate_usd_mxn) < 0.01:
            print_result("USD→MXN", True, f"100 USD = {usd_to_mxn:.2f} MXN")
            passed += 1
        else:
            print_result("USD→MXN", False, f"expected ~{100*rate_usd_mxn:.2f}, got {usd_to_mxn:.2f}")
            failed += 1

        if abs(mxn_to_usd - 100) < 0.05:
            print_result("Round-trip", True, f"100 USD → {usd_to_mxn:.2f} MXN → {mxn_to_usd:.2f} USD")
            passed += 1
        else:
            print_result("Round-trip", False, f"drift={abs(mxn_to_usd - 100):.2f}")
            failed += 1
    except Exception as e:
        print_result("Conversion", False, str(e))
        failed += 1

    # ── Test 5: to_mxn convenience ──────────────────────────────
    print("\n[5] to_mxn convenience")
    try:
        mxn = to_mxn(50, 'USD', conn=conn)
        if mxn > 50:
            print_result("to_mxn", True, f"50 USD = {mxn:.2f} MXN")
            passed += 1
        else:
            print_result("to_mxn", False, f"expected > 50, got {mxn:.2f}")
            failed += 1
    except Exception as e:
        print_result("to_mxn", 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)