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")