Update V3 CRUD router and object definitions.
- Added 'external_person_id' to Post searchable fields. - Updated api_crud_v3.py to respect 'fields_to_exclude_from_db' model attribute. - Cleaned up old verification scripts (moved to tests/).
This commit is contained in:
@@ -47,7 +47,7 @@ cms_obj_li = {
|
||||
# V3 Search Security:
|
||||
'searchable_fields': [
|
||||
'post_id_random', 'account_id_random', 'organization_id_random',
|
||||
'person_id_random', 'user_id_random', 'title', 'content',
|
||||
'person_id_random', 'user_id_random', 'external_person_id', 'title', 'content',
|
||||
'type_code', 'topic_code', 'category_code', 'tags', 'location',
|
||||
'enable', 'hide', 'priority', 'sort', 'group', 'notes',
|
||||
'archive_on', 'created_on', 'updated_on'
|
||||
|
||||
@@ -432,6 +432,12 @@ async def post_obj(
|
||||
for k in keys_to_remove:
|
||||
del data_to_insert[k]
|
||||
|
||||
# Filter out model-specific excluded fields (e.g., view-only fields like person_full_name in Journal)
|
||||
if hasattr(input_model, 'fields_to_exclude_from_db'):
|
||||
for k in input_model.fields_to_exclude_from_db:
|
||||
if k in data_to_insert:
|
||||
del data_to_insert[k]
|
||||
|
||||
if sql_insert_result := sql_insert(data=data_to_insert, table_name=table_name_insert):
|
||||
new_obj_id = sql_insert_result
|
||||
new_obj_id_random = get_id_random(record_id=new_obj_id, table_name=obj_name)
|
||||
@@ -489,6 +495,12 @@ async def patch_obj(
|
||||
for k in keys_to_remove:
|
||||
del obj_data[k]
|
||||
|
||||
# Filter out model-specific excluded fields (e.g., view-only fields like person_full_name in Journal)
|
||||
if hasattr(input_model, 'fields_to_exclude_from_db'):
|
||||
for k in input_model.fields_to_exclude_from_db:
|
||||
if k in obj_data:
|
||||
del obj_data[k]
|
||||
|
||||
if sql_update(data=obj_data, table_name=table_name_update, record_id=record_id):
|
||||
if return_obj:
|
||||
if sql_select_result := sql_select(table_name=table_name_select, record_id=record_id):
|
||||
|
||||
@@ -1,36 +0,0 @@
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add current directory to path
|
||||
sys.path.append(os.getcwd())
|
||||
|
||||
print("Attempting to import app.lib_general_v3...")
|
||||
try:
|
||||
import app.lib_general_v3
|
||||
print("Success: app.lib_general_v3")
|
||||
except Exception as e:
|
||||
print(f"Failed: app.lib_general_v3 - {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
print("-" * 20)
|
||||
|
||||
print("Attempting to import app.routers.api_crud_v3...")
|
||||
try:
|
||||
import app.routers.api_crud_v3
|
||||
print("Success: app.routers.api_crud_v3")
|
||||
except Exception as e:
|
||||
print(f"Failed: app.routers.api_crud_v3 - {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
print("-" * 20)
|
||||
|
||||
print("Attempting to import app.routers.agent_bridge...")
|
||||
try:
|
||||
import app.routers.agent_bridge
|
||||
print("Success: app.routers.agent_bridge")
|
||||
except Exception as e:
|
||||
print(f"Failed: app.routers.agent_bridge - {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
@@ -1,50 +0,0 @@
|
||||
import requests
|
||||
import json
|
||||
|
||||
# Configuration
|
||||
BASE_URL = "https://dev-api.oneskyit.com"
|
||||
SEARCH_ENDPOINT = f"{BASE_URL}/v3/crud/site_domain/search"
|
||||
RESTRICTED_ENDPOINT = f"{BASE_URL}/v3/crud/journal/search"
|
||||
|
||||
def test_site_domain_exception():
|
||||
print("--- Testing site_domain guest access (Exception) ---")
|
||||
search_query = {
|
||||
"q": "%", # Match all for testing
|
||||
"and": []
|
||||
}
|
||||
|
||||
try:
|
||||
# No Authorization or X-Account-ID headers provided
|
||||
response = requests.post(SEARCH_ENDPOINT, json=search_query)
|
||||
print(f"Status Code: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print("SUCCESS: site_domain search allowed without authentication.")
|
||||
print(f"Result count: {len(data.get('data', []))}")
|
||||
else:
|
||||
print(f"FAILED: site_domain search returned {response.status_code}")
|
||||
print(response.text)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error during site_domain test: {e}")
|
||||
|
||||
def test_restricted_search():
|
||||
print("\n--- Testing restricted search (Should fail) ---")
|
||||
search_query = {"q": "%"}
|
||||
|
||||
try:
|
||||
response = requests.post(RESTRICTED_ENDPOINT, json=search_query)
|
||||
print(f"Status Code: {response.status_code}")
|
||||
|
||||
if response.status_code == 403:
|
||||
print("SUCCESS: Restricted search was correctly blocked (403 Forbidden).")
|
||||
else:
|
||||
print(f"FAILED: Restricted search returned {response.status_code} instead of 403.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error during restricted test: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_site_domain_exception()
|
||||
test_restricted_search()
|
||||
Reference in New Issue
Block a user