FuJianModelPlatform/test_upload_model.py

import requests
import json
import os
import tempfile

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:
        # 创建一个临时测试文件
        with tempfile.NamedTemporaryFile(suffix='.txt', delete=False) as f:
            f.write(b"test content")
            temp_file_path = f.name
        
        print(f"临时测试文件创建成功: {temp_file_path}")
        
        # 准备上传参数
        files = {'file': open(temp_file_path, 'rb')}
        data = {
            'modelName': '上传测试模型',
            'modelType': 'HYDROPOWER',
            'modelFormat': 'OBJ',
            'uploadUnit': '测试部门',
            'status': 'NORMAL',
            'coordinates': '116.403874,39.915168'
        }
        
        print(f"请求 URL: {UPLOAD_API_URL}")
        print(f"请求参数: {json.dumps(data, indent=2, ensure_ascii=False)}")
        
        # 发送 POST 请求
        response = requests.post(UPLOAD_API_URL, files=files, data=data)
        print(f"响应状态码: {response.status_code}")
        print(f"响应内容:\n{response.text}")
        
        if response.status_code == 200:
            data = response.json()
            print("\n✅ 上传成功")
            print(f"响应数据结构: {json.dumps(data, indent=2, ensure_ascii=False)}")
            
            if 'data' in data and data['data']:
                model = data['data']
                print("\n上传的模型信息:")
                for key in model.keys():
                    print(f"  {key}: {model[key]}")
                    
                # 检查创建时间和更新时间是否已设置
                if 'created_at' in model and model['created_at']:
                    print("\n✅ 创建时间已正确设置")
                else:
                    print("\n❌ 创建时间未设置")
                    
                if 'updated_at' in model and model['updated_at']:
                    print("✅ 更新时间已正确设置")
                else:
                    print("❌ 更新时间未设置")
                    
            # 查询模型列表，检查是否包含刚刚上传的模型
            print("\n\n查询模型列表:")
            list_response = requests.get(LIST_API_URL, params={'page': 1, 'size': 10})
            print(f"列表查询响应状态码: {list_response.status_code}")
            
            if list_response.status_code == 200:
                list_data = list_response.json()
                print(f"列表查询成功，共找到 {list_data['total']} 个模型")
                
                if list_data['rows']:
                    print("\n第一个模型的字段:")
                    first_model = list_data['rows'][0]
                    for key in first_model.keys():
                        print(f"  {key}: {first_model[key]}")
                        
        else:
            print(f"\n❌ 请求失败: {response.text}")
            
        # 清理临时文件
        os.unlink(temp_file_path)
        
    except Exception as e:
        print(f"\n❌ 请求发生异常: {e}")
        try:
            os.unlink(temp_file_path)
        except:
            pass

if __name__ == "__main__":
    test_upload_model()