consultoria-as
ea29cc31c0
- Cleaned 137+ fake engine-displacement models from supplier imports (v3/v4 scripts: Chevrolet, Ford, Chrysler, Dodge, Jeep, Nissan, etc.) - Removed 1,251+ corrupted models (INT. prefixes, year-suffix, torque specs, empty names, trailing-year variants) - Migrated supplier tables to master DB (supplier_catalog, supplier_catalog_compat, supplier_catalog_interchange) - Fixed _get_mye_ids_with_parts() to query supplier_catalog_compat from master DB so supplier-only vehicles appear for all tenants - Added fuzzy model matcher with parenthesis stripping, noise suffix removal, compact matching, prefix/substring fallback, model aliases, and ±3 year proximity - Matched compat rows: KEEP GREEN +14,152, KNADIAN +3,021, VAZLO +127,500, LUK +477, RAYBESTOS +1,743 - Added KNADIAN catalog importer with year-range expansion and future-year filtering - Added VAZLO catalog importer with position parsing and SKU-in-model cleanup - Added Keep Green, LUK, Yokomitsu, Raybestos catalog importers - Cache clearing after cleanups (_classify_cache_*, nexus:mye_ids:*, nexus:brand_mye_counts:*) Final match rates: - KEEP GREEN: 90.3% - VAZLO: 93.6% - YOKOMITSU: 100.0% - KNADIAN: 57.4% - LUK: 51.0% - RAYBESTOS: 55.9%
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Test bulk import endpoint — CSV parsing, column normalisation, upsert logic."""
|
|
|
|
import os
|
|
import sys
|
|
import io
|
|
import csv
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
os.environ.setdefault('MASTER_DB_URL', 'postgresql://postgres@/nexus_autopartes')
|
|
os.environ.setdefault('TENANT_DB_URL_TEMPLATE', 'postgresql://postgres@/{db_name}')
|
|
os.environ.setdefault('POS_JWT_SECRET', 'test-secret-12345678901234567890123456789012')
|
|
|
|
from blueprints.inventory_bp import _to_decimal, _to_int
|
|
|
|
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 test_to_decimal():
|
|
assert _to_decimal("10.5") == 10.5
|
|
assert _to_decimal("1,000.50") == 1000.50
|
|
assert _to_decimal("") == 0
|
|
assert _to_decimal(None, 5) == 5
|
|
assert _to_decimal("abc", 99) == 99
|
|
print_result("_to_decimal parsing", True)
|
|
|
|
|
|
def test_to_int():
|
|
assert _to_int("42") == 42
|
|
assert _to_int("1,000") == 1000
|
|
assert _to_int("") == 0
|
|
assert _to_int(None, 7) == 7
|
|
assert _to_int("abc", 99) == 99
|
|
print_result("_to_int parsing", True)
|
|
|
|
|
|
def test_csv_column_normalisation():
|
|
"""Simulate the column normalisation done in bulk_import_items."""
|
|
raw_rows = [
|
|
{"SKU": "ABC123", "Nombre": "Filtro de aceite", "Marca": "Bosch", "Precio": "150", "Cantidad": "10"},
|
|
{"sku": "DEF456", "name": "Pastillas de freno", "brand": "TRW", "price": "450.50", "stock": "5"},
|
|
]
|
|
|
|
# Normalise keys
|
|
for r in raw_rows:
|
|
normalised = {}
|
|
for k, v in r.items():
|
|
nk = str(k).strip().lower().replace(' ', '_')
|
|
normalised[nk] = v
|
|
r.clear()
|
|
r.update(normalised)
|
|
|
|
col_map = {
|
|
'sku': 'part_number', 'numero_de_parte': 'part_number', 'parte': 'part_number',
|
|
'nombre': 'name', 'producto': 'name', 'descripcion': 'name',
|
|
'marca': 'brand', 'precio': 'price', 'costo': 'cost',
|
|
'cantidad': 'stock', 'existencia': 'stock', 'inventario': 'stock',
|
|
'ubicacion': 'location', 'categoria': 'category',
|
|
'fabricante': 'make', 'vehiculo': 'make', 'auto': 'make',
|
|
'modelo': 'model', 'anio': 'year', 'ano': 'year',
|
|
'motor': 'engine', 'codigo_motor': 'engine_code',
|
|
}
|
|
for r in raw_rows:
|
|
for old_k, new_k in col_map.items():
|
|
if old_k in r and new_k not in r:
|
|
r[new_k] = r.pop(old_k)
|
|
|
|
assert raw_rows[0]['part_number'] == 'ABC123'
|
|
assert raw_rows[0]['name'] == 'Filtro de aceite'
|
|
assert raw_rows[0]['stock'] == '10'
|
|
assert raw_rows[1]['part_number'] == 'DEF456'
|
|
assert raw_rows[1]['price'] == '450.50'
|
|
print_result("CSV column normalisation", True)
|
|
|
|
|
|
def test_csv_dict_reader():
|
|
"""Verify csv.DictReader produces the expected structure."""
|
|
csv_text = "sku,name,brand,price,stock\nABC123,Filtro,Bosch,150,10\nDEF456,Pastillas,TRW,450,5"
|
|
f = io.StringIO(csv_text)
|
|
reader = csv.DictReader(f)
|
|
rows = list(reader)
|
|
assert len(rows) == 2
|
|
assert rows[0]['sku'] == 'ABC123'
|
|
assert rows[1]['price'] == '450'
|
|
print_result("csv.DictReader parsing", True)
|
|
|
|
|
|
def run_all():
|
|
print("\nBulk Import Tests")
|
|
print("=" * 40)
|
|
test_to_decimal()
|
|
test_to_int()
|
|
test_csv_column_normalisation()
|
|
test_csv_dict_reader()
|
|
print("=" * 40)
|
|
print("Done.\n")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run_all()
|