diff --git a/tests/e2e/test_e2e_v3_action_download.py b/tests/e2e/test_e2e_v3_action_download.py index 447e654..4aed7ce 100644 --- a/tests/e2e/test_e2e_v3_action_download.py +++ b/tests/e2e/test_e2e_v3_action_download.py @@ -3,32 +3,44 @@ import json # Configuration BASE_URL = "https://dev-api.oneskyit.com/v3/action/hosted_file" -API_KEY = "IDF68Em5X4HTZlswRNgepQ" -ACCOUNT_ID = "Q8lR8Ai8hx2FjbQ3C_EH1Q" +API_KEY = "PMM4n50teUCaOMMTN8qOJA" +ACCOUNT_ID = "Q8lR8Ai8hx2FjbQ3C_EH1Q" # OSIT -# This file was created during our earlier upload tests -VALID_FILE_ID = "2R06T6yuQLw" -# Use a known site key from the DB for the bypass test -# SITE_KEY = "..." +# IDs for testing ID Vision resolution +HOSTED_FILE_ID = "PUwhbT2tMgU" +EVENT_FILE_ID = "cFyCd7FPZe9" +ARCHIVE_CONTENT_ID = "UjKzrk-GKu5" -def test_download_standard(): - print(f"--- Testing Standard Download via V3 Action: {VALID_FILE_ID} ---") - - url = f"{BASE_URL}/{VALID_FILE_ID}/download" - headers = { - "X-Aether-API-Key": API_KEY, - "x-account-id": ACCOUNT_ID - } +# Hash for testing content-addressable storage +FILE_HASH = "384fef832fa07b347a1d70927d0606b24cf41771202c1bfa00d7026764db2bb2" + +def test_download_by_id(label, test_id): + print(f"--- Testing Download via {label}: {test_id} ---") + url = f"{BASE_URL}/{test_id}/download" + headers = {"X-Aether-API-Key": API_KEY, "x-account-id": ACCOUNT_ID} try: - # We don't want to download the whole binary in a test, so we'll check headers response = requests.get(url, headers=headers, stream=True) print(f"Status: {response.status_code}") - print(f"Content-Type: {response.headers.get('Content-Type')}") - print(f"Content-Length: {response.headers.get('Content-Length')}") - if response.status_code == 200: - print("āœ… Success: Standard download works.") + print(f"āœ… Success: Download via {label} works.") + return True + else: + print(f"āŒ Failed: {response.text}") + return False + except Exception as e: + print(f"šŸ’„ Exception: {e}") + return False + +def test_download_by_hash_query(): + print(f"\n--- Testing Hash Download via Query Param API Key ---") + url = f"{BASE_URL}/hash/{FILE_HASH}/download?api_key={API_KEY}" + + try: + response = requests.get(url) + print(f"Status: {response.status_code}") + if response.status_code == 200: + print("āœ… Success: Hash download with query param works.") return True else: print(f"āŒ Failed: {response.text}") @@ -38,9 +50,8 @@ def test_download_standard(): return False def test_download_streaming(): - print(f"\n--- Testing Byte-Range Streaming: {VALID_FILE_ID} ---") - - url = f"{BASE_URL}/{VALID_FILE_ID}/download" + print(f"\n--- Testing Byte-Range Streaming ---") + url = f"{BASE_URL}/{HOSTED_FILE_ID}/download" headers = { "X-Aether-API-Key": API_KEY, "x-account-id": ACCOUNT_ID, @@ -50,8 +61,6 @@ def test_download_streaming(): try: response = requests.get(url, headers=headers) print(f"Status: {response.status_code} (Expected 206)") - print(f"Content-Range: {response.headers.get('Content-Range')}") - if response.status_code == 206: print("āœ… Success: Byte-range streaming works.") return True @@ -63,9 +72,15 @@ def test_download_streaming(): return False if __name__ == "__main__": - s1 = test_download_standard() - s2 = test_download_streaming() - if s1 and s2: - print("\nšŸŽ‰ ALL DOWNLOAD ACTION TESTS PASSED!") + results = [ + test_download_by_id("Hosted File", HOSTED_FILE_ID), + test_download_by_id("Event File (ID Vision)", EVENT_FILE_ID), + test_download_by_id("Archive Content (ID Vision)", ARCHIVE_CONTENT_ID), + test_download_by_hash_query(), + test_download_streaming() + ] + + if all(results): + print("\nšŸŽ‰ ALL DOWNLOAD PATTERNS VERIFIED!") else: - print("\nāŒ SOME TESTS FAILED.") + print("\nāŒ SOME TESTS FAILED.") \ No newline at end of file diff --git a/tests/e2e/test_e2e_v3_data_store_lookup.py b/tests/e2e/test_e2e_v3_data_store_lookup.py index 0dc41a3..2e3eb2a 100644 --- a/tests/e2e/test_e2e_v3_data_store_lookup.py +++ b/tests/e2e/test_e2e_v3_data_store_lookup.py @@ -1,25 +1,24 @@ import requests import json import argparse +import time # --- Configuration --- BASE_URL = "https://dev-api.oneskyit.com" -AGENT_API_KEY = "IDF68Em5X4HTZlswRNgepQ" +AGENT_API_KEY = "PMM4n50teUCaOMMTN8qOJA" # Default Contexts for Testing CONTEXTS = { "account_1": "_XY7DXtc9MY", "account_5": "xFP7AhU8Zlc", - "account_22": "3Iid1aIRY5j", - "account_23": "nqOzejLCDXM", "event_1358": "nmBfuGFeR0k" } -def run_lookup(code, description, account_id=None, for_type=None, for_id=None, limit=1, version="v3"): +def run_lookup(code, description, account_id=None, for_type=None, for_id=None, limit=1, delay_ms=0): """ Performs a Data Store lookup and prints standardized results. """ - print(f"[{version.upper()}] {description} (Limit: {limit})") + print(f"[V3] {description} (Limit: {limit}, Delay: {delay_ms}ms)") headers = { "X-Aether-API-Key": AGENT_API_KEY, @@ -27,61 +26,72 @@ def run_lookup(code, description, account_id=None, for_type=None, for_id=None, l } if account_id: - headers["X-Account-ID"] = account_id + headers["x-account-id"] = account_id else: - headers["X-No-Account-ID"] = "bypass" + headers["x-no-account-id"] = "bypass" - if version == "v3": - url = f"{BASE_URL}/v3/data_store/code/{code}" - params = {"for_type": for_type, "for_id": for_id, "limit": limit} - response = requests.get(url, headers=headers, params=params) - else: - # Legacy Endpoint - if for_type and for_id: - url = f"{BASE_URL}/data_store/code/{code}/{for_type}/{for_id}" - else: - url = f"{BASE_URL}/data_store/code/{code}" - response = requests.get(url, headers=headers, params={"limit": limit}) - - print(f" URL: {response.url}") - print(f" Status: {response.status_code}") + url = f"{BASE_URL}/v3/data_store/code/{code}" + params = {"for_type": for_type, "for_id": for_id, "limit": limit, "delay_ms": delay_ms} + + start_time = time.time() + response = requests.get(url, headers=headers, params=params) + duration = (time.time() - start_time) * 1000 + + print(f" Status: {response.status_code} ({duration:.0f}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)})") - for i, item in enumerate(data): - print(f" [{i+1}] ID: {item.get('id') or item.get('data_store_id')}, Name: {item.get('name')[:40]}...") else: print(f" Result: SUCCESS (Single Object)") - print(f" ID: {data.get('id') or data.get('data_store_id')}") - print(f" Name: {data.get('name')}") else: - print(f" Result: NULL (No record found or validation failed)") + 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}") + + 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": [ + {"field": "name", "op": "like", "value": "%"} + ] + } + + response = requests.post(url, headers=headers, json=search_query) + print(f" Status: {response.status_code}") + + if response.status_code == 200: + data = response.json().get('data', []) + print(f" Result: SUCCESS (Found {len(data)} results via POST Search)") else: print(f" Result: ERROR - {response.text[:200]}") print("-" * 60) if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Aether Data Store V3 Lookup Tester") - parser.add_argument("--code", default="event_launcher_main_info", help="The Data Store code to test") - args = parser.parse_args() + CODE = "event_launcher_main_info" + print(f"=== Aether Data Store V3 Parity Tester ===\n") - print(f"=== Aether Data Store Unified Tester ===") - print(f"Target: {BASE_URL}") - print(f"Code: {args.code}\n") + # 1. Standard Lookup + run_lookup(CODE, "Scenario: Standard Lookup", account_id=CONTEXTS["account_1"]) - # 1. Standard Single Result (Default) - run_lookup(args.code, "Scenario: Single Result (Default)", account_id=CONTEXTS["account_1"]) + # 2. Delay Simulation + run_lookup(CODE, "Scenario: Delay Simulation (500ms)", account_id=CONTEXTS["account_1"], delay_ms=500) - # 2. Multi-Result Override (Limit 5) - run_lookup(args.code, "Scenario: Multi-Result Override", account_id=CONTEXTS["account_1"], limit=5) + # 3. Advanced Search (POST) + run_search(CODE, "Scenario: Advanced Search via POST") - # 3. Object Specific Context (Event 1358) - run_lookup(args.code, "Scenario: Event 1358 (under Account 1)", - account_id=CONTEXTS["account_1"], - for_type="event", - for_id=CONTEXTS["event_1358"]) - - print("\nTests Complete.") + print("\nTests Complete.") \ No newline at end of file