jingrow-framework/test_api.py

#!/usr/bin/env python3
"""
测试本地版REST API和钩子功能
"""

import requests
import json
import time

# 配置
BASE_URL = "http://localhost:9001"
TEST_PAGETYPE = "Test Page"
TEST_NAME = "TEST-001"

def test_create_record():
    """测试创建记录"""
    print("=== 测试创建记录 ===")
    
    data = {
        "name": TEST_NAME,
        "title": "测试记录",
        "status": "Active"
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/data/{TEST_PAGETYPE}",
            json=data,
            headers={"Content-Type": "application/json"}
        )
        
        print(f"状态码: {response.status_code}")
        print(f"响应: {response.json()}")
        
        if response.status_code == 200:
            print("✅ 创建记录成功")
            return True
        else:
            print("❌ 创建记录失败")
            return False
            
    except Exception as e:
        print(f"❌ 创建记录异常: {e}")
        return False

def test_update_record():
    """测试更新记录"""
    print("\n=== 测试更新记录 ===")
    
    data = {
        "title": "更新后的标题",
        "status": "Inactive"
    }
    
    try:
        response = requests.put(
            f"{BASE_URL}/api/data/{TEST_PAGETYPE}/{TEST_NAME}",
            json=data,
            headers={"Content-Type": "application/json"}
        )
        
        print(f"状态码: {response.status_code}")
        print(f"响应: {response.json()}")
        
        if response.status_code == 200:
            print("✅ 更新记录成功")
            return True
        else:
            print("❌ 更新记录失败")
            return False
            
    except Exception as e:
        print(f"❌ 更新记录异常: {e}")
        return False

def test_get_record():
    """测试获取记录"""
    print("\n=== 测试获取记录 ===")
    
    try:
        response = requests.get(
            f"{BASE_URL}/api/data/{TEST_PAGETYPE}/{TEST_NAME}"
        )
        
        print(f"状态码: {response.status_code}")
        print(f"响应: {response.json()}")
        
        if response.status_code == 200:
            print("✅ 获取记录成功")
            return True
        else:
            print("❌ 获取记录失败")
            return False
            
    except Exception as e:
        print(f"❌ 获取记录异常: {e}")
        return False

def test_get_records():
    """测试获取记录列表"""
    print("\n=== 测试获取记录列表 ===")
    
    try:
        response = requests.get(
            f"{BASE_URL}/api/data/{TEST_PAGETYPE}",
            params={
                "limit_page_length": 10
            }
        )
        
        print(f"状态码: {response.status_code}")
        print(f"响应: {response.json()}")
        
        if response.status_code == 200:
            print("✅ 获取记录列表成功")
            return True
        else:
            print("❌ 获取记录列表失败")
            return False
            
    except Exception as e:
        print(f"❌ 获取记录列表异常: {e}")
        return False

def test_delete_record():
    """测试删除记录"""
    print("\n=== 测试删除记录 ===")
    
    try:
        response = requests.delete(
            f"{BASE_URL}/api/data/{TEST_PAGETYPE}/{TEST_NAME}"
        )
        
        print(f"状态码: {response.status_code}")
        print(f"响应: {response.json()}")
        
        if response.status_code == 200:
            print("✅ 删除记录成功")
            return True
        else:
            print("❌ 删除记录失败")
            return False
            
    except Exception as e:
        print(f"❌ 删除记录异常: {e}")
        return False

def test_hook_execution():
    """测试钩子执行"""
    print("\n=== 测试钩子执行 ===")
    
    data = {
        "pagetype": TEST_PAGETYPE,
        "name": TEST_NAME,
        "hook_name": "on_update",
        "data": {"test": "hook data"}
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/api/hooks/execute",
            json=data,
            headers={"Content-Type": "application/json"}
        )
        
        print(f"状态码: {response.status_code}")
        print(f"响应: {response.json()}")
        
        if response.status_code == 200:
            print("✅ 钩子执行成功")
            return True
        else:
            print("❌ 钩子执行失败")
            return False
            
    except Exception as e:
        print(f"❌ 钩子执行异常: {e}")
        return False

def main():
    """主测试函数"""
    print("开始测试本地版REST API和钩子功能...")
    print(f"测试目标: {BASE_URL}")
    print(f"测试PageType: {TEST_PAGETYPE}")
    print(f"测试记录名: {TEST_NAME}")
    
    # 等待服务启动
    print("\n等待服务启动...")
    time.sleep(2)
    
    # 执行测试
    tests = [
        test_create_record,
        test_update_record,
        test_get_record,
        test_get_records,
        test_hook_execution,
        test_delete_record
    ]
    
    passed = 0
    total = len(tests)
    
    for test in tests:
        try:
            if test():
                passed += 1
        except Exception as e:
            print(f"❌ 测试异常: {e}")
    
    print(f"\n=== 测试结果 ===")
    print(f"通过: {passed}/{total}")
    print(f"成功率: {passed/total*100:.1f}%")
    
    if passed == total:
        print("🎉 所有测试通过！")
    else:
        print("⚠️  部分测试失败，请检查日志")

if __name__ == "__main__":
    main()