Autoparts-DB/pos/tests/test_redis_cache.py

#!/usr/bin/env python3
"""Test Redis stock cache integration.

Validates:
1. Redis connectivity
2. Cache miss → PostgreSQL fallback → cache set
3. Cache hit (sub-millisecond)
4. Invalidation on stock mutation
5. Graceful degradation when Redis is unavailable
"""

import os
import sys
import time
import warnings

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# Must set env vars before importing config
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.redis_stock_cache import (
    get_cached_stock, set_cached_stock, invalidate_stock,
    invalidate_all_stock, health_check, _get_redis
)
from services.inventory_engine import get_stock, record_sale, record_purchase
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 main():
    print("=" * 60)
    print("REDIS STOCK CACHE — VALIDATION SUITE")
    print("=" * 60)

    passed = 0
    failed = 0

    # ── Test 1: Redis connectivity ──────────────────────────────
    print("\n[1] Redis Connectivity")
    if health_check():
        print_result("Redis PING", True, "responding")
        passed += 1
    else:
        print_result("Redis PING", False, "no response")
        failed += 1
        print(f"\n{RED}Redis unavailable — aborting remaining tests{RESET}")
        return passed, failed

    # ── Test 2: Basic cache operations ──────────────────────────
    print("\n[2] Basic Cache Operations")
    set_cached_stock(99999, 42, branch_id=1)
    cached = get_cached_stock(99999, branch_id=1)
    if cached == 42:
        print_result("SET + GET", True, f"value={cached}")
        passed += 1
    else:
        print_result("SET + GET", False, f"expected 42, got {cached}")
        failed += 1

    invalidate_stock(99999, branch_id=1)
    cached_after = get_cached_stock(99999, branch_id=1)
    if cached_after is None:
        print_result("Invalidation", True, "key removed")
        passed += 1
    else:
        print_result("Invalidation", False, f"expected None, got {cached_after}")
        failed += 1

    # ── Test 3: get_stock cache miss / hit ──────────────────────
    print("\n[3] get_stock() with PostgreSQL fallback")
    conn = get_tenant_conn_by_dbname('tenant_acct_test')

    # Ensure we have at least one inventory item with operations
    cur = conn.cursor()
    cur.execute("SELECT id FROM inventory WHERE is_active = true LIMIT 1")
    row = cur.fetchone()
    cur.close()

    if not row:
        print(f"  {YELLOW}SKIP{RESET} No inventory items in tenant_acct_test")
    else:
        inv_id = row[0]

        # Clear any existing cache
        invalidate_stock(inv_id, None)

        # Miss (should query PostgreSQL)
        t0 = time.perf_counter()
        stock_miss = get_stock(conn, inv_id)
        t_miss = (time.perf_counter() - t0) * 1000

        # Hit (should read from Redis)
        t0 = time.perf_counter()
        stock_hit = get_stock(conn, inv_id)
        t_hit = (time.perf_counter() - t0) * 1000

        if stock_miss == stock_hit:
            print_result("Consistency", True, f"PG={stock_miss}, Redis={stock_hit}")
            passed += 1
        else:
            print_result("Consistency", False, f"PG={stock_miss}, Redis={stock_hit}")
            failed += 1

        print(f"  Cache miss: {t_miss:.3f} ms")
        print(f"  Cache hit:  {t_hit:.3f} ms")

        if t_hit < t_miss:
            print_result("Performance", True, f"hit {t_hit:.3f}ms < miss {t_miss:.3f}ms")
            passed += 1
        else:
            print_result("Performance", False, "cache hit not faster")
            failed += 1

    conn.close()

    # ── Test 4: Invalidation on mutation ────────────────────────
    print("\n[4] Invalidation on stock mutation")
    conn = get_tenant_conn_by_dbname('tenant_acct_test')

    cur = conn.cursor()
    cur.execute("SELECT id FROM inventory WHERE is_active = true LIMIT 1")
    row = cur.fetchone()
    cur.close()

    if not row:
        print(f"  {YELLOW}SKIP{RESET} No inventory items available")
    else:
        inv_id = row[0]

        # Pre-populate cache
        invalidate_stock(inv_id, None)
        stock_before = get_stock(conn, inv_id)
        set_cached_stock(inv_id, stock_before)

        # Verify cache hit
        cached_before = get_cached_stock(inv_id)
        if cached_before != stock_before:
            print_result("Pre-populate", False, f"cache mismatch")
            failed += 1
        else:
            # Record a sale (negative operation)
            # Need a valid branch_id
            cur = conn.cursor()
            cur.execute("SELECT id FROM branches LIMIT 1")
            branch_row = cur.fetchone()
            cur.close()
            branch_id = branch_row[0] if branch_row else None

            if branch_id:
                record_sale(conn, inv_id, branch_id, 1)
                conn.commit()

                # Cache should be invalidated
                cached_after = get_cached_stock(inv_id)
                if cached_after is None:
                    print_result("Auto-invalidation", True, "cleared on record_sale")
                    passed += 1
                else:
                    print_result("Auto-invalidation", False, f"still cached: {cached_after}")
                    failed += 1
            else:
                print(f"  {YELLOW}SKIP{RESET} No branches available")

    conn.close()

    # ── Test 5: Bulk stock population ───────────────────────────
    print("\n[5] get_stock_bulk() populates Redis")
    conn = get_tenant_conn_by_dbname('tenant_acct_test')
    from services.inventory_engine import get_stock_bulk

    stock_map = get_stock_bulk(conn)
    if stock_map:
        sample_id = list(stock_map.keys())[0]
        cached = get_cached_stock(sample_id)
        if cached is not None:
            print_result("Bulk populate", True, f"{len(stock_map)} items cached")
            passed += 1
        else:
            print_result("Bulk populate", False, "sample not in cache")
            failed += 1
    else:
        print(f"  {YELLOW}SKIP{RESET} No stock data")
    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)