Opción C: Vinculación híbrida de inventario local con vehículos
- Nueva tabla inventory_vehicle_compat (v3.1) - Motor inventory_vehicle_compat.py: auto-match + gestión manual - catalog_service.get_parts_local() ahora incluye piezas locales vinculadas - inventory_bp: auto-match en create/update + endpoints REST /vehicles - Frontend catalog.js: badge 'Stock Local' para piezas nativas del tenant - Frontend inventory.js: panel de vehículos compatibles con auto-match - Tests: test_compatibility.py (9/9 pasan) Piezas locales aparecen en navegación por vehículo aunque no estén en TecDoc. Auto-match busca part_number en parts/aftermarket_parts y copia MYEs compatibles.
This commit is contained in:
141
pos/tests/test_compatibility.py
Normal file
141
pos/tests/test_compatibility.py
Normal file
@@ -0,0 +1,141 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Test vehicle compatibility engine (Opción C):
|
||||
- auto_match_vehicle_compatibility
|
||||
- get_compatibility / add / remove
|
||||
- get_inventory_by_vehicle (local items in catalog)
|
||||
"""
|
||||
|
||||
import os, 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 tenant_db import get_master_conn, get_tenant_conn_by_dbname
|
||||
from services.inventory_vehicle_compat import (
|
||||
auto_match_vehicle_compatibility,
|
||||
get_compatibility,
|
||||
add_compatibility,
|
||||
remove_compatibility,
|
||||
get_inventory_by_vehicle,
|
||||
)
|
||||
|
||||
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
|
||||
|
||||
|
||||
# ─── Setup ─────────────────────────────────────
|
||||
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]
|
||||
tenant = get_tenant_conn_by_dbname(db_name)
|
||||
master = get_master_conn()
|
||||
|
||||
# Pick a part_number known to have vehicle_parts
|
||||
TEST_PN = 'A700X6714GA'
|
||||
|
||||
print("=" * 60)
|
||||
print("VEHICLE COMPATIBILITY ENGINE — VALIDATION")
|
||||
print("=" * 60)
|
||||
|
||||
# ─── 1. Create test inventory item ─────────────
|
||||
cur = tenant.cursor()
|
||||
cur.execute("""
|
||||
INSERT INTO inventory (part_number, name, brand, barcode, unit, cost, price_1, price_2, price_3, is_active)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, true)
|
||||
RETURNING id
|
||||
""", (TEST_PN, 'Test Compat Item', 'TestBrand', 'COMPAT-TEST-001', 'PZA', 100, 150, 140, 130))
|
||||
item_id = cur.fetchone()[0]
|
||||
tenant.commit(); cur.close()
|
||||
print(f"\n[Test item created: id={item_id}]")
|
||||
|
||||
# ─── 2. Auto-match ─────────────────────────────
|
||||
print("\n[AUTO-MATCH]")
|
||||
try:
|
||||
result = auto_match_vehicle_compatibility(master, tenant, item_id, TEST_PN)
|
||||
matched = result.get('matched', 0)
|
||||
ok("Auto-match returned count", matched > 0, f"matched={matched}")
|
||||
|
||||
compat = get_compatibility(tenant, master, item_id)
|
||||
ok("Compatibilities created", len(compat) > 0, f"count={len(compat)}")
|
||||
ok("Compat has vehicle details", all(c.get('brand') for c in compat), f"sample={compat[0] if compat else None}")
|
||||
ok("Source is auto_match", all(c.get('source') == 'auto_match' for c in compat))
|
||||
except Exception as e:
|
||||
ok("Auto-match", False, str(e))
|
||||
|
||||
# ─── 3. Manual add/remove ──────────────────────
|
||||
print("\n[MANUAL COMPATIBILITY]")
|
||||
try:
|
||||
# Find an MYE not already linked
|
||||
cur = master.cursor()
|
||||
cur.execute("""
|
||||
SELECT mye.id_mye FROM model_year_engine mye
|
||||
JOIN models m ON m.id_model = mye.model_id
|
||||
LIMIT 1
|
||||
""")
|
||||
mye_id = cur.fetchone()[0]
|
||||
cur.close()
|
||||
|
||||
cid = add_compatibility(tenant, item_id, mye_id, source='manual')
|
||||
ok("Manual add", cid is not None, f"id={cid}")
|
||||
|
||||
compat_after_add = get_compatibility(tenant, master, item_id)
|
||||
manual_items = [c for c in compat_after_add if c.get('source') == 'manual']
|
||||
ok("Manual item in list", len(manual_items) >= 1)
|
||||
|
||||
removed = remove_compatibility(tenant, item_id, mye_id)
|
||||
ok("Manual remove", removed > 0, f"deleted={removed}")
|
||||
|
||||
compat_after_remove = get_compatibility(tenant, master, item_id)
|
||||
manual_items_after = [c for c in compat_after_remove if c.get('source') == 'manual']
|
||||
ok("Manual item removed", len(manual_items_after) == 0)
|
||||
except Exception as e:
|
||||
ok("Manual add/remove", False, str(e))
|
||||
|
||||
# ─── 4. Local inventory in catalog ─────────────
|
||||
print("\n[CATALOG INTEGRATION]")
|
||||
try:
|
||||
# Use the first MYE from auto-match
|
||||
compat = get_compatibility(tenant, master, item_id)
|
||||
if compat:
|
||||
test_mye = compat[0]['model_year_engine_id']
|
||||
local_rows = get_inventory_by_vehicle(tenant, master, test_mye)
|
||||
ok("Local items found for vehicle", any(r[0] == item_id for r in local_rows), f"count={len(local_rows)}")
|
||||
else:
|
||||
ok("Local items found for vehicle", False, "no compatibilities")
|
||||
except Exception as e:
|
||||
ok("Catalog integration", False, str(e))
|
||||
|
||||
# ─── Cleanup ───────────────────────────────────
|
||||
cur = tenant.cursor()
|
||||
cur.execute("DELETE FROM inventory_vehicle_compat WHERE inventory_id = %s", (item_id,))
|
||||
cur.execute("DELETE FROM inventory WHERE id = %s", (item_id,))
|
||||
tenant.commit(); cur.close()
|
||||
tenant.close(); master.close()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print(f"RESULTS: {PASS} {passed} passed, {FAIL} {failed} failed")
|
||||
print("=" * 60)
|
||||
|
||||
sys.exit(0 if failed == 0 else 1)
|
||||
Reference in New Issue
Block a user