DNchngguo/test_upload_api.py

import requests
import json
import os

BASE_URL = "http://localhost:8448"
UPLOAD_API_URL = f"{BASE_URL}/watershed/model/upload"
LIST_API_URL = f"{BASE_URL}/watershed/model/list/test"

def test_upload_model():
    print("开始测试上传模型接口...")
    try:
        # 准备上传文件（使用一个小的测试文件）
        test_file_path = "test_upload.txt"
        with open(test_file_path, "w") as f:
            f.write("这是一个测试文件")
            
        # 准备请求数据
        files = {
            'file': open(test_file_path, 'rb')
        }
        
        data = {
            'modelName': '测试模型',
            'modelType': 'RESERVOIR',
            'modelFormat': 'TXT',
            'uploadUnit': '测试部门',
            'status': 'NORMAL',
            'coordinates': '120.1234,30.5678'
        }
        
        print(f"请求 URL: {UPLOAD_API_URL}")
        print(f"请求数据: {json.dumps(data, ensure_ascii=False)}")
        
        response = requests.post(UPLOAD_API_URL, data=data, files=files)
        print(f"响应状态码: {response.status_code}")
        print(f"响应内容:\n{response.text}")
        
        if response.status_code == 200:
            result = response.json()
            print("\n✅ 上传成功")
            print(f"响应数据: {json.dumps(result, indent=2, ensure_ascii=False)}")
            
            # 检查是否返回了模型数据
            if 'data' in result:
                model = result['data']
                print("\n返回的模型数据:")
                for key, value in model.items():
                    print(f"  {key}: {value}")
                
                # 检查创建时间是否设置
                if 'created_at' in model and model['created_at']:
                    print("\n✅ 创建时间字段已正确设置")
                else:
                    print("\n❌ 创建时间字段为空")
        else:
            print(f"\n❌ 请求失败: {response.text}")
            
    except Exception as e:
        print(f"\n❌ 请求发生异常: {e}")
    finally:
        # 清理测试文件
        if os.path.exists(test_file_path):
            os.remove(test_file_path)

def test_get_models():
    print("\n--- 检查模型列表 ---")
    try:
        params = {
            'page': 1,
            'size': 20
        }
        
        response = requests.get(LIST_API_URL, params=params)
        if response.status_code == 200:
            data = response.json()
            print(f"模型总数: {data['total']}")
            
            if 'rows' in data and data['rows']:
                print("\n最近上传的模型:")
                for model in data['rows'][-3:]:  # 显示最后3条记录
                    print(f"\n模型ID: {model['id']}")
                    print(f"模型名称: {model['name']}")
                    print(f"创建时间: {model['created_at']}")
                    print(f"更新时间: {model['updated_at']}")
            else:
                print("暂无模型数据")
        else:
            print(f"获取模型列表失败: {response.text}")
    except Exception as e:
        print(f"获取模型列表异常: {e}")

if __name__ == "__main__":
    test_upload_model()
    test_get_models()