#!/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)