File size: 3,891 Bytes
307aee3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """
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())
|