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

import requests
import json
import time

# --- Configuration ---
BASE_URL = "https://dev-api.oneskyit.com/v3/crud"
API_KEY = "PMM4n50teUCaOMMTN8qOJA" # Agent API Key

# Stable parent IDs used for nested create regression tests.
# journal account: nqOzejLCDXM  |  event account: GpLf_bnywCs
JOURNAL_PARENT_ID = "OGQK-02-04-94"
EVENT_PARENT_ID = "vfzVJF0LH1O"
# event_person: ffkKxiHpOEC (16603) "Scott Idem" under Demo event
EVENT_PERSON_PARENT_ID = "ffkKxiHpOEC"
# event_badge_template: jgfixEpYp1B (18) "Dev Demo 202x"
EVENT_BADGE_TEMPLATE_ID = "jgfixEpYp1B"

# Test Targets: (Object Type, Valid ID Random)
# Note: These IDs are extracted from real active records.
TARGETS = [
    ("event_badge", "JPUG-87-80-88"),
    ("event_badge_template", "gDcA4kVb5B0"),
    ("event_exhibit", "xK_9yEj1bQY"),
    ("event_exhibit_tracking", "KVypw_xntSY"),
    ("event_file", "a2pPIT_W28o") # Regression Target for Relational ID bug
]

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_create_lifecycle(parent_type, parent_id, child_type, payload):
    """
    Regression test for nested POST create (parent FK injection).

    Bug: root_validators on child models stripped integer parent FKs before
    INSERT, causing MariaDB 1364 errors. Fixed in api_crud_v3_nested.py by
    re-injecting resolved_parent_id into data_to_insert after serialization.

    Verifies:
      1. POST /{parent_type}/{parent_id}/{child_type}/ returns 200
      2. Response data has a string 'id' (Vision Standard)
      3. Cleanup: DELETE the created record
    """
    label = f"Nested Create ({parent_type}/{child_type})"
    print(f"\n--- Regression: {label} ---")
    url = f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/"
    headers = get_headers()

    # --- CREATE ---
    resp = requests.post(url, headers=headers, json=payload)
    if resp.status_code != 200:
        print(f"  ❌ [FAIL] POST returned {resp.status_code}: {resp.text[:300]}")
        return False

    data = resp.json().get('data', {})
    new_id = data.get('id') or data.get('obj_id_random')

    if not new_id or not isinstance(new_id, str):
        print(f"  ❌ [FAIL] No string 'id' in response. Got: {data}")
        return False

    print(f"  ✅ [PASS] Created {child_type} with id: {new_id}")

    # --- VISION COMPLIANCE: parent FK must not appear as integer ---
    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):
                print(f"  ❌ [FAIL] Vision violation: {key} is {type(val).__name__} ({val})")
                return False

    print(f"  ✅ [PASS] Vision Standard: all ID fields are strings.")

    # --- CLEANUP ---
    delete_url = f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/{new_id}"
    del_resp = requests.delete(delete_url, headers=headers)
    if del_resp.status_code == 200:
        print(f"  ✅ [PASS] Cleanup: deleted {new_id}")
    else:
        print(f"  ⚠️  [WARN] Cleanup failed ({del_resp.status_code}) — manual cleanup may be needed for {new_id}")

    return True


def test_nested_create_secondary_fk(parent_type, parent_id, child_type, payload, required_fk_fields):
    """
    Regression test for secondary FK resolution in nested POST create.

    Bug: sanitize_payload ran BEFORE model instantiation in the nested POST handler.
    For FKs other than the parent FK (e.g. event_badge_template_id on event_badge),
    sanitize_payload resolved the string → integer, then the model's root_validator
    stripped the integer back to None (Vision ID anti-leakage guard).  The parent FK
    survived only because it was explicitly re-injected; secondary FKs were silently lost.

    Fix (api_crud_v3_nested.py): moved sanitize_payload to run on data_to_insert AFTER
    model serialization, matching the flat V3 POST pattern.

    Verifies:
      1. POST returns 200.
      2. Each field in required_fk_fields is present AND non-None in the response.
      3. All *_id fields are strings (Vision Standard).
      4. Cleanup: DELETE the created record.
    """
    label = f"Nested Secondary FK ({parent_type}/{child_type})"
    print(f"\n--- Regression: {label} ---")
    url = f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/"
    headers = get_headers()

    resp = requests.post(url, headers=headers, json=payload)
    if resp.status_code != 200:
        print(f"  ❌ [FAIL] POST returned {resp.status_code}: {resp.text[:300]}")
        return False

    data = resp.json().get('data', {})
    new_id = data.get('id') or data.get('obj_id_random')
    if not new_id or not isinstance(new_id, str):
        print(f"  ❌ [FAIL] No string 'id' in response. Got: {data}")
        return False
    print(f"  ✅ [PASS] Created {child_type} with id: {new_id}")

    # Check required secondary FK fields are present and non-None
    for field in required_fk_fields:
        val = data.get(field)
        if val is None:
            print(f"  ❌ [FAIL] Secondary FK '{field}' is None — was not saved to DB.")
            # Still attempt cleanup
            requests.delete(f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/{new_id}", headers=headers)
            return False
        if not isinstance(val, str):
            print(f"  ❌ [FAIL] Secondary FK '{field}' is {type(val).__name__} ({val}) — must be string (Vision Standard).")
            requests.delete(f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/{new_id}", headers=headers)
            return False
        print(f"  ✅ [PASS] Secondary FK '{field}' = {val}")

    # Vision compliance: 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):
                print(f"  ❌ [FAIL] Vision violation: {key} is {type(val).__name__} ({val})")
                requests.delete(f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/{new_id}", headers=headers)
                return False
    print(f"  ✅ [PASS] Vision Standard: all ID fields are strings.")

    # Cleanup
    del_resp = requests.delete(f"{BASE_URL}/{parent_type}/{parent_id}/{child_type}/{new_id}", headers=headers)
    if del_resp.status_code == 200:
        print(f"  ✅ [PASS] Cleanup: deleted {new_id}")
    else:
        print(f"  ⚠️  [WARN] Cleanup failed ({del_resp.status_code}) — manual cleanup may be needed for {new_id}")

    return True


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__":
    suite_start = time.time()
    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())

    # --- Nested Create Regression Tests ---
    # These guard against the Jan 2026 bug where child model root_validators
    # stripped the parent FK integer before INSERT, causing MariaDB 1364 errors.
    results.append(test_nested_create_lifecycle(
        parent_type='journal',
        parent_id=JOURNAL_PARENT_ID,
        child_type='journal_entry',
        payload={'name': '[e2e-test] nested create regression', 'enable': False},
    ))
    results.append(test_nested_create_lifecycle(
        parent_type='event',
        parent_id=EVENT_PARENT_ID,
        child_type='event_session',
        payload={'name': '[e2e-test] nested create regression', 'enable': False},
    ))
    # Secondary FK regression: event_badge_template_id must survive nested POST
    # (was silently dropped as NULL before the sanitize_payload order fix)
    results.append(test_nested_create_secondary_fk(
        parent_type='event_person',
        parent_id=EVENT_PERSON_PARENT_ID,
        child_type='event_badge',
        payload={
            'event_badge_template_id': EVENT_BADGE_TEMPLATE_ID,
            'given_name': '[e2e-test]',
            'family_name': 'secondary-fk-regression',
            'enable': False,
            'hide': True,
        },
        required_fk_fields=['event_badge_template_id'],
    ))

    elapsed = time.time() - suite_start
    if all(results):
        print(f"\n🏆 DEMO SUITE SUCCESS: All critical endpoints are verified stable. ({elapsed:.2f}s)")
    else:
        print(f"\n🚨 DEMO SUITE FAILURE: Some critical checks failed. ({elapsed:.2f}s)")