""" Test with the exact production MongoDB data. """ import asyncio import sys import os # Add parent directory to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.nosql import mongo_db from insightfy_utils.logging import get_logger logger = get_logger(__name__) MERCHANT_ID = "IN-NATUR-CHEANN-7D2B-O9BP1" ROLE_ID = "admin" WIDGET_ID = "wid_revenue_trend_12m_001" async def test_production_data(): """Test with production data.""" print(f"\n{'='*80}") print(f"Testing Production MongoDB Data") print(f"{'='*80}\n") # Test 1: Find the document print("Test 1: Finding the role document") print("-" * 80) role_doc = await mongo_db["access_roles"].find_one({ "merchant_id": MERCHANT_ID, "role_id": ROLE_ID }) if not role_doc: print("❌ Document not found!\n") return print(f"✓ Document found") print(f" _id: {role_doc.get('_id')}") print(f" merchant_id: {role_doc.get('merchant_id')}") print(f" role_id: {role_doc.get('role_id')}") print(f" widget_access type: {type(role_doc.get('widget_access'))}") print(f" widget_access length: {len(role_doc.get('widget_access', []))}\n") # Test 2: Check if widget is in array print("Test 2: Checking if widget is in array") print("-" * 80) widget_access = role_doc.get("widget_access", []) if WIDGET_ID in widget_access: print(f"✓ Widget '{WIDGET_ID}' is in the array\n") else: print(f"❌ Widget '{WIDGET_ID}' is NOT in the array") print(f" Available widgets: {widget_access}\n") return # Test 3: Test the exact query used by the API print("Test 3: Testing the exact API query") print("-" * 80) query = { "merchant_id": MERCHANT_ID, "role_id": ROLE_ID, "widget_access": WIDGET_ID } print(f"Query: {query}\n") result = await mongo_db["access_roles"].find_one(query) if result: print(f"✓ Query returned a document") print(f" Document _id: {result.get('_id')}\n") else: print(f"❌ Query returned None\n") # Debug: Check what's actually in widget_access print("Debug: Checking widget_access array contents") print("-" * 80) for i, widget in enumerate(widget_access): print(f" [{i}] '{widget}' (type: {type(widget)}, len: {len(widget)})") if widget == WIDGET_ID: print(f" ✓ Matches target widget") else: # Check for subtle differences if widget.strip() == WIDGET_ID: print(f" ⚠️ Has whitespace!") elif widget.lower() == WIDGET_ID.lower(): print(f" ⚠️ Case mismatch!") print(f"\nTarget widget: '{WIDGET_ID}' (type: {type(WIDGET_ID)}, len: {len(WIDGET_ID)})\n") return # Test 4: Test with different query formats print("Test 4: Testing alternative query formats") print("-" * 80) # Using $in operator query_in = { "merchant_id": MERCHANT_ID, "role_id": ROLE_ID, "widget_access": {"$in": [WIDGET_ID]} } result_in = await mongo_db["access_roles"].find_one(query_in) print(f"Query with $in: {result_in is not None}") # Using $elemMatch query_elem = { "merchant_id": MERCHANT_ID, "role_id": ROLE_ID, "widget_access": {"$elemMatch": {"$eq": WIDGET_ID}} } result_elem = await mongo_db["access_roles"].find_one(query_elem) print(f"Query with $elemMatch: {result_elem is not None}\n") print(f"{'='*80}") print(f"Test Complete") print(f"{'='*80}\n") if __name__ == "__main__": asyncio.run(test_production_data())