consultoria-as
efbd763e43
- 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.
142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
#!/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)
|