Autoparts-DB/pos/tests/test_suppliers.py

#!/usr/bin/env python3
"""Test supplier and purchase order support (Mejora #3).

Validates:
1. Create supplier
2. List suppliers
3. Create PO
4. Send PO
5. Receive PO (updates stock + accounting)
6. Cancel PO
"""

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.supplier_engine import (
    create_supplier, update_supplier, get_supplier, list_suppliers,
    create_po, send_po, receive_po, cancel_po, get_po, list_pos,
)
from services.inventory_engine import get_stock
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
        FROM inventory WHERE is_active = true LIMIT 1
    """)
    row = cur.fetchone()
    cur.close()
    return row


def main():
    print("=" * 60)
    print("SUPPLIERS & PURCHASE ORDERS — VALIDATION SUITE")
    print("=" * 60)

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

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

    inv_id, part_num, inv_name, branch_id = inv

    # ── Test 1: Create supplier ─────────────────────────────────
    print("\n[1] Create Supplier")
    try:
        supplier_id = create_supplier(conn, {
            'name': 'Proveedor de Prueba',
            'contact_name': 'Juan Perez',
            'phone': '555-1234',
            'email': 'juan@test.com',
            'rfc': 'TEST123456ABC',
            'address': 'Calle Falsa 123',
            'payment_terms': '30 dias',
        })
        conn.commit()
        print_result("Create", True, f"id={supplier_id}")
        passed += 1
    except Exception as e:
        conn.rollback()
        print_result("Create", False, str(e))
        failed += 1
        return passed, failed

    # ── Test 2: Get & list suppliers ────────────────────────────
    print("\n[2] Get & List Suppliers")
    try:
        s = get_supplier(conn, supplier_id)
        if s and s['name'] == 'Proveedor de Prueba':
            print_result("Get", True, s['name'])
            passed += 1
        else:
            print_result("Get", False, "mismatch")
            failed += 1

        suppliers = list_suppliers(conn, limit=10)
        if any(sup['id'] == supplier_id for sup in suppliers):
            print_result("List", True, f"{len(suppliers)} suppliers")
            passed += 1
        else:
            print_result("List", False, "new supplier not in list")
            failed += 1
    except Exception as e:
        print_result("Get/List", False, str(e))
        failed += 1

    # ── Test 3: Create PO ───────────────────────────────────────
    print("\n[3] Create Purchase Order")
    try:
        po_result = create_po(conn, {
            'supplier_id': supplier_id,
            'items': [{
                'inventory_id': inv_id,
                'part_number': part_num,
                'name': inv_name,
                'quantity': 10,
                'unit_price': 150.00,
            }],
            'notes': 'Orden de prueba',
            'expected_date': '2026-05-01',
        }, branch_id=branch_id, employee_id=None)
        conn.commit()
        po_id = po_result['po_id']
        print_result("Create PO", True, f"id={po_id}, total={po_result['total']:.2f}")
        passed += 1
    except Exception as e:
        conn.rollback()
        print_result("Create PO", False, str(e))
        failed += 1
        return passed, failed

    # ── Test 4: Send PO ─────────────────────────────────────────
    print("\n[4] Send PO")
    try:
        ok = send_po(conn, po_id)
        conn.commit()
        if ok:
            print_result("Send", True, "status=sent")
            passed += 1
        else:
            print_result("Send", False, "not updated")
            failed += 1
    except Exception as e:
        conn.rollback()
        print_result("Send", False, str(e))
        failed += 1

    # ── Test 5: Receive PO (partial) ────────────────────────────
    print("\n[5] Receive PO (partial)")
    try:
        # Get the actual PO item ID
        cur = conn.cursor()
        cur.execute("SELECT id FROM purchase_order_items WHERE po_id = %s LIMIT 1", (po_id,))
        poi_row = cur.fetchone()
        cur.close()
        poi_id = poi_row[0] if poi_row else None

        stock_before = get_stock(conn, inv_id, branch_id)
        receive_result = receive_po(conn, po_id, [
            {'po_item_id': poi_id, 'quantity': 6},  # receive 6 of 10
        ], supplier_invoice='FAC-001')
        conn.commit()

        stock_after = get_stock(conn, inv_id, branch_id)
        stock_increase = stock_after - stock_before

        checks = []
        if receive_result['status'] == 'partial':
            checks.append("status=partial")
        if receive_result['received_total'] == 6:
            checks.append("received=6")
        if stock_increase == 6:
            checks.append(f"stock+{stock_increase}")

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

    # ── Test 6: Accounting entry exists ─────────────────────────
    print("\n[6] Accounting entry for PO")
    try:
        cur = conn.cursor()
        cur.execute("""
            SELECT id FROM journal_entries
            WHERE reference_type = 'purchase' AND reference_id = %s
        """, (po_id,))
        je = cur.fetchone()
        cur.close()
        if je:
            print_result("Accounting", True, f"JE #{je[0]}")
            passed += 1
        else:
            print_result("Accounting", False, "no entry found")
            failed += 1
    except Exception as e:
        print_result("Accounting", False, str(e))
        failed += 1

    # ── Test 7: Get PO detail ───────────────────────────────────
    print("\n[7] Get PO Detail")
    try:
        po = get_po(conn, po_id)
        if po and po['items'] and len(po['items']) == 1:
            print_result("Detail", True, f"{len(po['items'])} items, status={po['status']}")
            passed += 1
        else:
            print_result("Detail", False, "missing items")
            failed += 1
    except Exception as e:
        print_result("Detail", False, str(e))
        failed += 1

    # ── Test 8: Cancel a new PO ─────────────────────────────────
    print("\n[8] Cancel PO")
    try:
        po_cancel = create_po(conn, {
            'supplier_id': supplier_id,
            'items': [{
                'inventory_id': inv_id,
                'part_number': part_num,
                'name': inv_name,
                'quantity': 5,
                'unit_price': 100.00,
            }],
        }, branch_id=branch_id)
        conn.commit()
        cancel_po(conn, po_cancel['po_id'], "Prueba de cancelacion")
        conn.commit()

        po_check = get_po(conn, po_cancel['po_id'])
        if po_check and po_check['status'] == 'cancelled':
            print_result("Cancel", True, f"PO #{po_cancel['po_id']} cancelled")
            passed += 1
        else:
            print_result("Cancel", False, "status not cancelled")
            failed += 1
    except Exception as e:
        conn.rollback()
        print_result("Cancel", 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)