yumoqing
0e0ee695e6
- Add orgid field (str32, not nullable) to hermes_services table - Replace user_id with orgid in all service CRUD operations (SQL + functions) - Update function signatures: get_all_services, create_service, delete_service, get_service_by_id, test_service_connection, create_session, send_message_to_service, get_session_messages all use orgid - Add orgid indexes: idx_hermes_services_orgid, idx_hermes_services_orgid_status - Add logined_userorgid filtering to CRUD definition for automatic framework-level isolation - Update all .dspy files to use get_userorgid() for org-scoped service queries - Update init/data.json and db_tables.py to reflect orgid field
192 lines
8.0 KiB
Python
192 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify orgid refactoring of hermes_services table.
|
|
Validates: function signatures, SQL templates, JSON definitions, .dspy files.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import json
|
|
import inspect
|
|
|
|
MODULE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
def test_table_definition():
|
|
"""Verify models/hermes_services.json has orgid field and correct indexes."""
|
|
path = os.path.join(MODULE_DIR, 'models', 'hermes_services.json')
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
|
|
fields = {f['name']: f for f in data['fields']}
|
|
|
|
# orgid field exists with correct properties
|
|
assert 'orgid' in fields, "orgid field missing from table definition"
|
|
assert fields['orgid']['type'] == 'str', "orgid type should be str"
|
|
assert fields['orgid']['length'] == 32, "orgid length should be 32"
|
|
assert fields['orgid']['nullable'] == 'no', "orgid should be not nullable"
|
|
|
|
# user_id should NOT exist
|
|
assert 'user_id' not in fields, "user_id field should be removed"
|
|
|
|
# orgid indexes exist
|
|
index_names = [i['name'] for i in data['indexes']]
|
|
assert 'idx_hermes_services_orgid' in index_names, "missing idx_hermes_services_orgid"
|
|
assert 'idx_hermes_services_orgid_status' in index_names, "missing idx_hermes_services_orgid_status"
|
|
|
|
# old user indexes removed
|
|
assert 'idx_hermes_services_user_id' not in index_names, "old user index should be removed"
|
|
|
|
print(" table definition OK")
|
|
|
|
def test_crud_definition():
|
|
"""Verify json/hermes_services.json has logined_userorgid param."""
|
|
path = os.path.join(MODULE_DIR, 'json', 'hermes_services.json')
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
|
|
params = data['params']
|
|
assert params.get('logined_userorgid') == 'orgid', "logined_userorgid should be 'orgid'"
|
|
assert 'apikey' in params.get('confidential_fields', []), "apikey should be confidential"
|
|
print(" CRUD definition OK")
|
|
|
|
def test_db_tables():
|
|
"""Verify db_tables.py SERVICES_TABLE uses orgid."""
|
|
path = os.path.join(MODULE_DIR, 'hermes_web_cli', 'db_tables.py')
|
|
with open(path) as f:
|
|
content = f.read()
|
|
|
|
assert '"orgid"' in content, "db_tables.py should contain orgid"
|
|
assert 'idx_services_orgid' in content, "should have idx_services_orgid index"
|
|
assert 'idx_services_orgid_status' in content, "should have idx_services_orgid_status index"
|
|
# old user_id references in SERVICES_TABLE should be gone
|
|
svc_start = content.index('SERVICES_TABLE = {')
|
|
svc_end = content.index('# Sessions table', svc_start)
|
|
svc_section = content[svc_start:svc_end]
|
|
assert '"user_id"' not in svc_section, "SERVICES_TABLE should not have user_id"
|
|
print(" db_tables.py OK")
|
|
|
|
def test_crud_ops():
|
|
"""Verify crud_ops.py SERVICES_CRUD SQL uses orgid."""
|
|
path = os.path.join(MODULE_DIR, 'hermes_web_cli', 'crud_ops.py')
|
|
with open(path) as f:
|
|
content = f.read()
|
|
|
|
svc_start = content.index('SERVICES_CRUD = {')
|
|
svc_end = content.index('# Sessions CRUD', svc_start)
|
|
svc_section = content[svc_start:svc_end]
|
|
|
|
assert '${orgid}$' in svc_section, "SQL should use ${orgid}$ parameter"
|
|
assert '${user_id}$' not in svc_section, "SQL should not use ${user_id}$ parameter"
|
|
assert 'orgid = ${orgid}$' in svc_section or 'AND orgid = ${orgid}$' in svc_section, "WHERE should filter by orgid"
|
|
print(" crud_ops.py OK")
|
|
|
|
def test_init_signatures():
|
|
"""Verify init.py function signatures use orgid."""
|
|
path = os.path.join(MODULE_DIR, 'hermes_web_cli', 'init.py')
|
|
with open(path) as f:
|
|
content = f.read()
|
|
|
|
# Check function signatures
|
|
assert 'async def get_all_services(orgid: str)' in content, "get_all_services should take orgid"
|
|
assert 'async def create_service(name: str, url: str, orgid: str' in content, "create_service should take orgid"
|
|
assert 'async def delete_service(service_id: str, orgid: str)' in content, "delete_service should take orgid"
|
|
assert 'async def get_service_by_id(service_id: str, orgid: str)' in content, "get_service_by_id should take orgid"
|
|
assert 'test_service_connection(service_id: str, orgid: str = "")' in content, "test_service_connection should take optional orgid"
|
|
assert 'async def create_session(service_id: str, user_id: str, orgid: str' in content, "create_session should take orgid"
|
|
assert 'send_message_to_service(service_id: str, session_id: str, message: str, user_id: str, orgid: str)' in content, "send_message_to_service should take orgid"
|
|
assert 'async def get_session_messages(session_id: str, user_id: str, orgid: str)' in content, "get_session_messages should take orgid"
|
|
|
|
# Verify SQL calls use orgid param
|
|
assert "'orgid': orgid" in content, "SQL params should pass orgid"
|
|
assert "'orgid': user_id" not in content, "should not pass user_id as orgid"
|
|
print(" init.py signatures OK")
|
|
|
|
def test_dspy_files():
|
|
"""Verify .dspy files use get_userorgid()."""
|
|
files_to_check = [
|
|
'wwwroot/services/list/index.dspy',
|
|
'wwwroot/services/test/index.dspy',
|
|
'wwwroot/services/remove/index.dspy',
|
|
'wwwroot/sessions/create_session.dspy',
|
|
]
|
|
|
|
for fpath in files_to_check:
|
|
full_path = os.path.join(MODULE_DIR, fpath)
|
|
with open(full_path) as f:
|
|
content = f.read()
|
|
assert 'get_userorgid()' in content, f"{fpath} should call get_userorgid()"
|
|
|
|
# services/list should pass orgid to get_all_services
|
|
with open(os.path.join(MODULE_DIR, 'wwwroot/services/list/index.dspy')) as f:
|
|
content = f.read()
|
|
assert 'get_all_services(orgid)' in content, "services/list should pass orgid to get_all_services"
|
|
|
|
# services/remove should pass orgid to delete_service
|
|
with open(os.path.join(MODULE_DIR, 'wwwroot/services/remove/index.dspy')) as f:
|
|
content = f.read()
|
|
assert 'delete_service(service_id, orgid)' in content, "services/remove should pass orgid to delete_service"
|
|
|
|
# create_session should pass both user_id and orgid
|
|
with open(os.path.join(MODULE_DIR, 'wwwroot/sessions/create_session.dspy')) as f:
|
|
content = f.read()
|
|
assert 'create_session(service_id, user_id, orgid' in content, "create_session should pass both user_id and orgid"
|
|
|
|
print(" .dspy files OK")
|
|
|
|
def test_init_data():
|
|
"""Verify init/data.json uses orgid field."""
|
|
path = os.path.join(MODULE_DIR, 'init', 'data.json')
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
|
|
for svc in data.get('hermes_services', []):
|
|
assert 'orgid' in svc, "init data should have orgid field"
|
|
assert 'user_id' not in svc, "init data should not have user_id field"
|
|
print(" init/data.json OK")
|
|
|
|
def test_syntax():
|
|
"""Verify all Python files compile."""
|
|
import py_compile
|
|
for fpath in ['hermes_web_cli/init.py', 'hermes_web_cli/crud_ops.py', 'hermes_web_cli/db_tables.py']:
|
|
full_path = os.path.join(MODULE_DIR, fpath)
|
|
try:
|
|
py_compile.compile(full_path, doraise=True)
|
|
except py_compile.PyCompileError as e:
|
|
assert False, f"Syntax error in {fpath}: {e}"
|
|
print(" Python syntax OK")
|
|
|
|
if __name__ == "__main__":
|
|
tests = [
|
|
("Syntax check", test_syntax),
|
|
("Table definition", test_table_definition),
|
|
("CRUD definition", test_crud_definition),
|
|
("db_tables.py", test_db_tables),
|
|
("crud_ops.py", test_crud_ops),
|
|
("init.py signatures", test_init_signatures),
|
|
(".dspy files", test_dspy_files),
|
|
("init/data.json", test_init_data),
|
|
]
|
|
|
|
passed = 0
|
|
failed = 0
|
|
|
|
for name, test_fn in tests:
|
|
try:
|
|
test_fn()
|
|
passed += 1
|
|
except AssertionError as e:
|
|
print(f" FAILED: {e}")
|
|
failed += 1
|
|
except Exception as e:
|
|
print(f" ERROR: {e}")
|
|
failed += 1
|
|
|
|
print(f"\n{'='*50}")
|
|
print(f"Results: {passed} passed, {failed} failed out of {len(tests)} tests")
|
|
if failed > 0:
|
|
print("FAILED - do not commit")
|
|
sys.exit(1)
|
|
else:
|
|
print("ALL TESTS PASSED")
|
|
sys.exit(0)
|