Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(v3-vision): strict integer stripping and demo parity verification

Browse Source 
1. Hardened all demo models to set non-string ID fields to None, ensuring full Vision Standard compliance.\n2. Added status_id_random to common field schema.\n3. Verified account_id availability in exhibit tracking.\n4. Added comprehensive E2E parity test suite for demo objects.\n5. Fixed NameError by importing root_validator.
This commit is contained in:
Scott Idem
2026-02-06 16:16:08 -05:00
parent f3662f9462
commit d5e685dee8
5 changed files with 111 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
app/models/common_field_schema.py
View File

@@ -80,6 +80,7 @@ base_fields['post_comment_id_random'] = xxx_id_random_field_schema
base_fields['product_id_random'] = xxx_id_random_field_schema
base_fields['site_id_random'] = xxx_id_random_field_schema
base_fields['site_domain_id_random'] = xxx_id_random_field_schema
base_fields['status_id_random'] = xxx_id_random_field_schema
base_fields['sponsorship_cfg_id_random'] = xxx_id_random_field_schema
base_fields['sponsorship_id_random'] = xxx_id_random_field_schema
base_fields['user_id_random'] = xxx_id_random_field_schema

9
app/models/event_badge_models.py
View File

@@ -1,7 +1,7 @@
import datetime, pytz
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator, root_validator
from app.db_sql import redis_lookup_id_random
from app.lib_general import log, logging
@@ -55,14 +55,11 @@ class Event_Badge_Base(BaseModel):
if ep_rid := values.get('event_person_id_random'): values['event_person_id'] = ep_rid
if p_rid := values.get('person_id_random'): values['person_id'] = p_rid
# 2. Prevent "Collision Population" or leakage of integers during API responses
# WE MUST NOT DELETE these if they are already integers during a POST operation
# as they have been resolved by sanitize_payload.
# 2. Prevent leakage of integers during API responses (Vision Standard)
for k in ['id', 'event_badge_id', 'event_id', 'event_id_only', 'event_badge_template_id', 'event_person_id', 'person_id']:
val = values.get(k)
if val is not None and not isinstance(val, str):
if values.get(f'{k}_random') or (k=='id' and values.get('id_random')) or (k=='event_id_only' and values.get('event_id_random_only')):
del values[k]
values[k] = None
return values

7
app/models/event_exhibit_models.py
View File

@@ -1,7 +1,7 @@
import datetime, pytz
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator, root_validator
from app.db_sql import redis_lookup_id_random
from app.lib_general import log, logging
@@ -53,12 +53,11 @@ class Event_Exhibit_Base(BaseModel):
if p_rid := values.get('person_id_random'): values['person_id'] = p_rid
if s_rid := values.get('status_id_random'): values['status_id'] = s_rid
# 2. Prevent "Collision Population" or leakage of integers during API responses
# 2. Prevent leakage of integers during API responses (Vision Standard)
for k in ['id', 'event_exhibit_id', 'account_id', 'event_id', 'organization_id', 'contact_id', 'person_id', 'status_id']:
val = values.get(k)
if val is not None and not isinstance(val, str):
if values.get(f'{k}_random') or (k=='id' and values.get('id_random')):
del values[k]
values[k] = None
return values

7
app/models/event_exhibit_tracking_models.py
View File

@@ -1,7 +1,7 @@
import datetime, pytz
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator, root_validator
from app.db_sql import redis_lookup_id_random
from app.lib_general import log, logging
@@ -53,12 +53,11 @@ class Event_Exhibit_Tracking_Base(BaseModel):
if ep_rid := values.get('event_person_id_random'): values['event_person_id'] = ep_rid
if eb_rid := values.get('event_badge_id_random'): values['event_badge_id'] = eb_rid
# 2. Prevent "Collision Population" or leakage of integers during API responses
# 2. Prevent leakage of integers during API responses (Vision Standard)
for k in ['id', 'event_exhibit_tracking_id', 'account_id', 'event_id', 'event_exhibit_id', 'event_person_id', 'event_badge_id']:
val = values.get(k)
if val is not None and not isinstance(val, str):
if values.get(f'{k}_random') or (k=='id' and values.get('id_random')):
del values[k]
values[k] = None
return values

101
tests/e2e/test_e2e_v3_demo_parity.py Normal file
View File

@@ -0,0 +1,101 @@
import requests
import json
# --- Configuration ---
BASE_URL = "https://dev-api.oneskyit.com/v3/crud"
API_KEY = "PMM4n50teUCaOMMTN8qOJA" # Agent API Key
# ACCOUNT_ID = "_XY7DXtc9MY"
# Test Targets: (Object Type, Valid ID Random)
# Note: These IDs are extracted from real active records.
TARGETS = [
("event_badge", "JPUG-87-80-88"),
("event_exhibit", "xK_9yEj1bQY"),
("event_exhibit_tracking", "KVypw_xntSY")
]
def get_headers():
return {
"Content-Type": "application/json",
"X-Aether-API-Key": API_KEY,
"x-no-account-id": "bypass"
}
def verify_demo_parity(obj_type, record_id):
"""
Verifies that the object returns ONLY string IDs for all ID fields (Vision Standard).
Specifically checks for account_id in tracking.
"""
print(f"--- Testing {obj_type}: {record_id} ---")
url = f"{BASE_URL}/{obj_type}/{record_id}"
try:
response = requests.get(url, headers=get_headers())
if response.status_code == 200:
data = response.json().get('data', {})
failures = []
# 1. Check Vision Standard (All *_id fields must be strings)
for key, val in data.items():
if key == "id" or (key.endswith("_id") and not key.endswith("external_id")):
if val is not None and not isinstance(val, str):
failures.append(f"{key} is {type(val).__name__} ({val})")
# 2. Specific check for account_id in tracking
if obj_type == "event_exhibit_tracking":
if "account_id" not in data or data["account_id"] is None:
failures.append("account_id is missing or null in tracking view")
elif not isinstance(data["account_id"], str):
failures.append(f"account_id is not a string ({type(data['account_id']).__name__})")
if not failures:
print(f" ✅ [PASS] All ID fields are strings.")
if obj_type == "event_exhibit_tracking":
print(f" ✅ [PASS] account_id found: {data.get('account_id')}")
return True
else:
print(f" ❌ [FAIL] Vision integrity error:")
for f in failures:
print(f" - {f}")
return False
else:
print(f" ❌ [ERROR] Status {response.status_code}: {response.text[:200]}")
return False
except Exception as e:
print(f" 💥 [EXCEPTION] {e}")
return False
def test_nested_alias_resolution():
"""
Verifies that the 'entry' alias and nested resolution works for journals.
(Testing the fix we just implemented to ensure no regressions).
"""
print("\n--- Testing Nested Alias Resolution (/journal/.../entry/) ---")
parent_id = "OGQK-02-04-94"
child_id = "xWX-NX-e6-EN"
url = f"{BASE_URL}/journal/{parent_id}/entry/{child_id}"
resp = requests.get(url, headers=get_headers())
if resp.status_code == 200:
print(f" ✅ [PASS] Nested alias resolution successful.")
return True
else:
print(f" ❌ [FAIL] Nested alias resolution failed (Status {resp.status_code})")
return False
if __name__ == "__main__":
print("🚀 Starting Aether V3 Demo Parity Suite\n")
results = []
for obj_type, record_id in TARGETS:
results.append(verify_demo_parity(obj_type, record_id))
print("-" * 40)
results.append(test_nested_alias_resolution())
if all(results):
print("\n🏆 DEMO SUITE SUCCESS: All critical endpoints are verified stable.")
else:
print("\n🚨 DEMO SUITE FAILURE: Some critical checks failed.")