Autoparts-DB/pos/tests/test_meilisearch.py

#!/usr/bin/env python3
"""Test Meilisearch integration (Mejora #2).

Validates:
1. Meilisearch health
2. Search returns results faster than PostgreSQL tsvector
3. Fallback to PostgreSQL when Meilisearch is unreachable
4. Results are enriched with local stock
"""

import os
import sys
import time

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.meili_search import health_check, search_parts
from services.catalog_service import smart_search, _search_meili_fallback
from tenant_db import get_master_conn, 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 main():
    print("=" * 60)
    print("MEILISEARCH — VALIDATION SUITE")
    print("=" * 60)

    passed = 0
    failed = 0

    # ── Test 1: Health check ────────────────────────────────────
    print("\n[1] Meilisearch Health")
    if health_check():
        print_result("Health", True, "available")
        passed += 1
    else:
        print_result("Health", False, "unreachable")
        failed += 1
        print(f"\n{RED}Meilisearch down — aborting{RESET}")
        return passed, failed

    # ── Test 2: Direct Meilisearch search ───────────────────────
    print("\n[2] Direct Meilisearch Search")
    try:
        t0 = time.perf_counter()
        result = search_parts("filtro de aceite", limit=10)
        t_meili = (time.perf_counter() - t0) * 1000

        if result and result.get('hits'):
            hits = result['hits']
            print_result("Search", True, f"{len(hits)} hits in {t_meili:.1f} ms")
            passed += 1
        else:
            print_result("Search", False, "no hits")
            failed += 1
    except Exception as e:
        print_result("Search", False, str(e))
        failed += 1

    # ── Test 3: smart_search uses Meilisearch ───────────────────
    print("\n[3] smart_search() with Meilisearch")
    try:
        master = get_master_conn()
        tenant = get_tenant_conn_by_dbname('tenant_acct_test')

        # Meilisearch path
        t0 = time.perf_counter()
        meili_results = smart_search(master, "filtro aceite", tenant, branch_id=None, limit=10)
        t_smart = (time.perf_counter() - t0) * 1000

        # Pure PostgreSQL path (force fallback by using a query that might not match)
        t0 = time.perf_counter()
        pg_results = smart_search(master, "zzzzzzzz", tenant, branch_id=None, limit=10)
        t_pg = (time.perf_counter() - t0) * 1000

        master.close()
        tenant.close()

        if meili_results and len(meili_results) > 0:
            print_result("Meili path", True, f"{len(meili_results)} results in {t_smart:.1f} ms")
            passed += 1
        else:
            print_result("Meili path", False, "no results")
            failed += 1

        print(f"  PostgreSQL fallback (no-match query): {t_pg:.1f} ms")
    except Exception as e:
        print_result("smart_search", False, str(e))
        failed += 1

    # ── Test 4: _search_meili_fallback returns None on error ────
    print("\n[4] Fallback behavior")
    try:
        master = get_master_conn()
        # Pass a garbage URL to force failure
        old_url = os.environ.get('MEILI_URL')
        os.environ['MEILI_URL'] = 'http://localhost:99999'
        from services.meili_search import reset_client
        reset_client()

        fallback = _search_meili_fallback(master, "aceite", 10)

        # Restore
        if old_url:
            os.environ['MEILI_URL'] = old_url
        else:
            os.environ.pop('MEILI_URL', None)
        reset_client()

        master.close()

        if fallback is None:
            print_result("Fallback", True, "returns None on unreachable Meilisearch")
            passed += 1
        else:
            print_result("Fallback", False, f"unexpected return: {fallback}")
            failed += 1
    except Exception as e:
        print_result("Fallback", False, str(e))
        failed += 1

    # ── 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)