OSIT-AE-API-FastAPI/tests/e2e/test_e2e_v3_search_engine.py

import requests
import json
import time

# --- Configuration ---
API_BASE = "https://dev-api.oneskyit.com/v3/crud"
API_KEY = "PMM4n50teUCaOMMTN8qOJA"
ACCOUNT_ID = "nqOzejLCDXM" # Standard Test Account

def get_headers(no_account=False):
    headers = {
        "Content-Type": "application/json",
        "X-Aether-API-Key": API_KEY
    }
    if no_account:
        headers["x-no-account-id"] = "bypass"
    else:
        headers["x-account-id"] = ACCOUNT_ID
    return headers

def print_result(label, success, message=""):
    status = "✅ PASS" if success else "❌ FAIL"
    print(f"[{status}] {label} {message}")

def test_basic_operators():
    """Tests contains, startswith, endswith logic."""
    print("\n--- Testing Basic Search Operators ---")
    query = {"and": [{"field": "name", "op": "contains", "value": "Journal"}]}
    resp = requests.post(f"{API_BASE}/journal/search", headers=get_headers(), json=query)
    print_result("Operator: contains", resp.status_code == 200)

    query = {"and": [{"field": "name", "op": "startswith", "value": "A"}]}
    resp = requests.post(f"{API_BASE}/journal/search", headers=get_headers(), json=query)
    print_result("Operator: startswith", resp.status_code == 200)

def test_registry_fields():
    """Tests searching by newly added registry fields (created_on, id_random)."""
    print("\n--- Testing Registry-Expanded Fields ---")
    query = {"and_filters": [{"field": "created_on", "op": "gt", "value": "2020-01-01"}]}
    resp = requests.post(f"{API_BASE}/journal/search", headers=get_headers(), json=query)
    print_result("Field: created_on", resp.status_code == 200)

    # Get a valid ID for exact match test
    res = requests.get(f"{API_BASE}/journal/", headers=get_headers(), params={"limit": 1})
    if res.status_code == 200 and res.json().get("data"):
        valid_id = res.json()["data"][0]["id"]
        query = {"and_filters": [{"field": "id_random", "op": "eq", "value": valid_id}]}
        resp = requests.post(f"{API_BASE}/journal/search", headers=get_headers(), json=query)
        print_result(f"Field: id_random ({valid_id})", resp.status_code == 200)

def test_nested_search():
    """Tests POST /search on child objects."""
    print("\n--- Testing Nested Advanced Search ---")
    parent_id = "--ghJX-ztEM" # Valid person
    url = f"{API_BASE}/person/{parent_id}/journal/search"
    query = {"and_filters": [{"field": "name", "op": "like", "value": "%"}]}
    resp = requests.post(url, headers=get_headers(), json=query)
    print_result("Nested Search (person -> journal)", resp.status_code == 200)

def test_extra_filters():
    """Tests enabled=all and hidden=all bypass filters."""
    print("\n--- Testing Extra Filters (enabled/hidden) ---")
    # Using User object as it often has disabled records
    resp = requests.get(f"{API_BASE}/user/?enabled=all&hidden=all", headers=get_headers())
    print_result("Bypass Filters (enabled=all)", resp.status_code == 200)

def test_event_session_qry_str_fields():
    """
    Regression test for event_presentation_li_qry_str and event_presenter_li_qry_str.
    These fields were lost during the v1/v2 -> v3 migration and restored May 2026.
    They live in v_event_session_w_file_count (triggered by ?inc_file_count=true).

    Demo session: DOW3h7v6H42 "How To Do Things" under Demo event pjrcghqwert
    """
    print("\n--- Testing event_session qry_str fields (regression: May 2026) ---")

    EVENT_ID = "pjrcghqwert"
    SESSION_ID = "DOW3h7v6H42"

    headers = {
        "Content-Type": "application/json",
        "X-Aether-API-Key": API_KEY,
        "x-no-account-id": "bypass"
    }

    # 1. Verify fields are returned in the GET response when inc_file_count=true
    url = f"{API_BASE}/event_session/{SESSION_ID}?inc_file_count=true"
    resp = requests.get(url, headers=headers)
    ok = resp.status_code == 200
    print_result("GET event_session with inc_file_count", ok, f"(status={resp.status_code})")
    if ok:
        data = resp.json().get("data", {})
        has_pres = "event_presentation_li_qry_str" in data
        has_presenter = "event_presenter_li_qry_str" in data
        print_result("Field present: event_presentation_li_qry_str", has_pres,
                     f"(value={data.get('event_presentation_li_qry_str')!r})")
        print_result("Field present: event_presenter_li_qry_str", has_presenter,
                     f"(value={data.get('event_presenter_li_qry_str')!r})")

    # 2. Verify searching by event_presentation_li_qry_str via ?view=alt (v_event_session_w_file_count)
    # These fields only exist in the alt view, so ?view=alt is required.
    search_url = f"{API_BASE}/event/{EVENT_ID}/event_session/search?view=alt"
    query = {"and": [{"field": "event_presentation_li_qry_str", "op": "like", "value": "%"}]}
    resp = requests.post(search_url, headers=headers, json=query)
    print_result("Search by event_presentation_li_qry_str (?view=alt)", resp.status_code == 200,
                 f"(status={resp.status_code})")

    # 3. Verify searching by event_presenter_li_qry_str via ?view=alt
    query = {"and": [{"field": "event_presenter_li_qry_str", "op": "like", "value": "%"}]}
    resp = requests.post(search_url, headers=headers, json=query)
    print_result("Search by event_presenter_li_qry_str (?view=alt)", resp.status_code == 200,
                 f"(status={resp.status_code})")

    # 4. Confirm search on default view still rejects these fields (expected 400 — not in v_event_session)
    search_url_default = f"{API_BASE}/event/{EVENT_ID}/event_session/search"
    query = {"and": [{"field": "event_presentation_li_qry_str", "op": "like", "value": "%"}]}
    resp = requests.post(search_url_default, headers=headers, json=query)
    print_result("Search on default view correctly rejects qry_str field (expect 400)", resp.status_code == 400,
                 f"(status={resp.status_code})")

if __name__ == "__main__":
    print(f"Starting Consolidated Search Engine E2E Suite")
    print(f"Target: {API_BASE}")

    start_time = time.time()
    try:
        test_basic_operators()
        test_registry_fields()
        test_nested_search()
        test_extra_filters()
        test_event_session_qry_str_fields()
    except Exception as e:
        print(f"💥 Suite Error: {e}")

    print(f"\nSuite completed in {time.time() - start_time:.2f}s")