feat(v3-data-store): harden search security and standardize test suite

app/routers/data_store.py

@@ -242,34 +242,75 @@ async def search_v3_data_store_obj_w_code(
         ):
     """
     Advanced Search for Data Store (within a code hierarchy).
+    Enforces Account Context isolation and uses ID Vision (v_data_store).
     """
     if delay.sleep_time_s > 0: await asyncio.sleep(delay.sleep_time_s)
 
-    # 1. Resolve IDs
+    # 1. Resolve parent IDs if provided
     resolved_for_id = None
     if for_type and for_id:
         resolved_for_id = redis_lookup_id_random(record_id_random=for_id, table_name=for_type)
         if not resolved_for_id:
             return mk_resp(data=False, status_code=404, status_message="Parent for_id not found.")
 
-    # 2. Apply advanced search on top of the standard lookup
+    # 2. Construct the hierarchical search SQL
-    # We use sql_select with the SearchQuery
+    # We must enforce that users only see their own account records OR global defaults (account_id IS NULL)
-    sql_result = sql_select(
+    from app.db_sql import sql_enable_part, sql_hidden_part, sql_search_qry_part, sql_limit_offset_part
-        table_name='data_store',
+    
-        enabled=status_filter.enabled,
+    sql_enabled, data_enabled = sql_enable_part('data_store', status_filter.enabled)
-        hidden=status_filter.hidden,
+    sql_hidden, data_hidden = sql_hidden_part('data_store', status_filter.hidden)
+    
+    # Generate search logic from the SearchQuery model
+    search_sql, search_data = sql_search_qry_part(
         search_query=search_query, 
-        field_name='code',
+        table_name='v_data_store'
-        field_value=data_store_code,
-        limit=pagination.limit,
-        offset=pagination.offset,
-        as_list=True,
     )
     
+    sql_limit = sql_limit_offset_part(limit=pagination.limit, offset=pagination.offset)
+    
+    # Prepare parameter dictionary
+    data = {
+        'code': data_store_code,
+        'account_id': account.account_id,
+        'for_type': for_type,
+        'for_id': resolved_for_id
+    }
+    data.update(search_data)
+    if data_enabled is not None: data['enable'] = data_enabled
+    if data_hidden is not None: data['hide'] = data_hidden
+
+    # Hierarchical Fallback Logic: (Object Override > Account Override > Global System Default)
+    # This matches the GET lookup logic but allows multiple results via search.
+    sql = f"""
+        SELECT *
+        FROM `v_data_store` AS `data_store`
+        WHERE
+            (
+                `data_store`.account_id = :account_id
+                OR `data_store`.account_id IS NULL
+                OR (`data_store`.for_type = :for_type AND `data_store`.for_id = :for_id)
+            )
+            AND `data_store`.code = :code
+            {sql_enabled}
+            {sql_hidden}
+            {search_sql}
+        ORDER BY `data_store`.for_id DESC, `data_store`.account_id DESC, `data_store`.created_on DESC
+        {sql_limit};
+    """
+
+    sql_result = sql_select(sql=sql, data=data, as_list=True)
+
     if sql_result is False:
         return mk_resp(data=False, status_code=500, status_message="Database error during search.")
 
-    return mk_resp(data=sql_result)
+    # 3. Parse results through Pydantic model to enforce ID Vision (random IDs only)
+    try:
+        data_objs = [Data_Store_Base(**r) for r in sql_result]
+    except Exception as e:
+        log.error(f"Validation error during Data Store search: {e}")
+        return mk_resp(data=False, status_code=500, status_message="Data integrity error during result mapping.")
+
+    return mk_resp(data=data_objs)
 
 
 async def handle_get_data_store_obj_w_code(

tests/README.md

@@ -44,6 +44,16 @@ These consolidated scripts are the primary verification tool for the V3 API.
 2.  **Archiving**: When a new specialized test is created, check if it can be combined into one of the "Primary" suites above. If so, combine and move the original to `archive/`.
 3.  **Cleanup**: Always use `tests/e2e/cleanup_test_files.py` after running file lifecycle tests.
 
+## 🛠️ Standards & Patterns
+
+To maintain a "nice" and readable test suite, follow these patterns in all new Python E2E scripts:
+
+- **Helper Functions**: Use a `print_result(label, success, message="")` helper to output standardized `✅ PASS` and `❌ FAIL` status lines.
+- **Functional Isolation**: Wrap test scenarios in descriptive functions (e.g., `test_hierarchical_fallback()`) rather than writing monolithic scripts.
+- **Vision Compliance**: Always verify that IDs in the response are strings (Random IDs) and not integers, to ensure "ID Vision" is working.
+- **Performance Logging**: Include a suite timer at the end of the `if __name__ == "__main__":` block to track execution speed.
+- **Clean Environment**: Do not leave temporary debug or local-only scripts in the `e2e/` directory.
+
 ---
 
 ## 🚀 How to Run

tests/e2e/test_e2e_v3_data_store_lookup.py

@@ -1,6 +1,5 @@
 import requests
 import json
-import argparse
 import time
 
 # --- Configuration ---
@@ -9,61 +8,61 @@ AGENT_API_KEY = "PMM4n50teUCaOMMTN8qOJA"
 
 # Default Contexts for Testing
 CONTEXTS = {
-    "account_1": "_XY7DXtc9MY",
+    "account_1": "_XY7DXtc9MY", # Standard Test Account
     "account_5": "xFP7AhU8Zlc",
-    "event_1358": "nmBfuGFeR0k"
+    "event_1": "nmBfuGFeR0k"
 }
 
-def run_lookup(code, description, account_id=None, for_type=None, for_id=None, limit=1, delay_ms=0):
+def get_headers(account_id=None):
-    """
-    Performs a Data Store lookup and prints standardized results.
-    """
-    print(f"[V3] {description} (Limit: {limit}, Delay: {delay_ms}ms)")
-    
     headers = {
         "X-Aether-API-Key": AGENT_API_KEY,
         "Content-Type": "application/json"
     }
-    
     if account_id:
         headers["x-account-id"] = account_id
     else:
         headers["x-no-account-id"] = "bypass"
+    return headers
 
+def print_result(label, success, message=""):
+    """Standardized output helper."""
+    status = "✅ PASS" if success else "❌ FAIL"
+    print(f"[{status}] {label} {message}")
+
+def test_standard_lookup():
+    """Verifies the cascading lookup (Object > Account > Global)."""
+    print("\n--- Testing Standard Cascading Lookup ---")
+    code = "event_launcher_main_info"
     url = f"{BASE_URL}/v3/data_store/code/{code}"
-    params = {"for_type": for_type, "for_id": for_id, "limit": limit, "delay_ms": delay_ms}
     
+    # Test Account Lookup
+    resp = requests.get(url, headers=get_headers(CONTEXTS["account_1"]))
+    print_result("Lookup: Account Context", resp.status_code == 200)
+
+    # Test Object Override Lookup
+    params = {"for_type": "event", "for_id": CONTEXTS["event_1"]}
+    resp = requests.get(url, headers=get_headers(CONTEXTS["account_1"]), params=params)
+    print_result("Lookup: Object Context Override", resp.status_code == 200)
+
+def test_delay_simulation():
+    """Verifies X-Delay-ms header and delay_ms query param."""
+    print("\n--- Testing Latency Simulation ---")
+    code = "event_launcher_main_info"
+    url = f"{BASE_URL}/v3/data_store/code/{code}"
+    
+    delay_ms = 500
     start_time = time.time()
-    response = requests.get(url, headers=headers, params=params)
+    resp = requests.get(url, headers=get_headers(CONTEXTS["account_1"]), params={"delay_ms": delay_ms})
     duration = (time.time() - start_time) * 1000
     
-    print(f"  Status: {response.status_code} ({duration:.0f}ms)")
+    # Allow for some network overhead but check if it's at least the delay
-    
+    print_result(f"Delay: query param ({delay_ms}ms)", resp.status_code == 200 and duration >= delay_ms)
-    if response.status_code == 200:
-        data = response.json().get('data')
-        if data:
-            if isinstance(data, list):
-                print(f"  Result: SUCCESS (List of {len(data)})")
-            else:
-                print(f"  Result: SUCCESS (Single Object)")
-        else:
-            print(f"  Result: NULL (No record found)")
-    else:
-        print(f"  Result: ERROR - {response.text[:200]}")
-    print("-" * 60)
-
-def run_search(code, description):
-    """
-    Tests the new POST /search endpoint for Data Store codes.
-    """
-    print(f"[V3-SEARCH] {description}")
 
+def test_advanced_search():
+    """Verifies POST /search with hierarchical logic and ID Vision."""
+    print("\n--- Testing Advanced POST Search ---")
+    code = "event_launcher_main_info"
     url = f"{BASE_URL}/v3/data_store/code/{code}/search"
-    headers = {
-        "X-Aether-API-Key": AGENT_API_KEY,
-        "x-no-account-id": "bypass",
-        "Content-Type": "application/json"
-    }
     
     search_query = {
         "and_filters": [
@@ -71,27 +70,31 @@ def run_search(code, description):
         ]
     }
     
-    response = requests.post(url, headers=headers, json=search_query)
+    resp = requests.post(url, headers=get_headers(), json=search_query)
-    print(f"  Status: {response.status_code}")
+    success = resp.status_code == 200
     
-    if response.status_code == 200:
+    vision_ok = True
-        data = response.json().get('data', [])
+    if success:
-        print(f"  Result: SUCCESS (Found {len(data)} results via POST Search)")
+        data = resp.json().get('data', [])
-    else:
+        if data:
-        print(f"  Result: ERROR - {response.text[:200]}")
+            # Check ID Vision (strings, not ints)
-    print("-" * 60)
+            item = data[0]
+            if not isinstance(item.get('id'), str) or not isinstance(item.get('account_id'), (str, type(None))):
+                vision_ok = False
+                print(f"  ❌ ID Vision Failure: id={item.get('id')}, account_id={item.get('account_id')}")
+
+    print_result("Search: POST /search with ID Vision", success and vision_ok)
 
 if __name__ == "__main__":
-    CODE = "event_launcher_main_info"
+    print(f"=== Aether Data Store V3 E2E Suite ===")
-    print(f"=== Aether Data Store V3 Parity Tester ===\n")
+    print(f"Target: {BASE_URL}")
     
-    # 1. Standard Lookup
+    start_time = time.time()
-    run_lookup(CODE, "Scenario: Standard Lookup", account_id=CONTEXTS["account_1"])
+    try:
+        test_standard_lookup()
+        test_delay_simulation()
+        test_advanced_search()
+    except Exception as e:
+        print(f"💥 Suite Error: {e}")
     
-    # 2. Delay Simulation
+    print(f"\nSuite completed in {time.time() - start_time:.2f}s")
-    run_lookup(CODE, "Scenario: Delay Simulation (500ms)", account_id=CONTEXTS["account_1"], delay_ms=500)
-
-    # 3. Advanced Search (POST)
-    run_search(CODE, "Scenario: Advanced Search via POST")
-
-    print("\nTests Complete.")