ccf2f30e11
ROW_NUMBER() was partitioning by `group`, collapsing all 12 US/* timezones (which share group="United States") down to a single record. Partitioning by `name` correctly deduplicates by timezone identity while still preserving the object > account > global override hierarchy. Priority-only list now returns the expected 72 entries. Adds a regression test asserting all 12 US/* timezones are present in the full list. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
import requests
|
|
import time
|
|
import sys
|
|
import os
|
|
|
|
# Ensure local imports work if needed
|
|
sys.path.append(os.getcwd())
|
|
|
|
# Configuration
|
|
BASE_URL = "https://dev-api.oneskyit.com/v3/lookup"
|
|
HEADERS = {
|
|
"x-aether-api-key": "PMM4n50teUCaOMMTN8qOJA", # Standard Agent Key
|
|
"x-account-id": "_XY7DXtc9MY" # One Sky IT Demo
|
|
}
|
|
|
|
# TODO: SET THIS to your demo site's random ID
|
|
SITE_ID_RANDOM = "92vkYC4fVEl"
|
|
|
|
def print_result(label, success, message=""):
|
|
status = "✅ PASS" if success else "❌ FAIL"
|
|
print(f"{status} | {label} {': ' + message if message else ''}")
|
|
|
|
def test_lookup_list(lu_type, site_id=None, only_priority=False):
|
|
label = f"GET /{lu_type}/list"
|
|
url = f"{BASE_URL}/{lu_type}/list"
|
|
params = {}
|
|
if site_id:
|
|
params["site_id"] = site_id
|
|
label += f" (Site: {site_id})"
|
|
if only_priority:
|
|
params["only_priority"] = "true"
|
|
label += " (Priority Only)"
|
|
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.get(url, headers=HEADERS, params=params)
|
|
duration = time.time() - start_time
|
|
|
|
if response.status_code == 200:
|
|
data = response.json().get('data', [])
|
|
msg = f"Found {len(data)} items ({duration:.2f}s)"
|
|
print_result(label, True, msg)
|
|
|
|
# Print top 10 for sorting verification
|
|
if data and not site_id: # Only print for full or priority lists
|
|
limit = 10 if not only_priority else len(data)
|
|
print(f" Items:")
|
|
for i, item in enumerate(data[:limit]):
|
|
prio = item.get('priority', 0)
|
|
sort = item.get('sort', 0)
|
|
print(f" [{i+1}] {item.get('name')} (Prio: {prio}, Sort: {sort})")
|
|
|
|
return data
|
|
else:
|
|
print_result(label, False, f"Status {response.status_code}: {response.text[:100]}")
|
|
return None
|
|
except Exception as e:
|
|
print_result(label, False, str(e))
|
|
return None
|
|
|
|
def test_lookup_resolve(lu_type, query):
|
|
url = f"{BASE_URL}/{lu_type}/resolve"
|
|
params = {"q": query}
|
|
try:
|
|
response = requests.get(url, headers=HEADERS, params=params)
|
|
if response.status_code == 200:
|
|
data = response.json().get('data', {})
|
|
name = data.get('name')
|
|
print_result(f"GET /{lu_type}/resolve?q={query}", True, f"Resolved to '{name}'")
|
|
return True
|
|
else:
|
|
print_result(f"GET /{lu_type}/resolve?q={query}", False, f"Status {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
print_result(f"GET /{lu_type}/resolve?q={query}", False, str(e))
|
|
return False
|
|
|
|
US_TIMEZONES = [
|
|
"US/Alaska", "US/Arizona", "US/Central", "US/East-Indiana",
|
|
"US/Eastern", "US/Hawaii", "US/Indiana-Starke", "US/Michigan",
|
|
"US/Mountain", "US/Pacific", "US/Pacific-New", "US/Samoa",
|
|
]
|
|
|
|
def test_timezone_us_dedup():
|
|
"""Regression: PARTITION BY `name` fix — all 12 US/* zones must appear."""
|
|
label = "time_zone US/* deduplication (regression)"
|
|
data = test_lookup_list("time_zone")
|
|
if data is None:
|
|
return
|
|
names = {item.get("name") for item in data}
|
|
missing = [tz for tz in US_TIMEZONES if tz not in names]
|
|
if missing:
|
|
print_result(label, False, f"Missing: {missing}")
|
|
else:
|
|
print_result(label, True, f"All 12 US/* timezones present ({len(data)} total)")
|
|
|
|
if __name__ == "__main__":
|
|
print(f"🚀 Starting V3 Lookup E2E Suite ({BASE_URL})\n")
|
|
start_suite = time.time()
|
|
|
|
# 1. Basic Lists (Phase 1)
|
|
test_lookup_list("country")
|
|
|
|
print("\n--- Regression: US/* timezone deduplication ---")
|
|
test_timezone_us_dedup()
|
|
|
|
print("\n--- Testing Priority Only ---")
|
|
test_lookup_list("time_zone", only_priority=True)
|
|
|
|
# 2. Whitelist Test (Phase 2)
|
|
if SITE_ID_RANDOM != "SET_ME_TO_SITE_ID":
|
|
print("\n--- Testing Site Whitelist Policy ---")
|
|
# Should return only whitelisted items
|
|
test_lookup_list("country", site_id=SITE_ID_RANDOM)
|
|
test_lookup_list("time_zone", site_id=SITE_ID_RANDOM)
|
|
else:
|
|
print("\n⚠️ Skipping Phase 2 test: SITE_ID_RANDOM not set.")
|
|
|
|
# 3. Resolve Test
|
|
print("\n--- Testing Resolve ---")
|
|
test_lookup_resolve("country", "US")
|
|
|
|
print(f"\n⏱️ Suite completed in {time.time() - start_suite:.2f}s")
|